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

Fix UUID in events

This commit is contained in:
Xavier Bustamante Talavera 2018-06-15 15:31:03 +02:00
parent e3a887d9fa
commit 5538e5ac69
7 changed files with 25 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
ereuse_devicehub/client.py
View File

@ -68,15 +68,12 @@ class Client(TealClient):
def get_many(self,
res: Union[Type[Union[models.Thing, schemas.Thing]], str],
resources: Iterable[dict],
resources: Iterable[Union[dict, int]],
key: str = None,
headers: dict = None,
token: str = None,
accept: str = JSON,
**kw) -> Iterable[Union[Dict[str, Any], str]]:
"""Like :meth:`.get` but with many resources."""
return (
self.get(res=res, item=r['key'] if key else r, headers=headers, token=token, **kw)[0]
self.get(res=res, item=r[key] if key else r, **kw)[0]
for r in resources
)

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

@ -37,10 +37,10 @@ class Device(Thing):
"""
return sorted(chain(self.events_multiple, self.events_one), key=attrgetter('created'))
def __init__(self, *args, **kw) -> None:
super().__init__(*args, **kw)
def __init__(self, **kw) -> None:
super().__init__(**kw)
with suppress(TypeError):
self.hid = Naming.hid(self.manufacturer, self.serial_number, self.model) # type: str
self.hid = Naming.hid(self.manufacturer, self.serial_number, self.model)
@property
def physical_properties(self) -> Dict[str, object or None]:
@ -114,7 +114,7 @@ class Component(Device):
cascade=CASCADE,
order_by=lambda: Component.id,
collection_class=OrderedSet),
primaryjoin=parent_id == Computer.id) # type: Device
primaryjoin=parent_id == Computer.id)
def similar_one(self, parent: Computer, blacklist: Set[int]) -> 'Component':
"""
@ -137,7 +137,7 @@ class Component(Device):
@property
def events(self) -> list:
return sorted(chain(super().events, self.events_components), key=attrgetter('id'))
return sorted(chain(super().events, self.events_components), key=attrgetter('created'))
class JoinedComponentTableMixin:

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

@ -12,7 +12,7 @@ class EventDef(Resource):
SCHEMA = Event
VIEW = EventView
AUTH = True
ID_CONVERTER = Converters.int
ID_CONVERTER = Converters.uuid
class AddDef(EventDef):

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

@ -17,7 +17,7 @@ from teal.resource import Schema
class Event(Thing):
id = Integer(dump_only=True)
id = UUID(dump_only=True)
name = String(default='', validate=Length(STR_BIG_SIZE), description=m.Event.name.comment)
date = DateTime('iso', description=m.Event.date.comment)
error = Boolean(default=False, description=m.Event.error.comment)

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

@ -1,4 +1,5 @@
from distutils.version import StrictVersion
from uuid import UUID
from flask import request
from sqlalchemy.util import OrderedSet
@ -11,7 +12,7 @@ from teal.resource import View
class EventView(View):
def one(self, id: int):
def one(self, id: UUID):
"""Gets one event."""
event = Event.query.filter_by(id=id).one()
return self.schema.jsonify(event)

4
ereuse_devicehub/resources/user/__init__.py
View File

@ -13,7 +13,7 @@ from teal.resource import Converters, Resource
class UserDef(Resource):
SCHEMA = UserS
VIEW = UserView
ID_CONVERTER = Converters.uid
ID_CONVERTER = Converters.uuid
AUTH = True
def __init__(self, app: 'devicehub.Devicehub', import_name=__package__, static_folder=None,
@ -40,7 +40,7 @@ class UserDef(Resource):
class OrganizationDef(Resource):
__type__ = 'Organization'
ID_CONVERTER = Converters.uid
ID_CONVERTER = Converters.uuid
AUTH = True
def __init__(self, app, import_name=__package__, static_folder=None, static_url_path=None,

26
tests/test_snapshot.py
View File

@ -149,10 +149,8 @@ def test_snapshot_component_add_remove(user: UserClient):
def get_events_info(events: List[dict]) -> tuple:
return tuple(
(
e['id'],
e['type'],
[c['serialNumber'] for c in e['components']],
e.get('snapshot', {}).get('id', None)
[c['serialNumber'] for c in e['components']]
)
for e in user.get_many(res=Event, resources=events, key='id')
)
@ -209,11 +207,11 @@ def test_snapshot_component_add_remove(user: UserClient):
# PC1
assert {c['serialNumber'] for c in pc1['components']} == {'p1c2s', 'p1c3s'}
assert all(c['parent'] == pc1_id for c in pc1['components'])
assert get_events_info(pc1['events']) == (
assert tuple(get_events_info(pc1['events'])) == (
# id, type, components, snapshot
(1, 'Snapshot', ['p1c1s', 'p1c2s', 'p1c3s'], None), # first Snapshot1
(3, 'Remove', ['p1c2s'], 2), # Remove Processor in Snapshot2
(4, 'Snapshot', ['p1c2s', 'p1c3s'], None) # This Snapshot3
('Snapshot', ['p1c1s', 'p1c2s', 'p1c3s']), # first Snapshot1
('Remove', ['p1c2s']), # Remove Processor in Snapshot2
('Snapshot', ['p1c2s', 'p1c3s']) # This Snapshot3
)
# PC2
assert tuple(c['serialNumber'] for c in pc2['components']) == ('p2c1s',)
@ -224,12 +222,12 @@ def test_snapshot_component_add_remove(user: UserClient):
)
# p1c2s has Snapshot, Remove and Add
p1c2s, _ = user.get(res=Device, item=pc1['components'][0]['id'])
assert get_events_info(p1c2s['events']) == (
(1, 'Snapshot', ['p1c1s', 'p1c2s', 'p1c3s'], None), # First Snapshot to PC1
(2, 'Snapshot', ['p1c2s', 'p2c1s'], None), # Second Snapshot to PC2
(3, 'Remove', ['p1c2s'], 2), # ...which caused p1c2s to be removed form PC1
(4, 'Snapshot', ['p1c2s', 'p1c3s'], None), # The third Snapshot to PC1
(5, 'Remove', ['p1c2s'], 4) # ...which caused p1c2 to be removed from PC2
assert tuple(get_events_info(p1c2s['events'])) == (
('Snapshot', ['p1c1s', 'p1c2s', 'p1c3s']), # First Snapshot to PC1
('Snapshot', ['p1c2s', 'p2c1s']), # Second Snapshot to PC2
('Remove', ['p1c2s']), # ...which caused p1c2s to be removed form PC1
('Snapshot', ['p1c2s', 'p1c3s']), # The third Snapshot to PC1
('Remove', ['p1c2s']) # ...which caused p1c2 to be removed from PC2
)
# We register the first device but without the processor,
@ -242,7 +240,7 @@ def test_snapshot_component_add_remove(user: UserClient):
assert {c['serialNumber'] for c in pc1['components']} == {'p1c3s', 'p1c4s'}
assert all(c['parent'] == pc1_id for c in pc1['components'])
# This last Snapshot only
assert get_events_info(pc1['events'])[-1] == (6, 'Snapshot', ['p1c3s', 'p1c4s'], None)
assert get_events_info(pc1['events'])[-1] == ('Snapshot', ['p1c3s', 'p1c4s'])
# PC2
# We haven't changed PC2
assert tuple(c['serialNumber'] for c in pc2['components']) == ('p2c1s',)