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

Simplify Snapshot; improve error-handling in tests

This commit is contained in:
Xavier Bustamante Talavera 2018-05-16 15:23:48 +02:00
parent 439f7b9d58
commit 4ab7d421a4
13 changed files with 104 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
ereuse_devicehub/client.py
View File

@ -7,30 +7,57 @@ from werkzeug.exceptions import HTTPException
from ereuse_devicehub.resources.models import Thing
from teal.client import Client as TealClient
from teal.marshmallow import ValidationError
class Client(TealClient):
def __init__(self, application, response_wrapper=None, use_cookies=False,
"""A client suited for Devicehub main usage."""
def __init__(self, application,
response_wrapper=None,
use_cookies=False,
allow_subdomain_redirects=False):
super().__init__(application, response_wrapper, use_cookies, allow_subdomain_redirects)
def open(self, uri: str, res: str or Type[Thing] = None, status: int or HTTPException = 200,
query: dict = {}, accept=JSON, content_type=JSON, item=None, headers: dict = None,
token: str = None, **kw) -> (dict or str, Response):
def open(self,
uri: str,
res: str or Type[Thing] = None,
status: Union[int, Type[HTTPException], Type[ValidationError]] = 200,
query: dict = {},
accept=JSON,
content_type=JSON,
item=None,
headers: dict = None,
token: str = None,
**kw) -> (dict or str, Response):
if issubclass(res, Thing):
res = res.__name__
return super().open(uri, res, status, query, accept, content_type, item, headers, token,
**kw)
def get(self, uri: str = '', res: Union[Type[Thing], str] = None, query: dict = {},
status: int or HTTPException = 200, item: Union[int, str] = None, accept: str = JSON,
headers: dict = None, token: str = None, **kw) -> (dict or str, Response):
def get(self,
uri: str = '',
res: Union[Type[Thing], str] = None,
query: dict = {},
status: Union[int, Type[HTTPException], Type[ValidationError]] = 200,
item: Union[int, str] = None,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw) -> (dict or str, Response):
return super().get(uri, res, query, status, item, accept, headers, token, **kw)
def post(self, data: str or dict, uri: str = '', res: Union[Type[Thing], str] = None,
query: dict = {}, status: int or HTTPException = 201, content_type: str = JSON,
accept: str = JSON, headers: dict = None, token: str = None, **kw) -> (
dict or str, Response):
def post(self,
data: str or dict,
uri: str = '',
res: Union[Type[Thing], str] = None,
query: dict = {},
status: Union[int, Type[HTTPException], Type[ValidationError]] = 201,
content_type: str = JSON,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw) -> (dict or str, Response):
return super().post(data, uri, res, query, status, content_type, accept, headers, token,
**kw)
@ -58,8 +85,16 @@ class UserClient(Client):
self.password = password # type: str
self.user = None # type: dict
def open(self, uri: str, res: str = None, status: int or HTTPException = 200, query: dict = {},
accept=JSON, content_type=JSON, item=None, headers: dict = None, token: str = None,
def open(self,
uri: str,
res: str = None,
status: int or HTTPException = 200,
query: dict = {},
accept=JSON,
content_type=JSON,
item=None,
headers: dict = None,
token: str = None,
**kw) -> (dict or str, Response):
return super().open(uri, res, status, query, accept, content_type, item, headers,
self.user['token'] if self.user else token, **kw)

12
ereuse_devicehub/config.py
View File

@ -3,18 +3,18 @@ from distutils.version import StrictVersion
from ereuse_devicehub.resources.device import ComponentDef, ComputerDef, DesktopDef, DeviceDef, \
GraphicCardDef, HardDriveDef, LaptopDef, MicrotowerDef, MotherboardDef, NetbookDef, \
NetworkAdapterDef, ProcessorDef, RamModuleDef, ServerDef
from ereuse_devicehub.resources.event import EventDef, SnapshotDef, TestDef, TestHardDriveDef, \
AddDef, RemoveDef
from ereuse_devicehub.resources.event import AddDef, EventDef, RemoveDef, SnapshotDef, TestDef, \
TestHardDriveDef
from ereuse_devicehub.resources.user import UserDef
from teal.config import Config
class DevicehubConfig(Config):
RESOURCE_DEFINITIONS = (
DeviceDef, ComputerDef, DesktopDef, LaptopDef, NetbookDef, ServerDef, MicrotowerDef,
ComponentDef, GraphicCardDef, HardDriveDef, MotherboardDef, NetworkAdapterDef,
RamModuleDef, ProcessorDef, UserDef, EventDef, AddDef, RemoveDef, SnapshotDef,
TestDef, TestHardDriveDef
DeviceDef, ComputerDef, DesktopDef, LaptopDef, NetbookDef, ServerDef,
MicrotowerDef, ComponentDef, GraphicCardDef, HardDriveDef, MotherboardDef,
NetworkAdapterDef, RamModuleDef, ProcessorDef, UserDef, EventDef, AddDef, RemoveDef,
SnapshotDef, TestDef, TestHardDriveDef
)
PASSWORD_SCHEMES = {'pbkdf2_sha256'}
SQLALCHEMY_DATABASE_URI = 'postgresql://localhost/dh-db1'

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

@ -1,4 +1,4 @@
from marshmallow import ValidationError
from teal.marshmallow import ValidationError
class MismatchBetweenIds(ValidationError):

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

@ -7,8 +7,8 @@ from ereuse_devicehub.resources.schemas import Thing, UnitCodes
class Device(Thing):
id = Integer(dump_only=True,
description='The identifier of the device for this database.')
# todo id is dump_only except when in Snapshot
id = Integer(description='The identifier of the device for this database.')
hid = Str(dump_only=True,
description='The Hardware ID is the unique ID traceability systems '
'use to ID a device globally.')

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

@ -18,8 +18,7 @@ class Sync:
@classmethod
def run(cls, device: Device,
components: Iterable[Component] or None,
force_creation: bool = False) -> (Device, List[Add or Remove]):
components: Iterable[Component] or None) -> (Device, List[Add or Remove]):
"""
Synchronizes the device and components with the database.
@ -37,16 +36,13 @@ class Sync:
no info about components and the already
existing components of the device (in case
the device already exists) won't be touch.
:param force_creation: Shall we create the device even if
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) whose
``components`` field contain the db version
of the passed-in components.
2. A list of Add / Remove (not yet added to session).
"""
db_device, _ = cls.execute_register(device, force_creation=force_creation)
db_device, _ = cls.execute_register(device)
db_components, events = [], []
if components is not None: # We have component info (see above)
blacklist = set() # type: Set[int]
@ -64,7 +60,6 @@ class Sync:
@classmethod
def execute_register(cls, device: Device,
blacklist: Set[int] = None,
force_creation: bool = False,
parent: Computer = None) -> (Device, bool):
"""
Synchronizes one device to the DB.
@ -80,12 +75,6 @@ class Sync:
:param device: The device to synchronize to the DB.
:param blacklist: A set of components already found by
Component.similar_one(). Pass-in an empty Set.
:param force_creation: Allow creating a device even if it
doesn't generate HID or doesn't have an
ID. Only valid for non-components.
Usually used when creating non-branded
custom computers (as they don't have
S/N).
:param parent: For components, the computer that contains them.
Helper used by Component.similar_one().
:return: A tuple with:
@ -110,7 +99,7 @@ class Sync:
# with the same physical properties
blacklist.add(db_component.id)
return cls.merge(device, db_component), False
elif not force_creation:
else:
raise NeedsId()
try:
with db.session.begin_nested():
@ -122,10 +111,11 @@ class Sync:
if e.orig.diag.sqlstate == UNIQUE_VIOLATION:
db.session.rollback()
# This device already exists in the DB
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()
field, value = (
x.replace('(', '').replace(')', '')
for x in re.findall('\(.*?\)', e.orig.diag.message_detail)
)
db_device = Device.query.filter_by(**{field: value}).one() # type: Device
return cls.merge(device, db_device), False
else:
raise e

1
ereuse_devicehub/resources/event/models.py
View File

@ -180,7 +180,6 @@ class Snapshot(JoinedTableMixin, EventWithOneDevice):
inventory_elapsed = Column(Interval) # type: timedelta
color = Column(ColorType) # type: Color
orientation = Column(DBEnum(Orientation)) # type: Orientation
force_creation = Column(Boolean)
@validates('components')
def validate_components_only_workbench(self, _, components):

1
ereuse_devicehub/resources/event/schemas.py
View File

@ -148,7 +148,6 @@ 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')
events = NestedOn(Event, many=True, dump_only=True)
@validates_schema

2
ereuse_devicehub/resources/event/views.py
View File

@ -34,7 +34,7 @@ class SnapshotView(View):
components = s.pop('components') if s['software'] == SoftwareType.Workbench else None
# noinspection PyArgumentList
snapshot = Snapshot(**s)
snapshot.device, snapshot.events = Sync.run(device, components, snapshot.force_creation)
snapshot.device, snapshot.events = Sync.run(device, components)
snapshot.components = snapshot.device.components
# commit will change the order of the components by what
# the DB wants. Let's get a copy of the list so we preserve

2
tests/test_auth.py
View File

@ -3,7 +3,7 @@ from uuid import uuid4
import pytest
from werkzeug.exceptions import Unauthorized
from ereuse_devicehub.client import UserClient, Client
from ereuse_devicehub.client import Client, UserClient
from ereuse_devicehub.devicehub import Devicehub
from tests.conftest import create_user

4
tests/test_device.py
View File

@ -171,10 +171,6 @@ def test_execute_register_computer_no_hid():
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
def test_get_device(app: Devicehub, user: UserClient):
"""Checks GETting a Desktop with its components."""

6
tests/test_schema.py
View File

@ -1,6 +0,0 @@
from flask_sqlalchemy import SQLAlchemy
def test_device_schema():
"""Tests device schema."""
pass

29
tests/test_snapshot.py
View File

@ -7,6 +7,7 @@ import pytest
from ereuse_devicehub.client import UserClient
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.device.exceptions import NeedsId
from ereuse_devicehub.resources.device.models import Device, Microtower
from ereuse_devicehub.resources.event.models import Appearance, Bios, Event, Functionality, \
Snapshot, SnapshotRequest, SoftwareType
@ -120,7 +121,12 @@ def test_snapshot_post(user: UserClient):
assert 'author' not in snapshot['device']
def test_snapshot_add_remove(user: UserClient):
def test_snapshot_component_add_remove(user: UserClient):
"""
Tests adding and removing components and some don't generate HID.
All computers generate HID.
"""
def get_events_info(events: List[dict]) -> tuple:
return tuple(
(
@ -222,3 +228,24 @@ def test_snapshot_add_remove(user: UserClient):
# We haven't changed PC2
assert tuple(c['serialNumber'] for c in pc2['components']) == ('p2c1s',)
assert all(c['parent'] == pc2_id for c in pc2['components'])
def _test_snapshot_computer_no_hid(user: UserClient):
"""
Tests inserting a computer that doesn't generate a HID, neither
some of its components.
"""
# PC with 2 components. PC doesn't have HID and neither 1st component
s = file('basic.snapshot')
del s['device']['model']
del s['components'][0]['model']
user.post(s, res=Snapshot, status=NeedsId)
# The system tells us that it could not register the device because
# the device (computer) cannot generate a HID.
# In such case we need to specify an ``id`` so the system can
# recognize the device. The ``id`` can reference to the same
# device, it already existed in the DB, or to a placeholder,
# if the device is new in the DB.
user.post(s, res=Device)
s['device']['id'] = 1 # Assign the ID of the placeholder
user.post(s, res=Snapshot)

10
tests/test_user.py
View File

@ -2,13 +2,15 @@ from base64 import b64decode
from uuid import UUID
from sqlalchemy_utils import Password
from werkzeug.exceptions import NotFound, Unauthorized, UnprocessableEntity
from werkzeug.exceptions import NotFound
from ereuse_devicehub.client import Client
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.user import UserDef
from ereuse_devicehub.resources.user.exceptions import WrongCredentials
from ereuse_devicehub.resources.user.models import User
from teal.marshmallow import ValidationError
from tests.conftest import create_user
@ -72,11 +74,11 @@ def test_login_failure(client: Client, app: Devicehub):
create_user()
client.post({'email': 'foo@foo.com', 'password': 'wrong pass'},
uri='/users/login',
status=Unauthorized)
status=WrongCredentials)
# Wrong URI
client.post({}, uri='/wrong-uri', status=NotFound)
# Malformed data
client.post({}, uri='/users/login', status=UnprocessableEntity)
client.post({}, uri='/users/login', status=ValidationError)
client.post({'email': 'this is not an email', 'password': 'nope'},
uri='/users/login',
status=UnprocessableEntity)
status=ValidationError)