trustchain-oc1-orchestral
/
devicehub-teal
Archived
10
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fixing conflict merge with testing

This commit is contained in:
Cayo Puigdefabregas 2020-12-09 11:12:47 +01:00
parent fb899b558f f0c17ebd81
commit eaa20b4302
25 changed files with 1182 additions and 266 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ereuse_devicehub/__init__.py
View File

@ -1 +1 @@
__version__ = "1.0b"
__version__ = "1.0.2-beta"

6
ereuse_devicehub/config.py
View File

@ -14,6 +14,7 @@ from ereuse_devicehub.resources.device import definitions
from ereuse_devicehub.resources.documents import documents
from ereuse_devicehub.resources.enums import PriceSoftware
from ereuse_devicehub.resources.versions import versions
from ereuse_devicehub.resources.metric import definitions as metric_def
class DevicehubConfig(Config):
@ -27,8 +28,9 @@ class DevicehubConfig(Config):
import_resource(proof),
import_resource(documents),
import_resource(inventory),
import_resource(versions)),
)
import_resource(versions),
import_resource(metric_def),
),)
PASSWORD_SCHEMES = {'pbkdf2_sha256'} # type: Set[str]
DB_USER = config('DB_USER', 'dhub')
DB_PASSWORD = config('DB_PASSWORD', 'ereuse')

2
ereuse_devicehub/migrations/versions/68a5c025ab8e_adding_owner_id_in_device.py
View File

@ -12,7 +12,7 @@ from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision = '68a5c025ab8e'
down_revision = 'b9b0ee7d9dca'
down_revision = 'e93aec8fc41f'
branch_labels = None
depends_on = None

73
ereuse_devicehub/migrations/versions/e93aec8fc41f_added_assigned_action.py Normal file
View File

@ -0,0 +1,73 @@
"""Added Assigned action
Revision ID: e93aec8fc41f
Revises: b9b0ee7d9dca
Create Date: 2020-11-17 13:22:56.790956
"""
from alembic import op
import sqlalchemy as sa
from alembic import context
import sqlalchemy_utils
import citext
import teal
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision = 'e93aec8fc41f'
down_revision = 'b9b0ee7d9dca'
branch_labels = None
depends_on = None
def get_inv():
INV = context.get_x_argument(as_dictionary=True).get('inventory')
if not INV:
raise ValueError("Inventory value is not specified")
return INV
def upgrade():
# Allocate action
op.drop_table('allocate', schema=f'{get_inv()}')
op.create_table('allocate',
sa.Column('final_user_code', citext.CIText(), default='', nullable=True,
comment = "This is a internal code for mainteing the secrets of the personal datas of the new holder"),
sa.Column('transaction', citext.CIText(), nullable=True, comment='The code used from the owner for relation with external tool.'),
sa.Column('end_users', sa.Numeric(precision=4), nullable=True),
sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
sa.ForeignKeyConstraint(['id'], [f'{get_inv()}.action.id'], ),
sa.PrimaryKeyConstraint('id'),
schema=f'{get_inv()}'
)
# Deallocate action
op.drop_table('deallocate', schema=f'{get_inv()}')
op.create_table('deallocate',
sa.Column('transaction', citext.CIText(), nullable=True, comment='The code used from the owner for relation with external tool.'),
sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
sa.ForeignKeyConstraint(['id'], [f'{get_inv()}.action.id'], ),
sa.PrimaryKeyConstraint('id'),
schema=f'{get_inv()}'
)
# Add allocate as a column in device
op.add_column('device', sa.Column('allocated', sa.Boolean(), nullable=True), schema=f'{get_inv()}')
# Receive action
op.drop_table('receive', schema=f'{get_inv()}')
# Live action
op.drop_table('live', schema=f'{get_inv()}')
op.create_table('live',
sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('serial_number', sa.Unicode(), nullable=True,
comment='The serial number of the Hard Disk in lower case.'),
sa.Column('usage_time_hdd', sa.Interval(), nullable=True),
sa.Column('snapshot_uuid', postgresql.UUID(as_uuid=True), nullable=False),
sa.ForeignKeyConstraint(['id'], [f'{get_inv()}.action.id'], ),
sa.PrimaryKeyConstraint('id'),
schema=f'{get_inv()}'
)
def downgrade():
op.drop_table('allocate', schema=f'{get_inv()}')

8
ereuse_devicehub/resources/CHANGELOG.md Normal file
View File

@ -0,0 +1,8 @@
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
## [1.0.1-beta] - 2020-11-16
- [fixed] #80 manual merged from website

17
ereuse_devicehub/resources/action/__init__.py
View File

@ -3,7 +3,7 @@ from typing import Callable, Iterable, Tuple
from teal.resource import Converters, Resource
from ereuse_devicehub.resources.action import schemas
from ereuse_devicehub.resources.action.views import ActionView
from ereuse_devicehub.resources.action.views import ActionView, AllocateView, DeallocateView
from ereuse_devicehub.resources.device.sync import Sync
@ -198,6 +198,16 @@ class ToPrepareDef(ActionDef):
SCHEMA = schemas.ToPrepare
class AllocateDef(ActionDef):
VIEW = AllocateView
SCHEMA = schemas.Allocate
class DeallocateDef(ActionDef):
VIEW = DeallocateView
SCHEMA = schemas.Deallocate
class PrepareDef(ActionDef):
VIEW = None
SCHEMA = schemas.Prepare
@ -253,11 +263,6 @@ class DisposeProductDef(ActionDef):
SCHEMA = schemas.DisposeProduct
class ReceiveDef(ActionDef):
VIEW = None
SCHEMA = schemas.Receive
class MigrateToDef(ActionDef):
VIEW = None
SCHEMA = schemas.MigrateTo

163
ereuse_devicehub/resources/action/models.py
View File

@ -10,6 +10,7 @@ to a structure based on:
Within the above general classes are subclasses in A order.
"""
import copy
from collections import Iterable
from contextlib import suppress
from datetime import datetime, timedelta, timezone
@ -43,7 +44,7 @@ from ereuse_devicehub.resources.device.models import Component, Computer, DataSt
Device, Laptop, Server
from ereuse_devicehub.resources.enums import AppearanceRange, BatteryHealth, BiosAccessRange, \
ErasureStandards, FunctionalityRange, PhysicalErasureMethod, PriceSoftware, \
R_NEGATIVE, R_POSITIVE, RatingRange, ReceiverRole, Severity, SnapshotSoftware, \
R_NEGATIVE, R_POSITIVE, RatingRange, Severity, SnapshotSoftware, \
TestDataStorageLength
from ereuse_devicehub.resources.models import STR_SM_SIZE, Thing
from ereuse_devicehub.resources.user.models import User
@ -91,7 +92,7 @@ class Action(Thing):
end_time = Column(db.TIMESTAMP(timezone=True))
end_time.comment = """When the action ends. For some actions like reservations
the time when they expire, for others like renting
the time the end rents. For punctual actions it is the time
the time the end rents. For punctual actions it is the time
they are performed; it differs with ``created`` in which
created is the where the system received the action.
"""
@ -115,7 +116,7 @@ class Action(Thing):
backref=backref('authored_actions', lazy=True, collection_class=set),
primaryjoin=author_id == User.id)
author_id.comment = """The user that recorded this action in the system.
This does not necessarily has to be the person that produced
the action in the real world. For that purpose see
``agent``.
@ -129,9 +130,8 @@ class Action(Thing):
agent = relationship(Agent,
backref=backref('actions_agent', lazy=True, **_sorted_actions),
primaryjoin=agent_id == Agent.id)
agent_id.comment = """The direct performer or driver of the action.
e.g. John wrote a book.
agent_id.comment = """The direct performer or driver of the action. e.g. John wrote a book.
It can differ with the user that registered the action in the
system, which can be in their behalf.
"""
@ -142,14 +142,14 @@ class Action(Thing):
order_by=lambda: Component.id,
collection_class=OrderedSet)
components.comment = """The components that are affected by the action.
When performing actions to parent devices their components are
affected too.
For example: an ``Allocate`` is performed to a Computer and this
relationship is filled with the components the computer had
at the time of the action.
For Add and Remove though, this has another meaning: the components
that are added or removed.
"""
@ -157,9 +157,9 @@ class Action(Thing):
parent = relationship(Computer,
backref=backref('actions_parent', lazy=True, **_sorted_actions),
primaryjoin=parent_id == Computer.id)
parent_id.comment = """For actions that are performed to components,
parent_id.comment = """For actions that are performed to components,
the device parent at that time.
For example: for a ``EraseBasic`` performed on a data storage, this
would point to the computer that contained this data storage, if any.
"""
@ -312,15 +312,21 @@ class Remove(ActionWithOneDevice):
class Allocate(JoinedTableMixin, ActionWithMultipleDevices):
to_id = Column(UUID, ForeignKey(User.id))
to = relationship(User, primaryjoin=User.id == to_id)
organization = Column(CIText())
"""The act of allocate one list of devices to one person
"""
final_user_code = Column(CIText(), default='', nullable=True)
final_user_code.comment = """This is a internal code for mainteing the secrets of the
personal datas of the new holder"""
transaction = Column(CIText(), default='', nullable=True)
transaction.comment = "The code used from the owner for relation with external tool."
end_users = Column(Numeric(precision=4), check_range('end_users', 0), nullable=True)
class Deallocate(JoinedTableMixin, ActionWithMultipleDevices):
from_id = Column(UUID, ForeignKey(User.id))
from_rel = relationship(User, primaryjoin=User.id == from_id)
organization = Column(CIText())
"""The act of deallocate one list of devices to one person of the system or not
"""
transaction= Column(CIText(), default='', nullable=True)
transaction.comment = "The code used from the owner for relation with external tool."
class EraseBasic(JoinedWithOneDeviceMixin, ActionWithOneDevice):
@ -533,7 +539,7 @@ class Snapshot(JoinedWithOneDeviceMixin, ActionWithOneDevice):
version = Column(StrictVersionType(STR_SM_SIZE), nullable=False)
software = Column(DBEnum(SnapshotSoftware), nullable=False)
elapsed = Column(Interval)
elapsed.comment = """For Snapshots made with Workbench, the total amount
elapsed.comment = """For Snapshots made with Workbench, the total amount
of time it took to complete.
"""
@ -680,11 +686,11 @@ class MeasureBattery(TestMixin, Test):
voltage = db.Column(db.Integer, nullable=False)
voltage.comment = """The actual voltage of the battery, in mV."""
cycle_count = db.Column(db.Integer)
cycle_count.comment = """The number of full charges – discharges
cycle_count.comment = """The number of full charges – discharges
cycles.
"""
health = db.Column(db.Enum(BatteryHealth))
health.comment = """The health of the Battery.
health.comment = """The health of the Battery.
Only reported in Android.
"""
@ -883,12 +889,12 @@ class TestBios(TestMixin, Test):
beeps_power_on = Column(Boolean)
beeps_power_on.comment = """Whether there are no beeps or error
codes when booting up.
Reference: R2 provision 6 page 23.
"""
access_range = Column(DBEnum(BiosAccessRange))
access_range.comment = """Difficulty to modify the boot menu.
This is used as an usability measure for accessing and modifying
a bios, specially as something as important as modifying the boot
menu.
@ -1294,25 +1300,77 @@ class Live(JoinedWithOneDeviceMixin, ActionWithOneDevice):
information about its state (in the form of a ``Snapshot`` action)
and usage statistics.
"""
ip = Column(IP, nullable=False,
comment='The IP where the live was triggered.')
subdivision_confidence = Column(SmallInteger,
check_range('subdivision_confidence', 0, 100),
nullable=False)
subdivision = Column(DBEnum(Subdivision), nullable=False)
city = Column(Unicode(STR_SM_SIZE), check_lower('city'), nullable=False)
city_confidence = Column(SmallInteger,
check_range('city_confidence', 0, 100),
nullable=False)
isp = Column(Unicode(STR_SM_SIZE), check_lower('isp'), nullable=False)
organization = Column(Unicode(STR_SM_SIZE), check_lower('organization'))
organization_type = Column(Unicode(STR_SM_SIZE), check_lower('organization_type'))
serial_number = Column(Unicode(), check_lower('serial_number'))
serial_number.comment = """The serial number of the Hard Disk in lower case."""
usage_time_hdd = Column(Interval, nullable=True)
snapshot_uuid = Column(UUID(as_uuid=True))
@property
def country(self) -> Country:
return self.subdivision.country
# todo relate to snapshot
# todo testing
def final_user_code(self):
""" show the final_user_code of the last action Allocate."""
actions = self.device.actions
actions.sort(key=lambda x: x.created)
for e in reversed(actions):
if isinstance(e, Allocate) and e.created < self.created:
return e.final_user_code
return ''
@property
def usage_time_allocate(self):
"""Show how many hours is used one device from the last check"""
self.sort_actions()
if self.usage_time_hdd is None:
return self.last_usage_time_allocate()
delta_zero = timedelta(0)
diff_time = self.diff_time()
if diff_time is None:
return delta_zero
if diff_time < delta_zero:
return delta_zero
return diff_time
def sort_actions(self):
self.actions = copy.copy(self.device.actions)
self.actions.sort(key=lambda x: x.created)
self.actions.reverse()
def last_usage_time_allocate(self):
"""If we don't have self.usage_time_hdd then we need search the last
action Live with usage_time_allocate valid"""
for e in self.actions:
if isinstance(e, Live) and e.created < self.created:
if not e.usage_time_allocate:
continue
return e.usage_time_allocate
return timedelta(0)
def diff_time(self):
for e in self.actions:
if e.created > self.created:
continue
if isinstance(e, Snapshot):
last_time = self.get_last_lifetime(e)
if not last_time:
continue
return self.usage_time_hdd - last_time
if isinstance(e, Live):
if e.snapshot_uuid == self.snapshot_uuid:
continue
if not e.usage_time_hdd:
continue
return self.usage_time_hdd - e.usage_time_hdd
return None
def get_last_lifetime(self, snapshot):
for a in snapshot.actions:
if a.type == 'TestDataStorage' and a.device.serial_number == self.serial_number:
return a.lifetime
return None
class Organize(JoinedTableMixin, ActionWithMultipleDevices):
@ -1348,7 +1406,7 @@ class Trade(JoinedTableMixin, ActionWithMultipleDevices):
extend `Schema's Trade <http://schema.org/TradeAction>`_.
"""
shipping_date = Column(db.TIMESTAMP(timezone=True))
shipping_date.comment = """When are the devices going to be ready
shipping_date.comment = """When are the devices going to be ready
for shipping?
"""
invoice_number = Column(CIText())
@ -1357,7 +1415,7 @@ class Trade(JoinedTableMixin, ActionWithMultipleDevices):
price = relationship(Price,
backref=backref('trade', lazy=True, uselist=False),
primaryjoin=price_id == Price.id)
price_id.comment = """The price set for this trade.
price_id.comment = """The price set for this trade.
If no price is set it is supposed that the trade was
not payed, usual in donations.
"""
@ -1371,8 +1429,7 @@ class Trade(JoinedTableMixin, ActionWithMultipleDevices):
confirms = relationship(Organize,
backref=backref('confirmation', lazy=True, uselist=False),
primaryjoin=confirms_id == Organize.id)
confirms_id.comment = """An organize action that this association confirms.
confirms_id.comment = """An organize action that this association confirms.
For example, a ``Sell`` or ``Rent``
can confirm a ``Reserve`` action.
"""
@ -1434,26 +1491,6 @@ class MakeAvailable(ActionWithMultipleDevices):
pass
class Receive(JoinedTableMixin, ActionWithMultipleDevices):
"""The act of physically taking delivery of a device.
The receiver confirms that the devices have arrived, and thus,
they are the
:attr:`ereuse_devicehub.resources.device.models.Device.physical_possessor`.
This differs from :class:`.Trade` in that trading changes the
political possession. As an example, a transporter can *receive*
a device but it is not it's owner. After the delivery, the
transporter performs another *receive* to the final owner.
The receiver can optionally take a
:class:`ereuse_devicehub.resources.enums.ReceiverRole`.
"""
role = Column(DBEnum(ReceiverRole),
nullable=False,
default=ReceiverRole.Intermediary)
class Migrate(JoinedTableMixin, ActionWithMultipleDevices):
"""Moves the devices to a new database/inventory. Devices cannot be
modified anymore at the previous database.

37
ereuse_devicehub/resources/action/models.pyi
View File

@ -447,26 +447,8 @@ class Prepare(ActionWithMultipleDevices):
class Live(ActionWithOneDevice):
ip = ... # type: Column
subdivision_confidence = ... # type: Column
subdivision = ... # type: Column
city = ... # type: Column
city_confidence = ... # type: Column
isp = ... # type: Column
organization = ... # type: Column
organization_type = ... # type: Column
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self.ip = ... # type: Union[ipaddress.IPv4Address, ipaddress.IPv6Address]
self.subdivision_confidence = ... # type: int
self.subdivision = ... # type: enums.Subdivision
self.city = ... # type: str
self.city_confidence = ... # type: int
self.isp = ... # type: str
self.organization = ... # type: str
self.organization_type = ... # type: str
self.country = ... # type: Country
serial_number = ... # type: Column
time = ... # type: Column
class Organize(ActionWithMultipleDevices):
@ -527,14 +509,15 @@ class DisposeProduct(Trade):
class TransferOwnershipBlockchain(Trade):
pass
class Allocate(ActionWithMultipleDevices):
code = ... # type: Column
end_users = ... # type: Column
class Receive(ActionWithMultipleDevices):
role = ... # type:Column
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self.role = ... # type: ReceiverRole
class Deallocate(ActionWithMultipleDevices):
code = ... # type: Column
class Migrate(ActionWithMultipleDevices):

86
ereuse_devicehub/resources/action/schemas.py
View File

@ -1,3 +1,5 @@
from datetime import datetime, timedelta
from dateutil.tz import tzutc
from flask import current_app as app
from marshmallow import Schema as MarshmallowSchema, ValidationError, fields as f, validates_schema
from marshmallow.fields import Boolean, DateTime, Decimal, Float, Integer, Nested, String, \
@ -14,7 +16,7 @@ from ereuse_devicehub.resources.action import models as m
from ereuse_devicehub.resources.agent import schemas as s_agent
from ereuse_devicehub.resources.device import schemas as s_device
from ereuse_devicehub.resources.enums import AppearanceRange, BiosAccessRange, FunctionalityRange, \
PhysicalErasureMethod, R_POSITIVE, RatingRange, ReceiverRole, \
PhysicalErasureMethod, R_POSITIVE, RatingRange, \
Severity, SnapshotSoftware, TestDataStorageLength
from ereuse_devicehub.resources.models import STR_BIG_SIZE, STR_SIZE
from ereuse_devicehub.resources.schemas import Thing
@ -64,21 +66,62 @@ class Remove(ActionWithOneDevice):
class Allocate(ActionWithMultipleDevices):
__doc__ = m.Allocate.__doc__
to = NestedOn(s_user.User,
description='The user the devices are allocated to.')
organization = SanitizedStr(validate=Length(max=STR_SIZE),
description='The organization where the '
'user was when this happened.')
start_time = DateTime(data_key='startTime', required=True,
description=m.Action.start_time.comment)
end_time = DateTime(data_key='endTime', required=False,
description=m.Action.end_time.comment)
final_user_code = SanitizedStr(data_key="finalUserCode",
validate=Length(min=1, max=STR_BIG_SIZE),
required=False,
description='This is a internal code for mainteing the secrets of the \
personal datas of the new holder')
transaction = SanitizedStr(validate=Length(min=1, max=STR_BIG_SIZE),
required=False,
description='The code used from the owner for \
relation with external tool.')
end_users = Integer(data_key='endUsers', validate=[Range(min=1, error="Value must be greater than 0")])
@validates_schema
def validate_allocate(self, data: dict):
txt = "You need to allocate for a day before today"
delay = timedelta(days=1)
today = datetime.now().replace(tzinfo=tzutc()) + delay
start_time = data['start_time'].replace(tzinfo=tzutc())
if start_time > today:
raise ValidationError(txt)
txt = "You need deallocate before allocate this device again"
for device in data['devices']:
if device.allocated:
raise ValidationError(txt)
device.allocated = True
class Deallocate(ActionWithMultipleDevices):
__doc__ = m.Deallocate.__doc__
from_rel = Nested(s_user.User,
data_key='from',
description='The user where the devices are not allocated to anymore.')
organization = SanitizedStr(validate=Length(max=STR_SIZE),
description='The organization where the '
'user was when this happened.')
start_time = DateTime(data_key='startTime', required=True,
description=m.Action.start_time.comment)
transaction = SanitizedStr(validate=Length(min=1, max=STR_BIG_SIZE),
required=False,
description='The code used from the owner for \
relation with external tool.')
@validates_schema
def validate_deallocate(self, data: dict):
txt = "You need to deallocate for a day before today"
delay = timedelta(days=1)
today = datetime.now().replace(tzinfo=tzutc()) + delay
start_time = data['start_time'].replace(tzinfo=tzutc())
if start_time > today:
raise ValidationError(txt)
txt = "Sorry some of this devices are actually deallocate"
for device in data['devices']:
if not device.allocated:
raise ValidationError(txt)
device.allocated = False
class EraseBasic(ActionWithOneDevice):
@ -369,15 +412,11 @@ class Prepare(ActionWithMultipleDevices):
class Live(ActionWithOneDevice):
__doc__ = m.Live.__doc__
ip = IP(dump_only=True)
subdivision_confidence = Integer(dump_only=True, data_key='subdivisionConfidence')
subdivision = EnumField(Subdivision, dump_only=True)
country = EnumField(Country, dump_only=True)
city = SanitizedStr(lower=True, dump_only=True)
city_confidence = Integer(dump_only=True, data_key='cityConfidence')
isp = SanitizedStr(lower=True, dump_only=True)
organization = SanitizedStr(lower=True, dump_only=True)
organization_type = SanitizedStr(lower=True, dump_only=True, data_key='organizationType')
final_user_code = SanitizedStr(data_key="finalUserCode", dump_only=True)
serial_number = SanitizedStr(data_key="serialNumber", dump_only=True)
usage_time_hdd = TimeDelta(data_key="usageTimeHdd", precision=TimeDelta.HOURS, dump_only=True)
usage_time_allocate = TimeDelta(data_key="usageTimeAllocate",
precision=TimeDelta.HOURS, dump_only=True)
class Organize(ActionWithMultipleDevices):
@ -437,11 +476,6 @@ class TransferOwnershipBlockchain(Trade):
__doc__ = m.TransferOwnershipBlockchain.__doc__
class Receive(ActionWithMultipleDevices):
__doc__ = m.Receive.__doc__
role = EnumField(ReceiverRole)
class Migrate(ActionWithMultipleDevices):
__doc__ = m.Migrate.__doc__
other = URL()

130
ereuse_devicehub/resources/action/views.py
View File

@ -3,18 +3,22 @@
import os
import json
import shutil
from datetime import datetime
from datetime import datetime, timedelta
from distutils.version import StrictVersion
from uuid import UUID
from flask.json import jsonify
from flask import current_app as app, request, g
from flask import current_app as app, request, g, redirect
from sqlalchemy.util import OrderedSet
from teal.marshmallow import ValidationError
from teal.resource import View
from teal.db import ResourceNotFound
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.action.models import Action, RateComputer, Snapshot, VisualTest, \
InitTransfer
from ereuse_devicehub.query import things_response
from ereuse_devicehub.resources.action.models import (Action, RateComputer, Snapshot, VisualTest,
InitTransfer, Live, Allocate, Deallocate)
from ereuse_devicehub.resources.device.models import Device, Computer, DataStorage
from ereuse_devicehub.resources.action.rate.v1_0 import CannotRate
from ereuse_devicehub.resources.enums import SnapshotSoftware, Severity
from ereuse_devicehub.resources.user.exceptions import InsufficientPermission
@ -61,20 +65,52 @@ def move_json(tmp_snapshots, path_name, user):
os.remove(path_name)
class AllocateMix():
model = None
def post(self):
""" Create one res_obj """
res_json = request.get_json()
res_obj = self.model(**res_json)
db.session.add(res_obj)
db.session().final_flush()
ret = self.schema.jsonify(res_obj)
ret.status_code = 201
db.session.commit()
return ret
def find(self, args: dict):
res_objs = self.model.query.filter_by(author=g.user) \
.order_by(self.model.created.desc()) \
.paginate(per_page=200)
return things_response(
self.schema.dump(res_objs.items, many=True, nested=0),
res_objs.page, res_objs.per_page, res_objs.total,
res_objs.prev_num, res_objs.next_num
)
class AllocateView(AllocateMix, View):
model = Allocate
class DeallocateView(AllocateMix, View):
model = Deallocate
class ActionView(View):
def post(self):
"""Posts an action."""
json = request.get_json(validate=False)
tmp_snapshots = app.config['TMP_SNAPSHOTS']
path_snapshot = save_json(json, tmp_snapshots, g.user.email)
json.pop('debug', None)
if not json or 'type' not in json:
raise ValidationError('Resource needs a type.')
# todo there should be a way to better get subclassess resource
# defs
resource_def = app.resources[json['type']]
a = resource_def.schema.load(json)
if json['type'] == Snapshot.t:
tmp_snapshots = app.config['TMP_SNAPSHOTS']
path_snapshot = save_json(json, tmp_snapshots, g.user.email)
json.pop('debug', None)
a = resource_def.schema.load(json)
response = self.snapshot(a, resource_def)
move_json(tmp_snapshots, path_snapshot, g.user.email)
return response
@ -82,8 +118,8 @@ class ActionView(View):
pass
# TODO JN add compute rate with new visual test and old components device
if json['type'] == InitTransfer.t:
move_json(tmp_snapshots, path_snapshot, g.user.email)
return self.transfer_ownership()
a = resource_def.schema.load(json)
Model = db.Model._decl_class_registry.data[json['type']]()
action = Model(**a)
db.session.add(action)
@ -91,7 +127,6 @@ class ActionView(View):
ret = self.schema.jsonify(action)
ret.status_code = 201
db.session.commit()
move_json(tmp_snapshots, path_snapshot, g.user.email)
return ret
def one(self, id: UUID):
@ -108,6 +143,16 @@ class ActionView(View):
# model object, when we flush them to the db we will flush
# snapshot, and we want to wait to flush snapshot at the end
# If the device is allocated, then snapshot is a live
live = self.live(snapshot_json)
if live:
db.session.add(live)
db.session().final_flush()
ret = self.schema.jsonify(live) # transform it back
ret.status_code = 201
db.session.commit()
return ret
device = snapshot_json.pop('device') # type: Computer
components = None
if snapshot_json['software'] == (SnapshotSoftware.Workbench or SnapshotSoftware.WorkbenchAndroid):
@ -158,6 +203,7 @@ class ActionView(View):
# Check if HID is null and add Severity:Warning to Snapshot
if snapshot.device.hid is None:
snapshot.severity = Severity.Warning
db.session.add(snapshot)
db.session().final_flush()
ret = self.schema.jsonify(snapshot) # transform it back
@ -165,6 +211,70 @@ class ActionView(View):
db.session.commit()
return ret
def get_hdd_details(self, snapshot, device):
"""We get the liftime and serial_number of the disk"""
usage_time_hdd = None
serial_number = None
for hd in snapshot['components']:
if not isinstance(hd, DataStorage):
continue
serial_number = hd.serial_number
for act in hd.actions:
if not act.type == "TestDataStorage":
continue
usage_time_hdd = act.lifetime
break
if usage_time_hdd:
break
if not serial_number:
"There aren't any disk"
raise ResourceNotFound("There aren't any disk in this device {}".format(device))
return usage_time_hdd, serial_number
def live(self, snapshot):
"""If the device.allocated == True, then this snapshot create an action live."""
device = snapshot.get('device') # type: Computer
# TODO @cayop dependency of pulls 85 and 83
# if the pr/85 and pr/83 is merged, then you need change this way for get the device
if not device.hid or not Device.query.filter(Device.hid==device.hid).count():
return None
device = Device.query.filter(Device.hid==device.hid).one()
if not device.allocated:
return None
usage_time_hdd, serial_number = self.get_hdd_details(snapshot, device)
data_live = {'usage_time_hdd': usage_time_hdd,
'serial_number': serial_number,
'snapshot_uuid': snapshot['uuid'],
'description': '',
'device': device}
live = Live(**data_live)
if not usage_time_hdd:
warning = f"We don't found any TestDataStorage for disk sn: {serial_number}"
live.severity = Severity.Warning
live.description = warning
return live
live.sort_actions()
diff_time = live.diff_time()
if diff_time is None:
warning = "Don't exist one previous live or snapshot as reference"
live.description += warning
live.severity = Severity.Warning
elif diff_time < timedelta(0):
warning = "The difference with the last live/snapshot is negative"
live.description += warning
live.severity = Severity.Warning
return live
def transfer_ownership(self):
"""Perform a InitTransfer action to change author_id of device"""
pass

8
ereuse_devicehub/resources/device/definitions.py
View File

@ -27,11 +27,13 @@ class DeviceDef(Resource):
url_prefix, subdomain, url_defaults, root_path, cli_commands)
device_merge = DeviceMergeView.as_view('merge-devices', definition=self, auth=app.auth)
if self.AUTH:
device_merge = app.auth.requires_auth(device_merge)
self.add_url_rule('/<{}:{}>/merge/'.format(self.ID_CONVERTER.value, self.ID_NAME),
view_func=device_merge,
methods={'POST'})
path = '/<{value}:dev1_id>/merge/<{value}:dev2_id>'.format(value=self.ID_CONVERTER.value)
self.add_url_rule(path, view_func=device_merge, methods={'POST'})
class ComputerDef(DeviceDef):

37
ereuse_devicehub/resources/device/models.py
View File

@ -112,6 +112,8 @@ class Device(Thing):
nullable=False,
default=lambda: g.user.id)
owner = db.relationship(User, primaryjoin=owner_id == User.id)
allocated = db.Column(Boolean, default=False)
allocated.comment = "device is allocated or not."
_NON_PHYSICAL_PROPS = {
'id',
@ -133,7 +135,8 @@ class Device(Thing):
'variant',
'version',
'sku',
'image'
'image',
'allocated'
}
__table_args__ = (
@ -156,7 +159,7 @@ class Device(Thing):
Actions are returned by descending ``created`` time.
"""
return sorted(chain(self.actions_multiple, self.actions_one))
return sorted(chain(self.actions_multiple, self.actions_one), key=lambda x: x.created)
@property
def problems(self):
@ -229,6 +232,22 @@ class Device(Thing):
action = self.last_action_of(*states.Physical.actions())
return states.Physical(action.__class__)
@property
def traking(self):
"""The actual traking state, None otherwise."""
from ereuse_devicehub.resources.device import states
with suppress(LookupError, ValueError):
action = self.last_action_of(*states.Traking.actions())
return states.Traking(action.__class__)
@property
def usage(self):
"""The actual usage state, None otherwise."""
from ereuse_devicehub.resources.device import states
with suppress(LookupError, ValueError):
action = self.last_action_of(*states.Usage.actions())
return states.Usage(action.__class__)
@property
def physical_possessor(self):
"""The actual physical possessor or None.
@ -245,10 +264,12 @@ class Device(Thing):
and :class:`ereuse_devicehub.resources.action.models.Receive`
changes it.
"""
from ereuse_devicehub.resources.action.models import Receive
with suppress(LookupError):
action = self.last_action_of(Receive)
return action.agent
pass
# TODO @cayop uncomment this lines for link the possessor with the device
# from ereuse_devicehub.resources.action.models import Receive
# with suppress(LookupError):
# action = self.last_action_of(Receive)
# return action.agent_to
@property
def working(self):
@ -284,7 +305,9 @@ class Device(Thing):
"""
try:
# noinspection PyTypeHints
return next(e for e in reversed(self.actions) if isinstance(e, types))
actions = self.actions
actions.sort(key=lambda x: x.created)
return next(e for e in reversed(actions) if isinstance(e, types))
except StopIteration:
raise LookupError('{!r} does not contain actions of types {}.'.format(self, types))

3
ereuse_devicehub/resources/device/schemas.py
View File

@ -52,6 +52,8 @@ class Device(Thing):
price = NestedOn('Price', dump_only=True, description=m.Device.price.__doc__)
trading = EnumField(states.Trading, dump_only=True, description=m.Device.trading.__doc__)
physical = EnumField(states.Physical, dump_only=True, description=m.Device.physical.__doc__)
traking= EnumField(states.Traking, dump_only=True, description=m.Device.physical.__doc__)
usage = EnumField(states.Usage, dump_only=True, description=m.Device.physical.__doc__)
physical_possessor = NestedOn('Agent', dump_only=True, data_key='physicalPossessor')
production_date = DateTime('iso',
description=m.Device.updated.comment,
@ -63,6 +65,7 @@ class Device(Thing):
variant = SanitizedStr(description=m.Device.variant.comment)
sku = SanitizedStr(description=m.Device.sku.comment)
image = URL(description=m.Device.image.comment)
allocated = Boolean(description=m.Device.allocated.comment)
@pre_load
def from_actions_to_actions_one(self, data: dict):

21
ereuse_devicehub/resources/device/states.py
View File

@ -51,11 +51,30 @@ class Physical(State):
:cvar Preparing: The device is going to be or being prepared.
:cvar Prepared: The device has been prepared.
:cvar Ready: The device is in working conditions.
:cvar InUse: The device is being reported to be in active use.
"""
ToBeRepaired = e.ToRepair
Repaired = e.Repair
Preparing = e.ToPrepare
Prepared = e.Prepare
Ready = e.Ready
class Traking(State):
"""Traking states.
:cvar Receive: The device changes hands
"""
# Receive = e.Receive
pass
class Usage(State):
"""Usage states.
:cvar Allocate: The device is allocate in other Agent (organization, person ...)
:cvar Deallocate: The device is deallocate and return to the owner
:cvar InUse: The device is being reported to be in active use.
"""
Allocate = e.Allocate
Deallocate = e.Deallocate
InUse = e.Live

2
ereuse_devicehub/resources/device/sync.py
View File

@ -155,6 +155,8 @@ class Sync:
if device.hid:
with suppress(ResourceNotFound):
db_device = Device.query.filter_by(hid=device.hid, owner_id=g.user.id).one()
if db_device and db_device.allocated:
raise ResourceNotFound('device is actually allocated {}'.format(device))
try:
tags = {Tag.from_an_id(tag.id).one() for tag in device.tags} # type: Set[Tag]
except ResourceNotFound:

96
ereuse_devicehub/resources/device/views.py
View File

@ -6,10 +6,13 @@ import marshmallow
from flask import g, current_app as app, render_template, request, Response
from flask.json import jsonify
from flask_sqlalchemy import Pagination
from sqlalchemy.util import OrderedSet
from marshmallow import fields, fields as f, validate as v, Schema as MarshmallowSchema
from teal import query
from teal.db import ResourceNotFound
from teal.cache import cache
from teal.resource import View
from teal.marshmallow import ValidationError
from ereuse_devicehub import auth
from ereuse_devicehub.db import db
@ -161,66 +164,83 @@ class DeviceView(View):
class DeviceMergeView(View):
"""View for merging two devices
Ex. ``device/<id>/merge/id=X``.
Ex. ``device/<dev1_id>/merge/<dev2_id>``.
"""
class FindArgs(MarshmallowSchema):
id = fields.Integer()
def post(self, dev1_id: int, dev2_id: int):
device = self.merge_devices(dev1_id, dev2_id)
def get_merge_id(self) -> uuid.UUID:
args = self.QUERY_PARSER.parse(self.find_args, request, locations=('querystring',))
return args['id']
def post(self, id: uuid.UUID):
device = Device.query.filter_by(id=id, owner_id=g.user.id).one()
with_device = Device.query.filter_by(id=self.get_merge_id(), owner_id=g.user.id).one()
self.merge_devices(device, with_device)
db.session().final_flush()
ret = self.schema.jsonify(device)
ret.status_code = 201
db.session.commit()
return ret
def merge_devices(self, base_device, with_device):
"""Merge the current device with `with_device` by
adding all `with_device` actions under the current device.
@auth.Auth.requires_auth
def merge_devices(self, dev1_id: int, dev2_id: int) -> Device:
"""Merge the current device with `with_device` (dev2_id) by
adding all `with_device` actions under the current device, (dev1_id).
This operation is highly costly as it forces refreshing
many models in session.
"""
snapshots = sorted(
filterfalse(lambda x: not isinstance(x, actions.Snapshot), (base_device.actions + with_device.actions)))
workbench_snapshots = [s for s in snapshots if
s.software == (SnapshotSoftware.Workbench or SnapshotSoftware.WorkbenchAndroid)]
latest_snapshot_device = [d for d in (base_device, with_device) if d.id == snapshots[-1].device.id][0]
latest_snapshotworkbench_device = \
[d for d in (base_device, with_device) if d.id == workbench_snapshots[-1].device.id][0]
# Adding actions of with_device
with_actions_one = [a for a in with_device.actions if isinstance(a, actions.ActionWithOneDevice)]
with_actions_multiple = [a for a in with_device.actions if isinstance(a, actions.ActionWithMultipleDevices)]
# base_device = Device.query.filter_by(id=dev1_id, owner_id=g.user.id).one()
self.base_device = Device.query.filter_by(id=dev1_id, owner_id=g.user.id).one()
self.with_device = Device.query.filter_by(id=dev2_id, owner_id=g.user.id).one()
if self.base_device.allocated or self.with_device.allocated:
# Validation than any device is allocated
msg = 'The device is allocated, please deallocated before merge.'
raise ValidationError(msg)
if not self.base_device.type == self.with_device.type:
# Validation than we are speaking of the same kind of devices
raise ValidationError('The devices is not the same type.')
# Adding actions of self.with_device
with_actions_one = [a for a in self.with_device.actions
if isinstance(a, actions.ActionWithOneDevice)]
with_actions_multiple = [a for a in self.with_device.actions
if isinstance(a, actions.ActionWithMultipleDevices)]
# Moving the tags from `with_device` to `base_device`
# Union of tags the device had plus the (potentially) new ones
self.base_device.tags.update([x for x in self.with_device.tags])
self.with_device.tags.clear() # We don't want to add the transient dummy tags
db.session.add(self.with_device)
# Moving the actions from `with_device` to `base_device`
for action in with_actions_one:
if action.parent:
action.parent = base_device
action.parent = self.base_device
else:
base_device.actions_one.add(action)
self.base_device.actions_one.add(action)
for action in with_actions_multiple:
if action.parent:
action.parent = base_device
action.parent = self.base_device
else:
base_device.actions_multiple.add(action)
self.base_device.actions_multiple.add(action)
# Keeping the components of latest SnapshotWorkbench
base_device.components = latest_snapshotworkbench_device.components
# Keeping the components of with_device
components = OrderedSet(c for c in self.with_device.components)
self.base_device.components = components
# Properties from latest Snapshot
base_device.type = latest_snapshot_device.type
base_device.hid = latest_snapshot_device.hid
base_device.manufacturer = latest_snapshot_device.manufacturer
base_device.model = latest_snapshot_device.model
base_device.chassis = latest_snapshot_device.chassis
# Properties from with_device
self.merge()
db.session().add(self.base_device)
db.session().final_flush()
return self.base_device
def merge(self):
"""Copies the physical properties of the base_device to the with_device.
This method mutates base_device.
"""
for field_name, value in self.with_device.physical_properties.items():
if value is not None:
setattr(self.base_device, field_name, value)
self.base_device.hid = self.with_device.hid
class ManufacturerView(View):

0
ereuse_devicehub/resources/metric/__init__.py Normal file
View File

10
ereuse_devicehub/resources/metric/definitions.py Normal file
View File

@ -0,0 +1,10 @@
from teal.resource import Resource
from ereuse_devicehub.resources.metric.schema import Metric
from ereuse_devicehub.resources.metric.views import MetricsView
class MetricDef(Resource):
__type__ = 'Metric'
VIEW = MetricsView
SCHEMA = Metric
AUTH = True

11
ereuse_devicehub/resources/metric/schema.py Normal file
View File

@ -0,0 +1,11 @@
from teal.resource import Schema
from marshmallow.fields import DateTime
class Metric(Schema):
"""
This schema filter dates for search the metrics
"""
start_time = DateTime(data_key='start_time', required=True,
description="Start date for search metrics")
end_time = DateTime(data_key='end_time', required=True,
description="End date for search metrics")

44
ereuse_devicehub/resources/metric/views.py Normal file
View File

@ -0,0 +1,44 @@
from flask import request, g, jsonify
from contextlib import suppress
from teal.resource import View
from ereuse_devicehub.resources.action import schemas
from ereuse_devicehub.resources.action.models import Allocate, Live, Action, ToRepair, ToPrepare
from ereuse_devicehub.resources.device import models as m
from ereuse_devicehub.resources.metric.schema import Metric
class MetricsView(View):
def find(self, args: dict):
metrics = {
"allocateds": self.allocated(),
"live": self.live(),
}
return jsonify(metrics)
def allocated(self):
# TODO @cayop we need uncomment when the pr/83 is approved
# return m.Device.query.filter(m.Device.allocated==True, owner==g.user).count()
return m.Device.query.filter(m.Device.allocated==True).count()
def live(self):
# TODO @cayop we need uncomment when the pr/83 is approved
# devices = m.Device.query.filter(m.Device.allocated==True, owner==g.user)
devices = m.Device.query.filter(m.Device.allocated==True)
count = 0
for dev in devices:
live = allocate = None
with suppress(LookupError):
live = dev.last_action_of(Live)
with suppress(LookupError):
allocate = dev.last_action_of(Allocate)
if not live:
continue
if allocate and allocate.created > live.created:
continue
count += 1
return count

353
tests/test_action.py
View File

@ -1,15 +1,18 @@
import ipaddress
from datetime import timedelta
import copy
import pytest
from datetime import datetime, timedelta
from decimal import Decimal
from typing import Tuple, Type
import pytest
from flask import current_app as app, g
from sqlalchemy.util import OrderedSet
from teal.enums import Currency, Subdivision
from ereuse_devicehub.client import UserClient
from ereuse_devicehub.db import db
from ereuse_devicehub.client import UserClient
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources import enums
from ereuse_devicehub.resources.action import models
from ereuse_devicehub.resources.device import states
@ -243,29 +246,329 @@ def test_generic_action(action_model_state: Tuple[models.Action, states.Trading]
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.auth_app_context.__name__)
def test_live():
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_live(user: UserClient, app: Devicehub):
"""Tests inserting a Live into the database and GETting it."""
db_live = models.Live(ip=ipaddress.ip_address('79.147.10.10'),
subdivision_confidence=84,
subdivision=Subdivision['ES-CA'],
city='barcelona',
city_confidence=20,
isp='acme',
device=Desktop(serial_number='sn1', model='ml1', manufacturer='mr1',
chassis=ComputerChassis.Docking),
organization='acme1',
organization_type='acme1bis')
db.session.add(db_live)
db.session.commit()
client = UserClient(app, 'foo@foo.com', 'foo', response_wrapper=app.response_class)
client.login()
live, _ = client.get(res=models.Action, item=str(db_live.id))
assert live['ip'] == '79.147.10.10'
assert live['subdivision'] == 'ES-CA'
assert live['country'] == 'ES'
device, _ = client.get(res=Device, item=live['device']['id'])
assert device['physical'] == states.Physical.InUse.name
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00"
}
user.post(res=models.Allocate, data=post_request)
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0]
hdd_action = [a for a in hdd['actions'] if a['type'] == 'TestDataStorage'][0]
hdd_action['lifetime'] += 1000
snapshot, _ = user.post(acer, res=models.Snapshot)
db_device = Device.query.filter_by(id=1).one()
action_live = [a for a in db_device.actions if a.type == 'Live']
assert len(action_live) == 1
assert action_live[0].usage_time_hdd == timedelta(hours=hdd_action['lifetime'])
assert action_live[0].usage_time_allocate == timedelta(hours=1000)
assert action_live[0].final_user_code == post_request['finalUserCode']
assert action_live[0].serial_number == 'wd-wx11a80w7430'
assert str(action_live[0].snapshot_uuid) == acer['uuid']
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_live_without_TestDataStorage(user: UserClient, app: Devicehub):
"""Tests inserting a Live into the database and GETting it.
If the live don't have a TestDataStorage, then save live and response None
"""
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00"
}
user.post(res=models.Allocate, data=post_request)
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
actions = [a for a in acer['components'][7]['actions'] if a['type'] != 'TestDataStorage']
acer['components'][7]['actions'] = actions
live, _ = user.post(acer, res=models.Snapshot)
assert live['type'] == 'Live'
assert live['serialNumber'] == 'wd-wx11a80w7430'
assert live['severity'] == 'Warning'
description = "We don't found any TestDataStorage for disk sn: wd-wx11a80w7430"
assert live['description'] == description
db_live = models.Live.query.filter_by(id=live['id']).one()
assert db_live.usage_time_hdd is None
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_live_without_hdd_1(user: UserClient, app: Devicehub):
"""Tests inserting a Live into the database and GETting it.
The snapshot have hdd but the live no, and response 404
"""
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00"
}
user.post(res=models.Allocate, data=post_request)
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
components = [a for a in acer['components'] if a['type'] != 'HardDrive']
acer['components'] = components
response, _ = user.post(acer, res=models.Snapshot, status=404)
assert "The There aren't any disk in this device" in response['message']
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_live_without_hdd_2(user: UserClient, app: Devicehub):
"""Tests inserting a Live into the database and GETting it.
The snapshot haven't hdd and the live neither, and response 404
"""
acer = file('acer.happy.battery.snapshot')
components = [a for a in acer['components'] if a['type'] != 'HardDrive']
acer['components'] = components
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00"
}
user.post(res=models.Allocate, data=post_request)
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
response, _ = user.post(acer, res=models.Snapshot, status=404)
assert "The There aren't any disk in this device" in response['message']
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_live_without_hdd_3(user: UserClient, app: Devicehub):
"""Tests inserting a Live into the database and GETting it.
The snapshot haven't hdd and the live have, and save the live
with usage_time_allocate == 0
"""
acer = file('acer.happy.battery.snapshot')
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
components = [a for a in acer['components'] if a['type'] != 'HardDrive']
acer['components'] = components
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00"
}
user.post(res=models.Allocate, data=post_request)
acer = file('acer.happy.battery.snapshot')
live, _ = user.post(acer, res=models.Snapshot)
assert live['type'] == 'Live'
assert live['serialNumber'] == 'wd-wx11a80w7430'
assert live['severity'] == 'Warning'
description = "Don't exist one previous live or snapshot as reference"
assert live['description'] == description
db_live = models.Live.query.filter_by(id=live['id']).one()
assert str(db_live.usage_time_hdd) == '195 days, 12:00:00'
assert str(db_live.usage_time_allocate) == '0:00:00'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_live_with_hdd_with_old_time(user: UserClient, app: Devicehub):
"""Tests inserting a Live into the database and GETting it.
The snapshot hdd have a lifetime higher than lifetime of the live action
save the live with usage_time_allocate == 0
"""
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00"
}
user.post(res=models.Allocate, data=post_request)
acer = file('acer.happy.battery.snapshot')
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
action = [a for a in acer['components'][7]['actions'] if a['type'] == 'TestDataStorage']
action[0]['lifetime'] -= 100
live, _ = user.post(acer, res=models.Snapshot)
assert live['type'] == 'Live'
assert live['serialNumber'] == 'wd-wx11a80w7430'
assert live['severity'] == 'Warning'
description = "The difference with the last live/snapshot is negative"
assert live['description'] == description
db_live = models.Live.query.filter_by(id=live['id']).one()
assert str(db_live.usage_time_hdd) == '191 days, 8:00:00'
assert str(db_live.usage_time_allocate) == '0:00:00'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_live_search_last_allocate(user: UserClient, app: Devicehub):
"""Tests inserting a Live into the database and GETting it.
"""
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00"
}
user.post(res=models.Allocate, data=post_request)
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0]
hdd_action = [a for a in hdd['actions'] if a['type'] == 'TestDataStorage'][0]
hdd_action['lifetime'] += 1000
live, _ = user.post(acer, res=models.Snapshot)
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec4"
actions = [a for a in acer['components'][7]['actions'] if a['type'] != 'TestDataStorage']
acer['components'][7]['actions'] = actions
live, _ = user.post(acer, res=models.Snapshot)
assert live['usageTimeAllocate'] == 1000
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_allocate(user: UserClient):
""" Tests allocate """
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
device_id = snapshot['device']['id']
post_request = {"transaction": "ccc",
"finalUserCode": "aabbcc",
"name": "John",
"severity": "Info",
"endUsers": 1,
"devices": [device_id],
"description": "aaa",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00",
}
allocate, _ = user.post(res=models.Allocate, data=post_request)
# Normal allocate
device, _ = user.get(res=Device, item=device_id)
assert device['allocated'] == True
action = [a for a in device['actions'] if a['type'] == 'Allocate'][0]
assert action['transaction'] == allocate['transaction']
assert action['finalUserCode'] == allocate['finalUserCode']
assert action['created'] == allocate['created']
assert action['startTime'] == allocate['startTime']
assert action['endUsers'] == allocate['endUsers']
assert action['name'] == allocate['name']
post_bad_request1 = copy.copy(post_request)
post_bad_request1['endUsers'] = 2
post_bad_request2 = copy.copy(post_request)
post_bad_request2['startTime'] = "2020-11-01T02:00:00+00:01"
post_bad_request3 = copy.copy(post_request)
post_bad_request3['transaction'] = "aaa"
res1, _ = user.post(res=models.Allocate, data=post_bad_request1, status=422)
res2, _ = user.post(res=models.Allocate, data=post_bad_request2, status=422)
res3, _ = user.post(res=models.Allocate, data=post_bad_request3, status=422)
for r in (res1, res2, res3):
assert r['code'] == 422
assert r['type'] == 'ValidationError'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_allocate_bad_dates(user: UserClient):
""" Tests allocate """
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
device_id = snapshot['device']['id']
delta = timedelta(days=30)
future = datetime.now() + delta
post_request = {"transaction": "ccc",
"finalUserCode": "aabbcc",
"name": "John",
"severity": "Info",
"end_users": 1,
"devices": [device_id],
"description": "aaa",
"start_time": future,
}
res, _ = user.post(res=models.Allocate, data=post_request, status=422)
assert res['code'] == 422
assert res['type'] == 'ValidationError'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_deallocate(user: UserClient):
""" Tests deallocate """
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
device_id = snapshot['device']['id']
post_deallocate = {"startTime": "2020-11-01T02:00:00+00:00",
"transaction": "ccc",
"devices": [device_id]
}
res, _ = user.post(res=models.Deallocate, data=post_deallocate, status=422)
assert res['code'] == 422
assert res['type'] == 'ValidationError'
post_allocate = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00"
}
user.post(res=models.Allocate, data=post_allocate)
device, _ = user.get(res=Device, item=device_id)
assert device['allocated'] == True
deallocate, _ = user.post(res=models.Deallocate, data=post_deallocate)
assert deallocate['startTime'] == post_deallocate['startTime']
assert deallocate['devices'][0]['id'] == device_id
assert deallocate['devices'][0]['allocated'] == False
res, _ = user.post(res=models.Deallocate, data=post_deallocate, status=422)
assert res['code'] == 422
assert res['type'] == 'ValidationError'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_deallocate_bad_dates(user: UserClient):
""" Tests deallocate with bad date of start_time """
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
device_id = snapshot['device']['id']
delta = timedelta(days=30)
future = datetime.now() + delta
post_deallocate = {"startTime": future,
"devices": [device_id]
}
post_allocate = {"devices": [device_id], "description": "aaa",
"startTime": "2020-11-01T02:00:00+00:00"
}
user.post(res=models.Allocate, data=post_allocate)
res, _ = user.post(res=models.Deallocate, data=post_deallocate, status=422)
assert res['code'] == 422
assert res['type'] == 'ValidationError'
@pytest.mark.mvp

113
tests/test_basic.py
View File

@ -30,77 +30,80 @@ def test_api_docs(client: Client):
assert set(docs['paths'].keys()) == {
'/actions/',
'/apidocs',
'/batteries/{id}/merge/',
'/bikes/{id}/merge/',
'/cameras/{id}/merge/',
'/cellphones/{id}/merge/',
'/components/{id}/merge/',
'/computer-accessories/{id}/merge/',
'/computer-monitors/{id}/merge/',
'/computers/{id}/merge/',
'/cookings/{id}/merge/',
'/data-storages/{id}/merge/',
'/dehumidifiers/{id}/merge/',
'/batteries/{dev1_id}/merge/{dev2_id}',
'/bikes/{dev1_id}/merge/{dev2_id}',
'/cameras/{dev1_id}/merge/{dev2_id}',
'/cellphones/{dev1_id}/merge/{dev2_id}',
'/components/{dev1_id}/merge/{dev2_id}',
'/computer-accessories/{dev1_id}/merge/{dev2_id}',
'/computer-monitors/{dev1_id}/merge/{dev2_id}',
'/computers/{dev1_id}/merge/{dev2_id}',
'/cookings/{dev1_id}/merge/{dev2_id}',
'/data-storages/{dev1_id}/merge/{dev2_id}',
'/dehumidifiers/{dev1_id}/merge/{dev2_id}',
'/deliverynotes/',
'/desktops/{id}/merge/',
'/desktops/{dev1_id}/merge/{dev2_id}',
'/devices/',
'/devices/static/{filename}',
'/devices/{id}/merge/',
'/displays/{id}/merge/',
'/diy-and-gardenings/{id}/merge/',
'/devices/{dev1_id}/merge/{dev2_id}',
'/displays/{dev1_id}/merge/{dev2_id}',
'/diy-and-gardenings/{dev1_id}/merge/{dev2_id}',
'/documents/devices/',
'/documents/erasures/',
'/documents/lots/',
'/documents/static/{filename}',
'/documents/stock/',
'/drills/{id}/merge/',
'/graphic-cards/{id}/merge/',
'/hard-drives/{id}/merge/',
'/homes/{id}/merge/',
'/hubs/{id}/merge/',
'/keyboards/{id}/merge/',
'/label-printers/{id}/merge/',
'/laptops/{id}/merge/',
'/drills/{dev1_id}/merge/{dev2_id}',
'/graphic-cards/{dev1_id}/merge/{dev2_id}',
'/hard-drives/{dev1_id}/merge/{dev2_id}',
'/homes/{dev1_id}/merge/{dev2_id}',
'/hubs/{dev1_id}/merge/{dev2_id}',
'/keyboards/{dev1_id}/merge/{dev2_id}',
'/label-printers/{dev1_id}/merge/{dev2_id}',
'/laptops/{dev1_id}/merge/{dev2_id}',
'/lots/',
'/lots/{id}/children',
'/lots/{id}/devices',
'/manufacturers/',
'/memory-card-readers/{id}/merge/',
'/mice/{id}/merge/',
'/microphones/{id}/merge/',
'/mixers/{id}/merge/',
'/mobiles/{id}/merge/',
'/monitors/{id}/merge/',
'/motherboards/{id}/merge/',
'/network-adapters/{id}/merge/',
'/networkings/{id}/merge/',
'/pack-of-screwdrivers/{id}/merge/',
'/printers/{id}/merge/',
'/processors/{id}/merge/',
'/memory-card-readers/{dev1_id}/merge/{dev2_id}',
'/mice/{dev1_id}/merge/{dev2_id}',
'/microphones/{dev1_id}/merge/{dev2_id}',
'/mixers/{dev1_id}/merge/{dev2_id}',
'/mobiles/{dev1_id}/merge/{dev2_id}',
'/monitors/{dev1_id}/merge/{dev2_id}',
'/motherboards/{dev1_id}/merge/{dev2_id}',
'/network-adapters/{dev1_id}/merge/{dev2_id}',
'/networkings/{dev1_id}/merge/{dev2_id}',
'/pack-of-screwdrivers/{dev1_id}/merge/{dev2_id}',
'/printers/{dev1_id}/merge/{dev2_id}',
'/processors/{dev1_id}/merge/{dev2_id}',
'/proofs/',
'/rackets/{id}/merge/',
'/ram-modules/{id}/merge/',
'/recreations/{id}/merge/',
'/routers/{id}/merge/',
'/sais/{id}/merge/',
'/servers/{id}/merge/',
'/smartphones/{id}/merge/',
'/solid-state-drives/{id}/merge/',
'/sound-cards/{id}/merge/',
'/sounds/{id}/merge/',
'/stairs/{id}/merge/',
'/switches/{id}/merge/',
'/tablets/{id}/merge/',
'/rackets/{dev1_id}/merge/{dev2_id}',
'/ram-modules/{dev1_id}/merge/{dev2_id}',
'/recreations/{dev1_id}/merge/{dev2_id}',
'/routers/{dev1_id}/merge/{dev2_id}',
'/sais/{dev1_id}/merge/{dev2_id}',
'/servers/{dev1_id}/merge/{dev2_id}',
'/smartphones/{dev1_id}/merge/{dev2_id}',
'/solid-state-drives/{dev1_id}/merge/{dev2_id}',
'/sound-cards/{dev1_id}/merge/{dev2_id}',
'/sounds/{dev1_id}/merge/{dev2_id}',
'/stairs/{dev1_id}/merge/{dev2_id}',
'/switches/{dev1_id}/merge/{dev2_id}',
'/tablets/{dev1_id}/merge/{dev2_id}',
'/tags/',
'/tags/{tag_id}/device/{device_id}',
'/television-sets/{id}/merge/',
'/television-sets/{dev1_id}/merge/{dev2_id}',
'/users/',
'/users/login/',
'/video-scalers/{id}/merge/',
'/videoconferences/{id}/merge/',
'/videos/{id}/merge/',
'/wireless-access-points/{id}/merge/',
'/versions/'
'/video-scalers/{dev1_id}/merge/{dev2_id}',
'/videoconferences/{dev1_id}/merge/{dev2_id}',
'/videos/{dev1_id}/merge/{dev2_id}',
'/wireless-access-points/{dev1_id}/merge/{dev2_id}',
'/versions/',
'/allocates/',
'/deallocates/',
'/metrics/',
}
assert docs['info'] == {'title': 'Devicehub', 'version': '0.2'}
assert docs['components']['securitySchemes']['bearerAuth'] == {
@ -111,4 +114,4 @@ def test_api_docs(client: Client):
'scheme': 'basic',
'name': 'Authorization'
}
assert len(docs['definitions']) == 122
assert len(docs['definitions']) == 124

87
tests/test_merge.py Normal file
View File

@ -0,0 +1,87 @@
import datetime
from uuid import UUID
from flask import g
import pytest
from ereuse_devicehub.client import Client, UserClient
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.action import models as m
from ereuse_devicehub.resources.device import models as d
from ereuse_devicehub.resources.tag import Tag
from tests import conftest
from tests.conftest import file as import_snap
@pytest.mark.mvp
def test_simple_merge(app: Devicehub, user: UserClient):
""" Check if is correct to do a manual merge """
snapshot1, _ = user.post(import_snap('real-custom.snapshot.11'), res=m.Snapshot)
snapshot2, _ = user.post(import_snap('real-hp.snapshot.11'), res=m.Snapshot)
pc1_id = snapshot1['device']['id']
pc2_id = snapshot2['device']['id']
with app.app_context():
pc1 = d.Device.query.filter_by(id=pc1_id).one()
pc2 = d.Device.query.filter_by(id=pc2_id).one()
n_actions1 = len(pc1.actions)
n_actions2 = len(pc2.actions)
action1 = pc1.actions[0]
action2 = pc2.actions[0]
assert not action2 in pc1.actions
tag = Tag(id='foo-bar', owner_id=user.user['id'])
pc2.tags.add(tag)
db.session.add(pc2)
db.session.commit()
components1 = [com for com in pc1.components]
components2 = [com for com in pc2.components]
components1_excluded = [com for com in pc1.components if not com in components2]
assert pc1.hid != pc2.hid
assert not tag in pc1.tags
uri = '/devices/%d/merge/%d' % (pc1_id, pc2_id)
result, _ = user.post({'id': 1}, uri=uri, status=201)
assert pc1.hid == pc2.hid
assert action1 in pc1.actions
assert action2 in pc1.actions
assert len(pc1.actions) == n_actions1 + n_actions2
assert set(pc2.components) == set()
assert tag in pc1.tags
assert not tag in pc2.tags
for com in components2:
assert com in pc1.components
for com in components1_excluded:
assert not com in pc1.components
@pytest.mark.mvp
def test_merge_two_device_with_differents_tags(app: Devicehub, user: UserClient):
""" Check if is correct to do a manual merge of 2 diferents devices with diferents tags """
snapshot1, _ = user.post(import_snap('real-custom.snapshot.11'), res=m.Snapshot)
snapshot2, _ = user.post(import_snap('real-hp.snapshot.11'), res=m.Snapshot)
pc1_id = snapshot1['device']['id']
pc2_id = snapshot2['device']['id']
with app.app_context():
pc1 = d.Device.query.filter_by(id=pc1_id).one()
pc2 = d.Device.query.filter_by(id=pc2_id).one()
tag1 = Tag(id='fii-bor', owner_id=user.user['id'])
tag2 = Tag(id='foo-bar', owner_id=user.user['id'])
pc1.tags.add(tag1)
pc2.tags.add(tag2)
db.session.add(pc1)
db.session.add(pc2)
db.session.commit()
uri = '/devices/%d/merge/%d' % (pc1_id, pc2_id)
result, _ = user.post({'id': 1}, uri=uri, status=201)
assert pc1.hid == pc2.hid
assert tag1 in pc1.tags
assert tag2 in pc1.tags

118
tests/test_metrics.py Normal file
View File

@ -0,0 +1,118 @@
import pytest
from ereuse_devicehub.client import UserClient
from ereuse_devicehub.resources.action import models as ma
from tests import conftest
from tests.conftest import file
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_simple_metrics(user: UserClient):
""" Checks one standard query of metrics """
# Insert computer
lenovo = file('desktop-9644w8n-lenovo-0169622.snapshot')
acer = file('acer.happy.battery.snapshot')
user.post(lenovo, res=ma.Snapshot)
snapshot, _ = user.post(acer, res=ma.Snapshot)
device_id = snapshot['device']['id']
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"finalUserCode": "abcdefjhi",
"devices": [device_id], "description": "aaa",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00"
}
# Create Allocate
user.post(res=ma.Allocate, data=post_request)
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0]
hdd_action = [a for a in hdd['actions'] if a['type'] == 'TestDataStorage'][0]
hdd_action['powerCycleCount'] += 1000
user.post(acer, res=ma.Snapshot)
# Create a live
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec4"
hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0]
hdd_action = [a for a in hdd['actions'] if a['type'] == 'TestDataStorage'][0]
hdd_action['powerCycleCount'] += 1000
user.post(acer, res=ma.Snapshot)
# Create an other live
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec5"
hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0]
hdd_action = [a for a in hdd['actions'] if a['type'] == 'TestDataStorage'][0]
hdd_action['powerCycleCount'] += 1000
user.post(acer, res=ma.Snapshot)
# Check metrics
metrics = {'allocateds': 1, 'live': 1}
res, _ = user.get("/metrics/")
assert res == metrics
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_second_hdd_metrics(user: UserClient):
""" Checks one standard query of metrics """
# Insert computer
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=ma.Snapshot)
device_id = snapshot['device']['id']
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"finalUserCode": "abcdefjhi",
"devices": [device_id], "description": "aaa",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00"
}
# Create Allocate
user.post(res=ma.Allocate, data=post_request)
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec3"
hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0]
hdd_action = [a for a in hdd['actions'] if a['type'] == 'TestDataStorage'][0]
hdd_action['powerCycleCount'] += 1000
user.post(acer, res=ma.Snapshot)
# Create a live
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec4"
hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0]
hdd_action = [a for a in hdd['actions'] if a['type'] == 'TestDataStorage'][0]
hdd_action['powerCycleCount'] += 1000
user.post(acer, res=ma.Snapshot)
# Create a second device
acer['uuid'] = "490fb8c0-81a1-42e9-95e0-5e7db7038ec5"
hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0]
hdd['serialNumber'] = 'WD-WX11A80W7440'
user.post(acer, res=ma.Snapshot)
# Check metrics if we change the hdd we need a result of one device
metrics = {'allocateds': 1, 'live': 1}
res, _ = user.get("/metrics/")
assert res == metrics
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_metrics_with_live_null(user: UserClient):
""" Checks one standard query of metrics """
# Insert computer
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=ma.Snapshot)
device_id = snapshot['device']['id']
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"finalUserCode": "abcdefjhi",
"devices": [device_id], "description": "aaa",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00"
}
# Create Allocate
user.post(res=ma.Allocate, data=post_request)
# Check metrics if we change the hdd we need a result of one device
metrics = {'allocateds': 1, 'live': 0}
res, _ = user.get("/metrics/")
assert res == metrics

21
tests/test_snapshot.py
View File

@ -19,7 +19,7 @@ from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.action.models import Action, BenchmarkDataStorage, \
BenchmarkProcessor, EraseSectors, RateComputer, Snapshot, SnapshotRequest, VisualTest, \
EreusePrice
EreusePrice, Ready
from ereuse_devicehub.resources.device import models as m
from ereuse_devicehub.resources.device.exceptions import NeedsId
from ereuse_devicehub.resources.device.models import SolidStateDrive
@ -562,6 +562,25 @@ def test_save_snapshot_in_file(app: Devicehub, user: UserClient):
assert snapshot['version'] == snapshot_no_hid['version']
assert snapshot['uuid'] == uuid
@pytest.mark.mvp
def test_action_no_snapshot_without_save_file(app: Devicehub, user: UserClient):
""" This test check if the function save_snapshot_in_file not work when we
send one other action different to snapshot
"""
s = file('laptop-hp_255_g3_notebook-hewlett-packard-cnd52270fw.snapshot')
snapshot, _ = user.post(res=Snapshot, data=s)
tmp_snapshots = app.config['TMP_SNAPSHOTS']
path_dir_base = os.path.join(tmp_snapshots, user.user['email'])
shutil.rmtree(tmp_snapshots)
action = {'type': Ready.t, 'devices': [snapshot['device']['id']]}
action, _ = user.post(action, res=Action)
assert os.path.exists(tmp_snapshots) == False
@pytest.mark.mvp
def test_save_snapshot_with_debug(app: Devicehub, user: UserClient):
""" This test check if works the function save_snapshot_in_file """