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

Add Device problems, privacy, working; Add event ErasePhysical

Browse source
This commit is contained in:
Xavier Bustamante Talavera 2018-11-09 11:22:13 +01:00
parent e009bf4bc1
commit bd0eb3aad3
9 changed files with 158 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
docs/states.puml
View file

@ -6,13 +6,15 @@ skinparam ranksep 1
[*] -> Registered
state Attributes {
state Broken : cannot turn on
state Owners
state Usufructuarees
state Reservees
state "Physical\nPossessor"
state "Waste\n\Product"
state problems : List of current events \nwith Warn/Error
state privacy : Set of\ncurrent erasures
state working : List of current events\naffecting working
}
state Physical {
@ -44,10 +46,4 @@ state Trading {
Renting --> Cancelled : Cancel
}
state DataStoragePrivacyCompliance {
state Erased
state Destroyed
}
@enduml

76
ereuse_devicehub/resources/device/models.py
View file

@ -8,6 +8,7 @@ from typing import Dict, List, Set
from boltons import urlutils
from citext import CIText
from ereuse_utils.naming import Naming
from more_itertools import unique_everseen
from sqlalchemy import BigInteger, Boolean, Column, Enum as DBEnum, Float, ForeignKey, Integer, \
Sequence, SmallInteger, Unicode, inspect, text
from sqlalchemy.ext.declarative import declared_attr
@ -22,8 +23,8 @@ from teal.marshmallow import ValidationError
from teal.resource import url_for_resource
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.enums import ComputerChassis, DataStorageInterface, \
DataStoragePrivacyCompliance, DisplayTech, PrinterTechnology, RamFormat, RamInterface
from ereuse_devicehub.resources.enums import ComputerChassis, DataStorageInterface, DisplayTech, \
PrinterTechnology, RamFormat, RamInterface, Severity
from ereuse_devicehub.resources.models import STR_SM_SIZE, Thing
@ -31,6 +32,7 @@ class Device(Thing):
"""
Base class for any type of physical object that can be identified.
"""
EVENT_SORT_KEY = attrgetter('created')
id = Column(BigInteger, Sequence('device_seq'), primary_key=True)
id.comment = """
@ -77,6 +79,11 @@ class Device(Thing):
'color'
}
def __init__(self, **kw) -> None:
super().__init__(**kw)
with suppress(TypeError):
self.hid = Naming.hid(self.manufacturer, self.serial_number, self.model)
@property
def events(self) -> list:
"""
@ -86,12 +93,25 @@ class Device(Thing):
Events are returned by ascending creation time.
"""
return sorted(chain(self.events_multiple, self.events_one), key=attrgetter('created'))
return sorted(chain(self.events_multiple, self.events_one), key=self.EVENT_SORT_KEY)
def __init__(self, **kw) -> None:
super().__init__(**kw)
with suppress(TypeError):
self.hid = Naming.hid(self.manufacturer, self.serial_number, self.model)
@property
def problems(self):
"""Current events with severity.Warning or higher.
There can be up to 3 events: current Snapshot,
current Physical event, current Trading event.
"""
from ereuse_devicehub.resources.device import states
from ereuse_devicehub.resources.event.models import Snapshot
events = set()
with suppress(LookupError, ValueError):
events.add(self.last_event_of(Snapshot))
with suppress(LookupError, ValueError):
events.add(self.last_event_of(*states.Physical.events()))
with suppress(LookupError, ValueError):
events.add(self.last_event_of(*states.Trading.events()))
return self._warning_events(events)
@property
def physical_properties(self) -> Dict[str, object or None]:
@ -164,6 +184,20 @@ class Device(Thing):
event = self.last_event_of(Receive)
return event.agent
@property
def working(self):
"""A list of the current tests with warning or errors. A
device is working if the list is empty.
This property returns, for the last test performed of each type,
the one with the worst severity of them, or None if no
test has been executed.
"""
from ereuse_devicehub.resources.event.models import Test
current_tests = unique_everseen((e for e in reversed(self.events) if isinstance(e, Test)),
key=attrgetter('type')) # last test of each type
return self._warning_events(current_tests)
@declared_attr
def __mapper_args__(cls):
"""
@ -188,6 +222,10 @@ class Device(Thing):
except StopIteration:
raise LookupError('{!r} does not contain events of types {}.'.format(self, types))
def _warning_events(self, events):
return sorted((ev for ev in events if ev.severity >= Severity.Warning),
key=self.EVENT_SORT_KEY)
def __lt__(self, other):
return self.id < other.id
@ -255,7 +293,7 @@ class Computer(Device):
@property
def events(self) -> list:
return sorted(chain(super().events, self.events_parent), key=attrgetter('created'))
return sorted(chain(super().events, self.events_parent), key=self.EVENT_SORT_KEY)
@property
def ram_size(self) -> int:
@ -294,6 +332,17 @@ class Computer(Device):
speeds[net.wireless] = max(net.speed or 0, speeds[net.wireless] or 0)
return speeds
@property
def privacy(self):
"""Returns the privacy of all DataStorage components when
it is None.
"""
return set(
privacy for privacy in
(hdd.privacy for hdd in self.components if isinstance(hdd, DataStorage))
if privacy
)
def __format__(self, format_spec):
if not format_spec:
return super().__format__(format_spec)
@ -405,7 +454,7 @@ class Component(Device):
@property
def events(self) -> list:
return sorted(chain(super().events, self.events_components), key=attrgetter('created'))
return sorted(chain(super().events, self.events_components), key=self.EVENT_SORT_KEY)
class JoinedComponentTableMixin:
@ -431,11 +480,12 @@ class DataStorage(JoinedComponentTableMixin, Component):
@property
def privacy(self):
"""Returns the privacy compliance state of the data storage."""
# todo add physical destruction event
from ereuse_devicehub.resources.event.models import EraseBasic
with suppress(LookupError):
erase = self.last_event_of(EraseBasic)
return DataStoragePrivacyCompliance.from_erase(erase)
try:
ev = self.last_event_of(EraseBasic)
except LookupError:
ev = None
return ev
def __format__(self, format_spec):
v = super().__format__(format_spec)

41
ereuse_devicehub/resources/device/models.pyi
View file

@ -1,5 +1,6 @@
from datetime import datetime
from typing import Dict, List, Set, Type, Union
from operator import attrgetter
from typing import Dict, Generator, Iterable, List, Optional, Set, Type
from boltons import urlutils
from boltons.urlutils import URL
@ -11,8 +12,8 @@ from teal.enums import Layouts
from ereuse_devicehub.resources.agent.models import Agent
from ereuse_devicehub.resources.device import states
from ereuse_devicehub.resources.enums import ComputerChassis, DataStorageInterface, \
DataStoragePrivacyCompliance, DisplayTech, PrinterTechnology, RamFormat, RamInterface
from ereuse_devicehub.resources.enums import ComputerChassis, DataStorageInterface, DisplayTech, \
PrinterTechnology, RamFormat, RamInterface
from ereuse_devicehub.resources.event import models as e
from ereuse_devicehub.resources.image.models import ImageList
from ereuse_devicehub.resources.lot.models import Lot
@ -21,6 +22,8 @@ from ereuse_devicehub.resources.tag import Tag
class Device(Thing):
EVENT_SORT_KEY = attrgetter('created')
id = ... # type: Column
type = ... # type: Column
hid = ... # type: Column
@ -48,7 +51,6 @@ class Device(Thing):
self.height = ... # type: float
self.depth = ... # type: float
self.color = ... # type: Color
self.events = ... # type: List[e.Event]
self.physical_properties = ... # type: Dict[str, object or None]
self.events_multiple = ... # type: Set[e.EventWithMultipleDevices]
self.events_one = ... # type: Set[e.EventWithOneDevice]
@ -57,33 +59,48 @@ class Device(Thing):
self.lots = ... # type: Set[Lot]
self.production_date = ... # type: datetime
@property
def events(self) -> List[e.Event]:
pass
@property
def problems(self) -> List[e.Event]:
pass
@property
def url(self) -> urlutils.URL:
pass
@property
def rate(self) -> Union[e.AggregateRate, None]:
def rate(self) -> Optional[e.AggregateRate]:
pass
@property
def price(self) -> Union[e.Price, None]:
def price(self) -> Optional[e.Price]:
pass
@property
def trading(self) -> Union[states.Trading, None]:
def trading(self) -> Optional[states.Trading]:
pass
@property
def physical(self) -> Union[states.Physical, None]:
def physical(self) -> Optional[states.Physical]:
pass
@property
def physical_possessor(self) -> Union[Agent, None]:
def physical_possessor(self) -> Optional[Agent]:
pass
@property
def working(self) -> List[e.Test]:
pass
def last_event_of(self, *types: Type[e.Event]) -> e.Event:
pass
def _warning_events(self, events: Iterable[e.Event]) -> Generator[e.Event]:
pass
class DisplayMixin:
technology = ... # type: Column
@ -139,6 +156,10 @@ class Computer(DisplayMixin, Device):
def network_speeds(self) -> List[int]:
pass
@property
def privacy(self) -> Set[e.EraseBasic]:
pass
class Desktop(Computer):
pass
@ -219,7 +240,7 @@ class DataStorage(Component):
self.interface = ... # type: DataStorageInterface
@property
def privacy(self) -> DataStoragePrivacyCompliance:
def privacy(self) -> Optional[e.EraseBasic]:
pass

23
ereuse_devicehub/resources/device/schemas.py
View file

@ -8,9 +8,8 @@ from teal.marshmallow import EnumField, SanitizedStr, URL, ValidationError
from teal.resource import Schema
from ereuse_devicehub.marshmallow import NestedOn
from ereuse_devicehub.resources import enums
from ereuse_devicehub.resources.device import models as m, states
from ereuse_devicehub.resources.enums import ComputerChassis, DataStorageInterface, \
DataStoragePrivacyCompliance, DisplayTech, PrinterTechnology, RamFormat, RamInterface
from ereuse_devicehub.resources.models import STR_BIG_SIZE, STR_SIZE
from ereuse_devicehub.resources.schemas import Thing, UnitCodes
@ -31,6 +30,7 @@ class Device(Thing):
depth = Float(validate=Range(0.1, 5), unit=UnitCodes.m, description=m.Device.depth.comment)
events = NestedOn('Event', many=True, dump_only=True, description=m.Device.events.__doc__)
events_one = NestedOn('Event', many=True, load_only=True, collection_class=OrderedSet)
problems = NestedOn('Event', many=True, dump_only=True, description=m.Device.problems.__doc__)
url = URL(dump_only=True, description=m.Device.url.__doc__)
lots = NestedOn('Lot',
many=True,
@ -44,6 +44,10 @@ class Device(Thing):
production_date = DateTime('iso',
description=m.Device.updated.comment,
data_key='productionDate')
working = NestedOn('Event',
many=True,
dump_only=True,
description=m.Device.working.__doc__)
@pre_load
def from_events_to_events_one(self, data: dict):
@ -72,12 +76,13 @@ class Device(Thing):
class Computer(Device):
components = NestedOn('Component', many=True, dump_only=True, collection_class=OrderedSet)
chassis = EnumField(ComputerChassis, required=True)
chassis = EnumField(enums.ComputerChassis, required=True)
ram_size = Integer(dump_only=True, data_key='ramSize')
data_storage_size = Integer(dump_only=True, data_key='dataStorageSize')
processor_model = Str(dump_only=True, data_key='processorModel')
graphic_card_model = Str(dump_only=True, data_key='graphicCardModel')
network_speeds = List(Integer(dump_only=True), dump_only=True, data_key='networkSpeeds')
privacy = NestedOn('Event', many=True, dump_only=True, collection_class=set)
class Desktop(Computer):
@ -94,7 +99,7 @@ class Server(Computer):
class DisplayMixin:
size = Float(description=m.DisplayMixin.size.comment, validate=Range(2, 150))
technology = EnumField(DisplayTech,
technology = EnumField(enums.DisplayTech,
description=m.DisplayMixin.technology.comment)
resolution_width = Integer(data_key='resolutionWidth',
validate=Range(10, 20000),
@ -168,8 +173,8 @@ class DataStorage(Component):
size = Integer(validate=Range(0, 10 ** 8),
unit=UnitCodes.mbyte,
description=m.DataStorage.size.comment)
interface = EnumField(DataStorageInterface)
privacy = EnumField(DataStoragePrivacyCompliance, dump_only=True)
interface = EnumField(enums.DataStorageInterface)
privacy = NestedOn('Event', dump_only=True)
class HardDrive(DataStorage):
@ -203,8 +208,8 @@ class Processor(Component):
class RamModule(Component):
size = Integer(validate=Range(min=128, max=17000), unit=UnitCodes.mbyte)
speed = Integer(validate=Range(min=100, max=10000), unit=UnitCodes.mhz)
interface = EnumField(RamInterface)
format = EnumField(RamFormat)
interface = EnumField(enums.RamInterface)
format = EnumField(enums.RamFormat)
class SoundCard(Component):
@ -264,7 +269,7 @@ class WirelessAccessPoint(Networking):
class Printer(Device):
wireless = Boolean(required=True, missing=False)
scanning = Boolean(required=True, missing=False)
technology = EnumField(PrinterTechnology, required=True)
technology = EnumField(enums.PrinterTechnology, required=True)
monochrome = Boolean(required=True, missing=True)

19
ereuse_devicehub/resources/enums.py
View file

@ -260,25 +260,6 @@ class ReceiverRole(Enum):
Transporter = 'An user that ships the devices to another one.'
class DataStoragePrivacyCompliance(Enum):
EraseBasic = 'EraseBasic'
EraseBasicError = 'EraseBasicError'
EraseSectors = 'EraseSectors'
EraseSectorsError = 'EraseSectorsError'
Destruction = 'Destruction'
DestructionError = 'DestructionError'
@classmethod
def from_erase(cls, erasure) -> 'DataStoragePrivacyCompliance':
"""Returns the correct enum depending of the passed-in erasure."""
from ereuse_devicehub.resources.event.models import EraseSectors
if isinstance(erasure, EraseSectors):
c = cls.EraseSectors if erasure.severity != Severity.Error else cls.EraseSectorsError
else:
c = cls.EraseBasic if erasure.severity == Severity.Error else cls.EraseBasicError
return c
class PrinterTechnology(Enum):
"""Technology of the printer."""
Toner = 'Toner / Laser'

6
ereuse_devicehub/resources/event/models.py
View file

@ -299,6 +299,12 @@ class EraseSectors(EraseBasic):
pass
class ErasePhysical(EraseBasic):
"""Physical destruction of a data storage unit."""
# todo add attributes
pass
class Step(db.Model):
erasure_id = Column(UUID(as_uuid=True), ForeignKey(EraseBasic.id), primary_key=True)
type = Column(Unicode(STR_SM_SIZE), nullable=False)

5
tests/test_device.py
View file

@ -73,6 +73,11 @@ def test_device_model():
assert d.GraphicCard.query.first() is None, 'We should have deleted it –it was inside the pc'
@pytest.mark.xfail(reason='Test not developed')
def test_device_problems():
pass
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_device_schema():
"""Ensures the user does not upload non-writable or extra fields."""

29
tests/test_event.py
View file

@ -86,18 +86,35 @@ def test_erase_sectors_steps():
@pytest.mark.usefixtures(conftest.auth_app_context.__name__)
def test_test_data_storage():
def test_test_data_storage_working():
"""Tests TestDataStorage with the resulting properties in Device."""
hdd = HardDrive(serial_number='foo', manufacturer='bar', model='foo-bar')
test = models.TestDataStorage(
device=HardDrive(serial_number='foo', manufacturer='bar', model='foo-bar'),
severity=Severity.Info,
device=hdd,
severity=Severity.Error,
elapsed=timedelta(minutes=25),
length=TestDataStorageLength.Short,
status='ok!',
status=':-(',
lifetime=timedelta(days=120)
)
db.session.add(test)
db.session.commit()
assert models.TestDataStorage.query.one()
db.session.flush()
assert hdd.working == [test]
assert not hdd.problems
# Add new test overriding the first test in the problems
# / working condition
test2 = models.TestDataStorage(
device=hdd,
severity=Severity.Warning,
elapsed=timedelta(minutes=25),
length=TestDataStorageLength.Short,
status=':-(',
lifetime=timedelta(days=120)
)
db.session.add(test2)
db.session.flush()
assert hdd.working == [test2]
assert hdd.problems == []
@pytest.mark.usefixtures(conftest.auth_app_context.__name__)

19
tests/test_snapshot.py
View file

@ -289,8 +289,10 @@ def test_snapshot_component_containing_components(user: UserClient):
user.post(s, res=Snapshot, status=ValidationError)
def test_erase(user: UserClient):
"""Tests a Snapshot with EraseSectors."""
def test_erase_privacy(user: UserClient):
"""Tests a Snapshot with EraseSectors and the resulting
privacy properties.
"""
s = file('erase-sectors.snapshot')
snapshot = snapshot_and_check(user, s, (EraseSectors.t,), perform_second_snapshot=True)
storage, *_ = snapshot['components']
@ -312,14 +314,19 @@ def test_erase(user: UserClient):
assert step['type'] == 'StepZero'
assert step['severity'] == 'Info'
assert 'num' not in step
assert storage['privacy'] == erasure['device']['privacy'] == 'EraseSectors'
assert storage['privacy']['type'] == 'EraseSectors'
pc, _ = user.get(res=m.Device, item=snapshot['device']['id'])
assert pc['privacy'] == [storage['privacy']]
# Let's try a second erasure with an error
s['uuid'] = uuid4()
s['components'][0]['events'][0]['severity'] = 'Error'
snapshot, _ = user.post(s, res=Snapshot)
assert snapshot['components'][0]['hid'] == 'c1mr-c1s-c1ml'
assert snapshot['components'][0]['privacy'] == 'EraseSectorsError'
storage, _ = user.get(res=m.Device, item=storage['id'])
assert storage['hid'] == 'c1mr-c1s-c1ml'
assert storage['privacy']['type'] == 'EraseSectors'
pc, _ = user.get(res=m.Device, item=snapshot['device']['id'])
assert pc['privacy'] == [storage['privacy']]
def test_test_data_storage(user: UserClient):
@ -330,7 +337,7 @@ def test_test_data_storage(user: UserClient):
ev for ev in snapshot['events']
if ev.get('reallocatedSectorCount', None) == 15
)
incidence_test['severity'] == 'Error'
assert incidence_test['severity'] == 'Error'
def test_snapshot_computer_monitor(user: UserClient):