First Snapshot attempt
This commit is contained in:
parent
8723b379b0
commit
78b5a230d4
|
@ -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)
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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 = []
|
||||
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 = cls.execute_register(component, blacklist, parent=db_device)
|
||||
if id(component) != id(db_component):
|
||||
cls.merge(component, db_component)
|
||||
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
|
||||
: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:
|
||||
with db.session.begin_nested():
|
||||
# Create transaction savepoint to auto-rollback on insertion err
|
||||
# Let's try to insert or update
|
||||
db.session.insert(device)
|
||||
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
|
||||
]
|
||||
|
||||
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 get_parent(component: Component):
|
||||
return component.parent
|
||||
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
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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)
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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:
|
||||
|
|
14
tests/files/pc-components.db.yaml
Normal file
14
tests/files/pc-components.db.yaml
Normal file
|
@ -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'
|
|
@ -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
|
||||
|
|
|
@ -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
|
|
@ -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,13 +13,12 @@ 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(),
|
||||
|
@ -33,7 +34,6 @@ def test_snapshot_model(app: Devicehub):
|
|||
condition=5,
|
||||
elapsed=timedelta(seconds=25))
|
||||
snapshot.device = device
|
||||
snapshot.author = user
|
||||
snapshot.request = SnapshotRequest(request={'foo': 'bar'})
|
||||
|
||||
db.session.add(snapshot)
|
||||
|
|
Reference in a new issue