diff --git a/ereuse_devicehub/marshmallow.py b/ereuse_devicehub/marshmallow.py index 5e835521..0c672f54 100644 --- a/ereuse_devicehub/marshmallow.py +++ b/ereuse_devicehub/marshmallow.py @@ -6,6 +6,7 @@ from teal.marshmallow import NestedOn as TealNestedOn class NestedOn(TealNestedOn): - def __init__(self, nested, polymorphic_on='type', default=missing_, exclude=tuple(), - only=None, db: SQLAlchemy = db, **kwargs): - super().__init__(nested, polymorphic_on, default, exclude, only, db, **kwargs) + + def __init__(self, nested, polymorphic_on='type', db: SQLAlchemy = db, default=missing_, + exclude=tuple(), only=None, **kwargs): + super().__init__(nested, polymorphic_on, db, default, exclude, only, **kwargs) diff --git a/ereuse_devicehub/resources/device/models.py b/ereuse_devicehub/resources/device/models.py index 0ecec0fb..c45921bf 100644 --- a/ereuse_devicehub/resources/device/models.py +++ b/ereuse_devicehub/resources/device/models.py @@ -5,7 +5,7 @@ from ereuse_utils.naming import Naming from sqlalchemy import BigInteger, Column, Float, ForeignKey, Integer, Sequence, SmallInteger, \ Unicode, inspect from sqlalchemy.ext.declarative import declared_attr -from sqlalchemy.orm import backref, relationship +from sqlalchemy.orm import ColumnProperty, backref, relationship from ereuse_devicehub.resources.models import STR_BIG_SIZE, STR_SIZE, STR_SM_SIZE, Thing, \ check_range @@ -28,8 +28,13 @@ class Device(Thing): height = Column(Float(precision=3, decimal_return_scale=3), check_range('height', 0.1, 3)) # type: float + def __init__(self, *args, **kw) -> None: + super().__init__(*args, **kw) + with suppress(TypeError): + self.hid = Naming.hid(self.manufacturer, self.serial_number, self.model) # type: str + @property - def physical_properties(self) -> Dict[Column, object or None]: + def physical_properties(self) -> Dict[str, object or None]: """ Fields that describe the physical properties of a device. @@ -39,9 +44,11 @@ class Device(Thing): """ # todo ensure to remove materialized values when start using them # todo or self.__table__.columns if inspect fails - return {c: getattr(self, c.name, None) + return {c.key: getattr(self, c.key, None) for c in inspect(self.__class__).attrs - if not c.foreign_keys and c not in {self.id, self.type}} + if isinstance(c, ColumnProperty) + and not getattr(c, 'foreign_keys', None) + and c.key not in {'id', 'type', 'created', 'updated', 'parent_id', 'hid'}} @declared_attr def __mapper_args__(cls): @@ -57,10 +64,8 @@ class Device(Thing): args[POLYMORPHIC_ON] = cls.type return args - def __init__(self, *args, **kw) -> None: - super().__init__(*args, **kw) - with suppress(TypeError): - self.hid = Naming.hid(self.manufacturer, self.serial_number, self.model) + def __lt__(self, other): + return self.id < other.id class Computer(Device): diff --git a/ereuse_devicehub/resources/device/sync.py b/ereuse_devicehub/resources/device/sync.py index 23125e57..670e2df0 100644 --- a/ereuse_devicehub/resources/device/sync.py +++ b/ereuse_devicehub/resources/device/sync.py @@ -1,3 +1,4 @@ +import re from contextlib import suppress from itertools import groupby from typing import Iterable, List, Set @@ -18,7 +19,7 @@ class Sync: @classmethod def run(cls, device: Device, components: Iterable[Component] or None, - force_creation: bool = False) -> (Device, List[Component], List[Add or Remove]): + force_creation: bool = False) -> (Device, List[Add or Remove]): """ Synchronizes the device and components with the database. @@ -40,39 +41,41 @@ class Sync: it doesn't generate HID or have an ID? Only for the device param. :return: A tuple of: - 1. The device from the database (with an ID). - 2. The same passed-in components from the database (with - ids). - 3. A list of Add / Remove (not yet added to session). + 1. The device from the database (with an ID) whose + ``components`` field contain the db version + of the passed-in components. + 2. A list of Add / Remove (not yet added to session). """ - blacklist = set() # Helper for execute_register() - db_device = cls.execute_register(device, blacklist, force_creation) - if id(device) != id(db_device): - # Did I get another device from db? - # In such case update the device from db with new stuff - cls.merge(device, db_device) - db_components = [] - for component in components: - db_component = cls.execute_register(component, blacklist, parent=db_device) - if id(component) != id(db_component): - cls.merge(component, db_component) + db_device, _ = cls.execute_register(device, force_creation=force_creation) + db_components, events = [], [] + if components is not None: # We have component info (see above) + blacklist = set() # type: Set[int] + not_new_components = set() + for component in components: + db_component, is_new = cls.execute_register(component, blacklist, parent=db_device) db_components.append(db_component) - events = tuple() - if components is not None: - # Only perform Add / Remove when - events = cls.add_remove(db_device, set(db_components)) - return db_device, db_components, events + if not is_new: + not_new_components.add(db_component) + # We only want to perform Add/Remove to not new components + events = cls.add_remove(db_device, not_new_components) + db_device.components = db_components + return db_device, events - @staticmethod - def execute_register(device: Device, - blacklist: Set[int], + @classmethod + def execute_register(cls, device: Device, + blacklist: Set[int] = None, force_creation: bool = False, - parent: Computer = None) -> Device: + parent: Computer = None) -> (Device, bool): """ Synchronizes one device to the DB. - This method tries to update the device in the database if it - already exists, otherwise it creates a new one. + This method tries to create a device into the database, and + if it already exists it returns a "local synced version", + this is the same ``device`` you passed-in but with updated + values from the database one (like the id value). + + When we say "local" we mean that if, the device existed on the + database, we do not "touch" any of its values on the DB. :param device: The device to synchronize to the DB. :param blacklist: A set of components already found by @@ -85,8 +88,11 @@ class Sync: S/N). :param parent: For components, the computer that contains them. Helper used by Component.similar_one(). - :return: A synchronized device with the DB. It can be a new - device or an already existing one. + :return: A tuple with: + 1. A synchronized device with the DB. It can be a new + device or an already existing one. + 2. A flag stating if the device is new or it existed + already in the DB. :raise NeedsId: The device has not any identifier we can use. To still create the device use ``force_creation``. @@ -103,56 +109,73 @@ class Sync: # ensure we don't get it again for another component # with the same physical properties blacklist.add(db_component.id) - return db_component + return cls.merge(device, db_component), False elif not force_creation: raise NeedsId() - db.session.begin_nested() # Create transaction savepoint to auto-rollback on insertion err try: - # Let's try to insert or update - db.session.insert(device) - db.session.flush() + with db.session.begin_nested(): + # Create transaction savepoint to auto-rollback on insertion err + # Let's try to insert or update + db.session.add(device) + db.session.flush() except IntegrityError as e: if e.orig.diag.sqlstate == UNIQUE_VIOLATION: + db.session.rollback() # This device already exists in the DB - field, value = 'az' # todo get from e.orig.diag - return Device.query.find(getattr(device.__class__, field) == value).one() + field, value = re.findall('\(.*?\)', e.orig.diag.message_detail) # type: str + field = field.replace('(', '').replace(')', '') + value = value.replace('(', '').replace(')', '') + db_device = Device.query.filter(getattr(device.__class__, field) == value).one() + return cls.merge(device, db_device), False else: raise e else: - return device # Our device is new + return device, True # Our device is new @classmethod def merge(cls, device: Device, db_device: Device): """ Copies the physical properties of the device to the db_device. """ - for field, value in device.physical_properties: + for field_name, value in device.physical_properties.items(): if value is not None: - setattr(db_device, field.name, value) + setattr(db_device, field_name, value) return db_device @classmethod def add_remove(cls, device: Device, - new_components: Set[Component]) -> List[Add or Remove]: + components: Set[Component]) -> List[Add or Remove]: """ - Generates the Add and Remove events by evaluating the - differences between the components the - :param device: - :param new_components: - :return: + Generates the Add and Remove events (but doesn't add them to + session). + + :param device: A device which ``components`` attribute contains + the old list of components. The components that + are not in ``components`` will be Removed. + :param components: List of components that are potentially to + be Added. Some of them can already exist + on the device, in which case they won't + be re-added. + :return: A list of Add / Remove events. """ + events = [] old_components = set(device.components) - add = Add(device=Device, components=list(new_components - old_components)) - events = [ - Remove(device=device, components=list(old_components - new_components)), - add - ] - # For the components we are adding, let's remove them from their old parents - def get_parent(component: Component): - return component.parent + adding = components - old_components + if adding: + add = Add(device=device, components=list(adding)) + + # For the components we are adding, let's remove them from their old parents + def g_parent(component: Component) -> int: + return component.parent or Computer(id=0) # Computer with id 0 is our Identity + + for parent, _components in groupby(sorted(add.components, key=g_parent), key=g_parent): + if parent.id != 0: + events.append(Remove(device=parent, components=list(_components))) + events.append(add) + + removing = old_components - components + if removing: + events.append(Remove(device=device, components=list(removing))) - for parent, components in groupby(sorted(add.components, key=get_parent), key=get_parent): - if parent is not None: - events.append(Remove(device=parent, components=list(components))) return events diff --git a/ereuse_devicehub/resources/event/models.py b/ereuse_devicehub/resources/event/models.py index a20c6ee7..3c05104c 100644 --- a/ereuse_devicehub/resources/event/models.py +++ b/ereuse_devicehub/resources/event/models.py @@ -1,6 +1,7 @@ from datetime import timedelta from colour import Color +from flask import g from sqlalchemy import BigInteger, Boolean, CheckConstraint, Column, DateTime, Enum as DBEnum, \ ForeignKey, Integer, Interval, JSON, Sequence, SmallInteger, Unicode from sqlalchemy.dialects.postgresql import UUID @@ -15,7 +16,8 @@ from ereuse_devicehub.resources.event.enums import Appearance, Bios, Functionali from ereuse_devicehub.resources.models import STR_BIG_SIZE, STR_SIZE, STR_SM_SIZE, Thing, \ check_range from ereuse_devicehub.resources.user.models import User -from teal.db import CASCADE, CASCADE_OWN, INHERIT_COND, POLYMORPHIC_ID, POLYMORPHIC_ON +from teal.db import CASCADE, CASCADE_OWN, INHERIT_COND, POLYMORPHIC_ID, POLYMORPHIC_ON, \ + StrictVersionType class JoinedTableMixin: @@ -40,7 +42,10 @@ class Event(Thing): backref=backref('events', lazy=True, cascade=CASCADE), primaryjoin='Event.snapshot_id == Snapshot.id') - author_id = Column(UUID(as_uuid=True), ForeignKey(User.id), nullable=False) + author_id = Column(UUID(as_uuid=True), + ForeignKey(User.id), + nullable=False, + default=lambda: g.user.id) author = relationship(User, backref=backref('events', lazy=True), primaryjoin=author_id == User.id) @@ -142,28 +147,26 @@ class Step(db.Model): class Snapshot(JoinedTableMixin, EventWithOneDevice): uuid = Column(UUID(as_uuid=True), nullable=False, unique=True) # type: UUID - version = Column(Unicode(STR_SM_SIZE), nullable=False) # type: str + version = Column(StrictVersionType(STR_SM_SIZE), nullable=False) # type: str software = Column(DBEnum(SoftwareType), nullable=False) # type: SoftwareType - appearance = Column(DBEnum(Appearance), nullable=False) # type: Appearance + appearance = Column(DBEnum(Appearance)) # type: Appearance appearance_score = Column(SmallInteger, - check_range('appearance_score', -3, 5), - nullable=False) # type: int - functionality = Column(DBEnum(Functionality), nullable=False) # type: Functionality + check_range('appearance_score', -3, 5)) # type: int + functionality = Column(DBEnum(Functionality)) # type: Functionality functionality_score = Column(SmallInteger, - check_range('functionality_score', min=-3, max=5), - nullable=False) # type: int + check_range('functionality_score', min=-3, max=5)) # type: int labelling = Column(Boolean) # type: bool bios = Column(DBEnum(Bios)) # type: Bios condition = Column(SmallInteger, - check_range('condition', min=0, max=5), - nullable=False) # type: int + check_range('condition', min=0, max=5)) # type: int elapsed = Column(Interval, nullable=False) # type: timedelta install_name = Column(Unicode(STR_BIG_SIZE)) # type: str install_elapsed = Column(Interval) # type: timedelta install_success = Column(Boolean) # type: bool inventory_elapsed = Column(Interval) # type: timedelta color = Column(ColorType) # type: Color - orientation = DBEnum(Orientation) # type: Orientation + orientation = Column(DBEnum(Orientation)) # type: Orientation + force_creation = Column(Boolean) @validates('components') def validate_components_only_workbench(self, _, components): diff --git a/ereuse_devicehub/resources/event/schemas.py b/ereuse_devicehub/resources/event/schemas.py index 9af523f9..aa93af61 100644 --- a/ereuse_devicehub/resources/event/schemas.py +++ b/ereuse_devicehub/resources/event/schemas.py @@ -1,5 +1,5 @@ from flask import current_app as app -from marshmallow import ValidationError, validates_schema +from marshmallow import ValidationError, post_load, validates_schema from marshmallow.fields import Boolean, DateTime, Integer, Nested, String, TimeDelta, UUID from marshmallow.validate import Length, Range from marshmallow_enum import EnumField @@ -137,6 +137,7 @@ class Snapshot(EventWithOneDevice): inventory = Nested(Inventory) color = Color(description='Main color of the device.') orientation = EnumField(Orientation, description='Is the device main stand wider or larger?') + force_creation = Boolean(data_key='forceCreation') @validates_schema def validate_workbench_version(self, data: dict): @@ -154,6 +155,14 @@ class Snapshot(EventWithOneDevice): raise ValidationError('Only Workbench can add component info', field_names=['components']) + @post_load + def normalize_nested(self, data: dict): + data.update(data.pop('condition')) + data['condition'] = data.pop('general', None) + data.update({'install_' + key: value for key, value in data.pop('install', {})}) + data['inventory_elapsed'] = data.get('inventory', {}).pop('elapsed', None) + return data + class Test(EventWithOneDevice): elapsed = TimeDelta(precision=TimeDelta.SECONDS, required=True) diff --git a/ereuse_devicehub/resources/event/views.py b/ereuse_devicehub/resources/event/views.py index 37420634..658b97ce 100644 --- a/ereuse_devicehub/resources/event/views.py +++ b/ereuse_devicehub/resources/event/views.py @@ -1,6 +1,6 @@ from distutils.version import StrictVersion -from flask import request +from flask import request, Response from ereuse_devicehub.db import db from ereuse_devicehub.resources.device.sync import Sync @@ -21,13 +21,19 @@ SUPPORTED_WORKBENCH = StrictVersion('11.0') class SnapshotView(View): def post(self): """Creates a Snapshot.""" - snapshot = Snapshot(**request.get_json()) # todo put this in schema.load()? + s = request.get_json() + # Note that if we set the device / components into the snapshot + # model object, when we flush them to the db we will flush + # snapshot, and we want to wait to flush snapshot at the end + device = s.pop('device') + components = s.pop('components') if s['software'] == SoftwareType.Workbench else None # noinspection PyArgumentList - c = snapshot.components if snapshot.software == SoftwareType.Workbench else None - snapshot.device, snapshot.components, snapshot.events = Sync.run(snapshot.device, c) + del s['type'] + snapshot = Snapshot(**s) + snapshot.device, snapshot.events = Sync.run(device, components, snapshot.force_creation) db.session.add(snapshot) # transform it back - return self.schema.jsonify(snapshot) + return Response(status=201) class TestHardDriveView(View): diff --git a/tests/conftest.py b/tests/conftest.py index 3fb0d069..a9dc1bb0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,4 +1,3 @@ -import json as stdlib_json from pathlib import Path import pytest @@ -41,6 +40,12 @@ def client(app: Devicehub) -> Client: return app.test_client() +@pytest.fixture() +def app_context(app: Devicehub): + with app.app_context(): + yield + + @pytest.fixture() def user(app: Devicehub) -> UserClient: """Gets a client with a logged-in dummy user.""" @@ -61,6 +66,20 @@ def create_user(email='foo@foo.com', password='foo') -> User: return user +@pytest.fixture() +def auth_app_context(app: Devicehub): + """Creates an app context with a set user.""" + with app.app_context(): + user = create_user() + + class Auth: # Mock + username = user.token + password = '' + + app.auth.perform_auth(Auth()) + yield + + def file(name: str) -> dict: """Opens and parses a JSON file from the ``files`` subdir.""" with Path(__file__).parent.joinpath('files').joinpath(name + '.yaml').open() as f: diff --git a/tests/files/pc-components.db.yaml b/tests/files/pc-components.db.yaml new file mode 100644 index 00000000..2f80937c --- /dev/null +++ b/tests/files/pc-components.db.yaml @@ -0,0 +1,14 @@ +device: + type: 'Microtower' + serial_number: 'd1s' + model: 'd1ml' + manufacturer: 'd1mr' +components: + - type: 'GraphicCard' + serial_number: 'gc1s' + model: 'gc1ml' + manufacturer: 'gc1mr' + - type: 'RamModule' + serial_number: 'rm1s' + model: 'rm1ml' + manufacturer: 'rm1mr' diff --git a/tests/test_device.py b/tests/test_device.py index 1e04f596..bf4229d3 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -1,7 +1,15 @@ +import pytest + from ereuse_devicehub.db import db from ereuse_devicehub.devicehub import Devicehub -from ereuse_devicehub.resources.device.models import Desktop, Device, GraphicCard, NetworkAdapter +from ereuse_devicehub.resources.device.exceptions import NeedsId +from ereuse_devicehub.resources.device.models import Component, Computer, Desktop, Device, \ + GraphicCard, Motherboard, NetworkAdapter from ereuse_devicehub.resources.device.schemas import Device as DeviceS +from ereuse_devicehub.resources.device.sync import Sync +from ereuse_devicehub.resources.event.models import Add, Remove +from teal.db import ResourceNotFound +from tests.conftest import file def test_device_model(app: Devicehub): @@ -47,6 +55,120 @@ def test_device_model(app: Devicehub): def test_device_schema(): """Ensures the user does not upload non-writable or extra fields.""" device_s = DeviceS() - device_s.load({'serial_number': 'foo1', 'model': 'foo', 'manufacturer': 'bar2'}) - + device_s.load({'serialNumber': 'foo1', 'model': 'foo', 'manufacturer': 'bar2'}) device_s.dump({'id': 1}) + + +@pytest.mark.usefixtures('app_context') +def test_physical_properties(): + c = Motherboard(slots=2, + usb=3, + serial_number='sn', + model='ml', + manufacturer='mr', + width=2.0, + pid='abc') + pc = Computer(components=[c]) + db.session.add(pc) + db.session.commit() + assert c.physical_properties == { + 'gid': None, + 'usb': 3, + 'pid': 'abc', + 'serial_number': 'sn', + 'pcmcia': None, + 'model': 'ml', + 'slots': 2, + 'serial': None, + 'firewire': None, + 'manufacturer': 'mr', + 'weight': None, + 'height': None, + 'width': 2.0 + } + + +@pytest.mark.usefixtures('app_context') +def test_component_similar_one(): + snapshot = file('pc-components.db') + d = snapshot['device'] + snapshot['components'][0]['serial_number'] = snapshot['components'][1]['serial_number'] = None + pc = Computer(**d, components=[Component(**c) for c in snapshot['components']]) + component1, component2 = pc.components # type: Component + db.session.add(pc) + # Let's create a new component named 'A' similar to 1 + componentA = Component(model=component1.model, manufacturer=component1.manufacturer) + similar_to_a = componentA.similar_one(pc, set()) + assert similar_to_a == component1 + # Component B does not have the same model + componentB = Component(model='nope', manufacturer=component1.manufacturer) + with pytest.raises(ResourceNotFound): + assert componentB.similar_one(pc, set()) + # If we blacklist component A we won't get anything + with pytest.raises(ResourceNotFound): + assert componentA.similar_one(pc, blacklist={componentA.id}) + + +@pytest.mark.usefixtures('auth_app_context') +def test_add_remove(): + # Original state: + # pc has c1 and c2 + # pc2 has c3 + # c4 is not with any pc + values = file('pc-components.db') + pc = values['device'] + c1, c2 = [Component(**c) for c in values['components']] + pc = Computer(**pc, components=[c1, c2]) + db.session.add(pc) + c3 = Component(serial_number='nc1') + pc2 = Computer(serial_number='s2', components=[c3]) + c4 = Component(serial_number='c4s') + db.session.add(pc2) + db.session.add(c4) + db.session.commit() + + # Test: + # pc has only c3 + events = Sync.add_remove(device=pc, components={c3, c4}) + assert len(events) == 3 + assert isinstance(events[0], Remove) + assert events[0].device == pc2 + assert events[0].components == [c3] + assert isinstance(events[1], Add) + assert events[1].device == pc + assert set(events[1].components) == {c3, c4} + assert isinstance(events[2], Remove) + assert events[2].device == pc + assert set(events[2].components) == {c1, c2} + + +@pytest.mark.usefixtures('app_context') +def test_execute_register_computer(): + # Case 1: device does not exist on DB + pc = Computer(**file('pc-components.db')['device']) + db_pc, _ = Sync.execute_register(pc, set()) + assert pc.physical_properties == db_pc.physical_properties + + +@pytest.mark.usefixtures('app_context') +def test_execute_register_computer_existing(): + pc = Computer(**file('pc-components.db')['device']) + db.session.add(pc) + db.session.commit() # We need two separate sessions + pc = Computer(**file('pc-components.db')['device']) + # 1: device exists on DB + db_pc, _ = Sync.execute_register(pc, set()) + assert pc.physical_properties == db_pc.physical_properties + + +@pytest.mark.usefixtures('app_context') +def test_execute_register_computer_no_hid(): + pc = Computer(**file('pc-components.db')['device']) + # 1: device has no HID + pc.hid = pc.model = None + with pytest.raises(NeedsId): + Sync.execute_register(pc, set()) + + # 2: device has no HID and we force it + db_pc, _ = Sync.execute_register(pc, set(), force_creation=True) + assert pc.physical_properties == db_pc.physical_properties diff --git a/tests/test_event.py b/tests/test_event.py index e69de29b..29f53cc2 100644 --- a/tests/test_event.py +++ b/tests/test_event.py @@ -0,0 +1,24 @@ +import pytest +from flask import g + +from ereuse_devicehub.db import db +from ereuse_devicehub.resources.device.models import Device +from ereuse_devicehub.resources.event.models import EventWithOneDevice +from tests.conftest import create_user + + +@pytest.mark.usefixtures('app_context') +def test_author(): + """ + Checks the default created author. + + Note that the author can be accessed after inserting the row. + """ + user = create_user() + g.user = user + e = EventWithOneDevice(device=Device()) + db.session.add(e) + assert e.author is None + assert e.author_id is None + db.session.commit() + assert e.author == user diff --git a/tests/test_snapshot.py b/tests/test_snapshot.py index dff45be3..1f584de9 100644 --- a/tests/test_snapshot.py +++ b/tests/test_snapshot.py @@ -1,6 +1,8 @@ from datetime import datetime, timedelta from uuid import uuid4 +import pytest + from ereuse_devicehub.client import UserClient from ereuse_devicehub.db import db from ereuse_devicehub.devicehub import Devicehub @@ -11,42 +13,40 @@ from ereuse_devicehub.resources.user.models import User from tests.conftest import file -def test_snapshot_model(app: Devicehub): +@pytest.mark.usefixtures('auth_app_context') +def test_snapshot_model(): """ Tests creating a Snapshot with its relationships ensuring correct DB mapping. """ - with app.app_context(): - user = User(email='foo@bar.com') - device = Microtower(serial_number='a1') - # noinspection PyArgumentList - snapshot = Snapshot(uuid=uuid4(), - date=datetime.now(), - version='1.0', - software=SoftwareType.DesktopApp, - appearance=Appearance.A, - appearance_score=5, - functionality=Functionality.A, - functionality_score=5, - labelling=False, - bios=Bios.C, - condition=5, - elapsed=timedelta(seconds=25)) - snapshot.device = device - snapshot.author = user - snapshot.request = SnapshotRequest(request={'foo': 'bar'}) + device = Microtower(serial_number='a1') + # noinspection PyArgumentList + snapshot = Snapshot(uuid=uuid4(), + date=datetime.now(), + version='1.0', + software=SoftwareType.DesktopApp, + appearance=Appearance.A, + appearance_score=5, + functionality=Functionality.A, + functionality_score=5, + labelling=False, + bios=Bios.C, + condition=5, + elapsed=timedelta(seconds=25)) + snapshot.device = device + snapshot.request = SnapshotRequest(request={'foo': 'bar'}) - db.session.add(snapshot) - db.session.commit() - device = Microtower.query.one() # type: Microtower - assert device.events_one[0].type == Snapshot.__name__ - db.session.delete(device) - db.session.commit() - assert Snapshot.query.one_or_none() is None - assert SnapshotRequest.query.one_or_none() is None - assert User.query.one() is not None - assert Microtower.query.one_or_none() is None - assert Device.query.one_or_none() is None + db.session.add(snapshot) + db.session.commit() + device = Microtower.query.one() # type: Microtower + assert device.events_one[0].type == Snapshot.__name__ + db.session.delete(device) + db.session.commit() + assert Snapshot.query.one_or_none() is None + assert SnapshotRequest.query.one_or_none() is None + assert User.query.one() is not None + assert Microtower.query.one_or_none() is None + assert Device.query.one_or_none() is None def test_snapshot_schema(app: Devicehub):