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

Update to functional endpoint

This commit is contained in:
Xavier Bustamante Talavera 2018-04-27 19:16:43 +02:00
parent c4b6553c8c
commit 8723b379b0
38 changed files with 1165 additions and 242 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
ereuse_devicehub/auth.py
View File

@ -1,6 +1,14 @@
from sqlalchemy.exc import DataError
from werkzeug.exceptions import Unauthorized
from ereuse_devicehub.resources.user.models import User
from teal.auth import TokenAuth
from teal.db import ResourceNotFound
class Auth(TokenAuth):
pass
def authenticate(self, token: str, *args, **kw) -> User:
try:
return User.query.filter_by(token=token).one()
except (ResourceNotFound, DataError):
raise Unauthorized('Provide a suitable token.')

38
ereuse_devicehub/client.py
View File

@ -1,5 +1,41 @@
from ereuse_utils.test import JSON
from flask import Response
from werkzeug.exceptions import HTTPException
from teal.client import Client as TealClient
class Client(TealClient):
pass
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 login(self, email: str, password: str):
assert isinstance(email, str)
assert isinstance(password, str)
return self.post({'email': email, 'password': password}, '/users/login', status=200)
class UserClient(Client):
"""
A client that identifies all of its requests with a specific user.
It will automatically perform login on the first request.
"""
def __init__(self, application,
email: str,
password: str,
response_wrapper=None,
use_cookies=False,
allow_subdomain_redirects=False):
super().__init__(application, response_wrapper, use_cookies, allow_subdomain_redirects)
self.email = email # type: str
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,
**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)

17
ereuse_devicehub/config.py
View File

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

6
ereuse_devicehub/db.py
View File

@ -1,5 +1,3 @@
from flask_sqlalchemy import SQLAlchemy
from teal.db import SQLAlchemy
from teal.db import Model
db = SQLAlchemy(model_class=Model)
db = SQLAlchemy()

21
ereuse_devicehub/devicehub.py
View File

@ -1,6 +1,27 @@
from typing import Type
from flask_sqlalchemy import SQLAlchemy
from ereuse_devicehub.auth import Auth
from ereuse_devicehub.client import Client
from ereuse_devicehub.db import db
from teal.config import Config as ConfigClass
from teal.teal import Teal
class Devicehub(Teal):
test_client_class = Client
def __init__(self, config: ConfigClass,
db: SQLAlchemy = db,
import_name=__package__,
static_path=None,
static_url_path=None,
static_folder='static',
template_folder='templates',
instance_path=None,
instance_relative_config=False,
root_path=None,
Auth: Type[Auth] = Auth):
super().__init__(config, db, import_name, static_path, static_url_path, static_folder,
template_folder, instance_path, instance_relative_config, root_path, Auth)

11
ereuse_devicehub/marshmallow.py Normal file
View File

@ -0,0 +1,11 @@
from marshmallow.fields import missing_
from ereuse_devicehub.db import db
from teal.db import SQLAlchemy
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)

55
ereuse_devicehub/resources/device/__init__.py
View File

@ -1,9 +1,60 @@
from ereuse_devicehub.resources.device.schemas import Device
from ereuse_devicehub.resources.device.schemas import Component, Computer, Desktop, Device, \
GraphicCard, HardDrive, Laptop, Microtower, Motherboard, Netbook, NetworkAdapter, RamModule, \
Server
from ereuse_devicehub.resources.device.views import DeviceView
from teal.resource import Resource, Converters
from teal.resource import Converters, Resource
class DeviceDef(Resource):
SCHEMA = Device
VIEW = DeviceView
ID_CONVERTER = Converters.int
AUTH = True
class ComputerDef(DeviceDef):
SCHEMA = Computer
class DesktopDef(ComputerDef):
SCHEMA = Desktop
class LaptopDef(ComputerDef):
SCHEMA = Laptop
class NetbookDef(ComputerDef):
SCHEMA = Netbook
class ServerDef(ComputerDef):
SCHEMA = Server
class MicrotowerDef(ComputerDef):
SCHEMA = Microtower
class ComponentDef(DeviceDef):
SCHEMA = Component
class GraphicCardDef(ComponentDef):
SCHEMA = GraphicCard
class HardDriveDef(ComponentDef):
SCHEMA = HardDrive
class MotherboardDef(ComponentDef):
SCHEMA = Motherboard
class NetworkAdapterDef(ComponentDef):
SCHEMA = NetworkAdapter
class RamModuleDef(ComponentDef):
SCHEMA = RamModule

14
ereuse_devicehub/resources/device/exceptions.py Normal file
View File

@ -0,0 +1,14 @@
from marshmallow import ValidationError
class MismatchBetweenIds(ValidationError):
def __init__(self, other_device_id: int, field: str, value: str):
message = 'The device {} has the same {} than this one ({}).'.format(other_device_id,
field, value)
super().__init__(message, field_names=[field])
class NeedsId(ValidationError):
def __init__(self):
message = 'We couldn\'t get an ID for this device. Is this a custom PC?'
super().__init__(message)

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

@ -1,24 +1,47 @@
from contextlib import suppress
from typing import Dict, Set
from ereuse_utils.naming import Naming
from sqlalchemy import BigInteger, Column, Float, ForeignKey, Integer, Sequence, SmallInteger, \
Unicode
Unicode, inspect
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import backref, relationship
from ereuse_devicehub.resources.model import STR_BIG_SIZE, STR_SIZE, Thing, check_range
from teal.db import POLYMORPHIC_ID, POLYMORPHIC_ON, CASCADE
from ereuse_devicehub.resources.models import STR_BIG_SIZE, STR_SIZE, STR_SM_SIZE, Thing, \
check_range
from teal.db import CASCADE, POLYMORPHIC_ID, POLYMORPHIC_ON, ResourceNotFound
class Device(Thing):
id = Column(BigInteger, Sequence('device_seq'), primary_key=True)
type = Column(Unicode)
pid = Column(Unicode(STR_SIZE), unique=True)
gid = Column(Unicode(STR_SIZE), unique=True)
hid = Column(Unicode(STR_BIG_SIZE), unique=True)
model = Column(Unicode(STR_BIG_SIZE))
manufacturer = Column(Unicode(STR_SIZE))
serial_number = Column(Unicode(STR_SIZE))
weight = Column(Float(precision=3), check_range('weight', min=0.1))
width = Column(Float(precision=3), check_range('width', min=0.1))
height = Column(Float(precision=3), check_range('height', min=0.1))
id = Column(BigInteger, Sequence('device_seq'), primary_key=True) # type: int
type = Column(Unicode(STR_SM_SIZE), nullable=False)
hid = Column(Unicode(STR_BIG_SIZE), unique=True) # type: str
pid = Column(Unicode(STR_SIZE)) # type: str
gid = Column(Unicode(STR_SIZE)) # type: str
model = Column(Unicode(STR_BIG_SIZE)) # type: str
manufacturer = Column(Unicode(STR_SIZE)) # type: str
serial_number = Column(Unicode(STR_SIZE)) # type: str
weight = Column(Float(precision=3, decimal_return_scale=3),
check_range('weight', 0.1, 3)) # type: float
width = Column(Float(precision=3, decimal_return_scale=3),
check_range('width', 0.1, 3)) # type: float
height = Column(Float(precision=3, decimal_return_scale=3),
check_range('height', 0.1, 3)) # type: float
@property
def physical_properties(self) -> Dict[Column, object or None]:
"""
Fields that describe the physical properties of a device.
:return A generator where each value is a tuple with tho fields:
- Column.
- Actual value of the column or None.
"""
# 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)
for c in inspect(self.__class__).attrs
if not c.foreign_keys and c not in {self.id, self.type}}
@declared_attr
def __mapper_args__(cls):
@ -34,9 +57,14 @@ 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)
class Computer(Device):
id = Column(BigInteger, ForeignKey(Device.id), primary_key=True)
id = Column(BigInteger, ForeignKey(Device.id), primary_key=True) # type: int
class Desktop(Computer):
@ -60,29 +88,58 @@ class Microtower(Computer):
class Component(Device):
id = Column(BigInteger, ForeignKey(Device.id), primary_key=True)
id = Column(BigInteger, ForeignKey(Device.id), primary_key=True) # type: int
parent_id = Column(BigInteger, ForeignKey('computer.id'))
parent = relationship(Computer,
backref=backref('components', lazy=True, cascade=CASCADE),
primaryjoin='Component.parent_id == Computer.id')
primaryjoin='Component.parent_id == Computer.id') # type: Device
def similar_one(self, parent: Computer, blacklist: Set[int]) -> 'Component':
"""
Gets a component that:
- has the same parent.
- Doesn't generate HID.
- Has same physical properties.
:param parent:
:param blacklist: A set of components to not to consider
when looking for similar ones.
"""
assert self.hid is None, 'Don\'t use this method with a component that has HID'
component = self.__class__.query \
.filter_by(parent=parent, hid=None, **self.physical_properties) \
.filter(~Component.id.in_(blacklist)) \
.first()
if not component:
raise ResourceNotFound(self.type)
return component
class GraphicCard(Component):
memory = Column(SmallInteger, check_range('memory', min=0.1))
id = Column(BigInteger, ForeignKey(Component.id), primary_key=True) # type: int
memory = Column(SmallInteger, check_range('memory', min=1, max=10000)) # type: int
class HardDrive(Component):
size = Column(Integer, check_range('size', min=0.1))
id = Column(BigInteger, ForeignKey(Component.id), primary_key=True) # type: int
size = Column(Integer, check_range('size', min=1, max=10 ** 8)) # type: int
class Motherboard(Component):
slots = Column(SmallInteger, check_range('slots'))
usb = Column(SmallInteger, check_range('usb'))
firewire = Column(SmallInteger, check_range('firewire'))
serial = Column(SmallInteger, check_range('serial'))
pcmcia = Column(SmallInteger, check_range('pcmcia'))
id = Column(BigInteger, ForeignKey(Component.id), primary_key=True) # type: int
slots = Column(SmallInteger, check_range('slots')) # type: int
usb = Column(SmallInteger, check_range('usb')) # type: int
firewire = Column(SmallInteger, check_range('firewire')) # type: int
serial = Column(SmallInteger, check_range('serial')) # type: int
pcmcia = Column(SmallInteger, check_range('pcmcia')) # type: int
class NetworkAdapter(Component):
speed = Column(SmallInteger, check_range('speed'))
id = Column(BigInteger, ForeignKey(Component.id), primary_key=True) # type: int
speed = Column(SmallInteger, check_range('speed', min=10, max=10000)) # type: int
class RamModule(Component):
id = Column(BigInteger, ForeignKey(Component.id), primary_key=True) # type: int
size = Column(SmallInteger, check_range('size', min=128, max=17000))
speed = Column(Float, check_range('speed', min=100, max=10000))

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

@ -1,12 +1,13 @@
from marshmallow.fields import Float, Integer, Nested, Str
from marshmallow.validate import Length, Range
from ereuse_devicehub.resources.model import STR_BIG_SIZE, STR_SIZE
from ereuse_devicehub.resources.schema import Thing
from ereuse_devicehub.resources.models import STR_BIG_SIZE, STR_SIZE
from ereuse_devicehub.resources.schemas import Thing, UnitCodes
class Device(Thing):
id = Str(dump_only=True)
id = Integer(dump_only=True,
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.')
@ -17,11 +18,17 @@ class Device(Thing):
validate=Length(max=STR_SIZE))
model = Str(validate=Length(max=STR_BIG_SIZE))
manufacturer = Str(validate=Length(max=STR_SIZE))
serial_number = Str(load_from='serialNumber', dump_to='serialNumber')
product_id = Str(load_from='productId', dump_to='productId')
weight = Float(validate=Range(0.1, 3))
width = Float(validate=Range(0.1, 3))
height = Float(validate=Range(0.1, 3))
serial_number = Str(data_key='serialNumber')
product_id = Str(data_key='productId')
weight = Float(validate=Range(0.1, 3),
unit=UnitCodes.kgm,
description='The weight of the device in Kgm.')
width = Float(validate=Range(0.1, 3),
unit=UnitCodes.m,
description='The width of the device in meters.')
height = Float(validate=Range(0.1, 3),
unit=UnitCodes.m,
description='The height of the device in meters.')
events = Nested('Event', many=True, dump_only=True, only='id')
@ -51,19 +58,22 @@ class Microtower(Computer):
class Component(Device):
parent = Nested(Device, dump_only=True)
parent = Nested(Device, dump_only=True, only='id')
class GraphicCard(Component):
memory = Integer(validate=Range(0, 10000))
memory = Integer(validate=Range(0, 10000),
unit=UnitCodes.mbyte,
description='The amount of memory of the Graphic Card in MB.')
class HardDrive(Component):
size = Integer(validate=Range(0, 10 ** 8))
size = Integer(validate=Range(0, 10 ** 8),
unit=UnitCodes.mbyte,
description='The size of the hard-drive in MB.')
erasure = Nested('EraseBasic', load_only=True)
erasures = Nested('EraseBasic', dump_only=True, many=True)
tests = Nested('TestHardDrive', many=True)
benchmarks = Nested('BenchmarkHardDrive', many=True)
tests = Nested('TestHardDrive', many=True, load_only=True)
benchmarks = Nested('BenchmarkHardDrive', load_only=True, many=True)
class Motherboard(Component):
@ -75,4 +85,11 @@ class Motherboard(Component):
class NetworkAdapter(Component):
speed = Integer(validate=Range(min=10, max=10000))
speed = Integer(validate=Range(min=10, max=10000),
unit=UnitCodes.mbps,
description='The maximum speed this network adapter can handle, in mbps.')
class RamModule(Component):
size = Integer(validate=Range(min=128, max=17000), unit=UnitCodes.mbyte)
speed = Float(validate=Range(min=100, max=10000), unit=UnitCodes.mhz)

158
ereuse_devicehub/resources/device/sync.py Normal file
View File

@ -0,0 +1,158 @@
from contextlib import suppress
from itertools import groupby
from typing import Iterable, List, Set
from psycopg2.errorcodes import UNIQUE_VIOLATION
from sqlalchemy.exc import IntegrityError
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.device.exceptions import NeedsId
from ereuse_devicehub.resources.device.models import Component, Computer, Device
from ereuse_devicehub.resources.event.models import Add, Remove
from teal.db import ResourceNotFound
class Sync:
"""Synchronizes the device and components with the database."""
@classmethod
def run(cls, device: Device,
components: Iterable[Component] or None,
force_creation: bool = False) -> (Device, List[Component], List[Add or Remove]):
"""
Synchronizes the device and components with the database.
Identifies if the device and components exist in the database
and updates / inserts them as necessary.
This performs Add / Remove as necessary.
:param device: The device to add / update to the database.
:param components: Components that are inside of the device.
This method performs Add and Remove events
so the device ends up with these components.
Components are added / updated accordingly.
If this is empty, all components are removed.
If this is None, it means that there is
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).
2. The same passed-in components from the database (with
ids).
3. 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_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
@staticmethod
def execute_register(device: Device,
blacklist: Set[int],
force_creation: bool = False,
parent: Computer = None) -> Device:
"""
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.
: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 synchronized device with the DB. It can be a new
device or an already existing one.
:raise NeedsId: The device has not any identifier we can use.
To still create the device use
``force_creation``.
:raise DatabaseError: Any other error from the DB.
"""
# Let's try to create the device
if not device.hid and not device.id:
# We won't be able to surely identify this device
if isinstance(device, Component):
with suppress(ResourceNotFound):
# Is there a component similar to ours?
db_component = device.similar_one(parent, blacklist)
# We blacklist this component so we
# ensure we don't get it again for another component
# with the same physical properties
blacklist.add(db_component.id)
return db_component
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()
except IntegrityError as e:
if e.orig.diag.sqlstate == UNIQUE_VIOLATION:
# 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()
else:
raise e
else:
return device # 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:
if value is not None:
setattr(db_device, field.name, value)
return db_device
@classmethod
def add_remove(cls, device: Device,
new_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:
"""
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
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

6
ereuse_devicehub/resources/device/views.py
View File

@ -1,8 +1,8 @@
from ereuse_devicehub.resources.device.models import Device
from teal.resource import View
class DeviceView(View):
def one(self, id):
def one(self, id: int):
"""Gets one device."""
raise NotImplementedError
return Device.query.filter_by(id=id).one()

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

@ -1,7 +1,15 @@
from ereuse_devicehub.resources.event.views import EventView
from teal.resource import Resource
from ereuse_devicehub.resources.event.schemas import Snapshot, Event
from ereuse_devicehub.resources.event.views import EventView, SnapshotView
from teal.resource import Converters, Resource
class EventDef(Resource):
SCHEMA = None
SCHEMA = Event
VIEW = EventView
AUTH = True
ID_CONVERTER = Converters.int
class SnapshotDef(EventDef):
SCHEMA = Snapshot
VIEW = SnapshotView

51
ereuse_devicehub/resources/event/enums.py Normal file
View File

@ -0,0 +1,51 @@
from enum import Enum
class StepTypes(Enum):
Zeros = 1
Random = 2
class SoftwareType(Enum):
"""The software used to perform the Snapshot."""
Workbench = 'Workbench'
AndroidApp = 'AndroidApp'
Web = 'Web'
DesktopApp = 'DesktopApp'
class Appearance(Enum):
"""Grades the imperfections that aesthetically affect the device, but not its usage."""
Z = '0. The device is new.'
A = 'A. Is like new (without visual damage)'
B = 'B. Is in really good condition (small visual damage in difficult places to spot)'
C = 'C. Is in good condition (small visual damage in parts that are easy to spot, not screens)'
D = 'D. Is acceptable (visual damage in visible parts, not screens)'
E = 'E. Is unacceptable (considerable visual damage that can affect usage)'
class Functionality(Enum):
"""Grades the defects of a device that affect its usage."""
A = 'A. Everything works perfectly (buttons, and in case of screens there are no scratches)'
B = 'B. There is a button difficult to press or a small scratch in an edge of a screen'
C = 'C. A non-important button (or similar) doesn\'t work; screen has multiple scratches in edges'
D = 'D. Multiple buttons don\'t work; screen has visual damage resulting in uncomfortable usage'
class Bios(Enum):
"""How difficult it has been to set the bios to boot from the network."""
A = 'A. If by pressing a key you could access a boot menu with the network boot'
B = 'B. You had to get into the BIOS, and in less than 5 steps you could set the network boot'
C = 'C. Like B, but with more than 5 steps'
D = 'D. Like B or C, but you had to unlock the BIOS (i.e. by removing the battery)'
E = 'E. The device could not be booted through the network.'
class Orientation(Enum):
Vertical = 'vertical'
Horizontal = 'Horizontal'
class TestHardDriveLength(Enum):
Short = 'Short'
Extended = 'Extended'

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

@ -1,15 +1,20 @@
from enum import Enum
from datetime import timedelta
from colour import Color
from sqlalchemy import BigInteger, Boolean, CheckConstraint, Column, DateTime, Enum as DBEnum, \
ForeignKey, Integer, Interval, JSON, Sequence, SmallInteger, Unicode
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import backref, relationship
from sqlalchemy.orm import backref, relationship, validates
from sqlalchemy_utils import ColorType
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.device.models import Device
from ereuse_devicehub.resources.model import STR_SIZE, Thing, check_range
from ereuse_devicehub.resources.user.model import User
from ereuse_devicehub.resources.event.enums import Appearance, Bios, Functionality, Orientation, \
SoftwareType, StepTypes, TestHardDriveLength
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
@ -21,10 +26,12 @@ class JoinedTableMixin:
class Event(Thing):
id = Column(BigInteger, Sequence('event_seq'), primary_key=True)
title = Column(Unicode(STR_BIG_SIZE), default='', nullable=False)
date = Column(DateTime)
secured = Column(Boolean, default=False, nullable=False)
type = Column(Unicode)
incidence = Column(Boolean, default=False, nullable=False)
description = Column(Unicode, default='', nullable=False)
snapshot_id = Column(BigInteger, ForeignKey('snapshot.id',
use_alter=True,
@ -33,7 +40,7 @@ class Event(Thing):
backref=backref('events', lazy=True, cascade=CASCADE),
primaryjoin='Event.snapshot_id == Snapshot.id')
author_id = Column(BigInteger, ForeignKey(User.id), nullable=False)
author_id = Column(UUID(as_uuid=True), ForeignKey(User.id), nullable=False)
author = relationship(User,
backref=backref('events', lazy=True),
primaryjoin=author_id == User.id)
@ -94,13 +101,15 @@ class Remove(EventWithOneDevice):
class Allocate(JoinedTableMixin, EventWithMultipleDevices):
to_id = Column(BigInteger, ForeignKey(User.id))
to_id = Column(UUID, ForeignKey(User.id))
to = relationship(User, primaryjoin=User.id == to_id)
organization = Column(Unicode(STR_SIZE))
class Deallocate(JoinedTableMixin, EventWithMultipleDevices):
from_id = Column(BigInteger, ForeignKey(User.id))
from_id = Column(UUID, ForeignKey(User.id))
from_rel = relationship(User, primaryjoin=User.id == from_id)
organization = Column(Unicode(STR_SIZE))
class EraseBasic(JoinedTableMixin, EventWithOneDevice):
@ -116,14 +125,9 @@ class EraseSectors(EraseBasic):
pass
class StepTypes(Enum):
Zeros = 1
Random = 2
class Step(db.Model):
id = Column(BigInteger, Sequence('step_seq'), primary_key=True)
num = Column(SmallInteger, primary_key=True)
num = Column(SmallInteger, nullable=False)
type = Column(DBEnum(StepTypes), nullable=False)
success = Column(Boolean, nullable=False)
starting_time = Column(DateTime, nullable=False)
@ -136,55 +140,37 @@ class Step(db.Model):
erasure = relationship(EraseBasic, backref=backref('steps', cascade=CASCADE_OWN))
class SoftwareType(Enum):
Workbench = 'Workbench'
AndroidApp = 'AndroidApp'
Web = 'Web'
DesktopApp = 'DesktopApp'
class Appearance(Enum):
"""Grades the imperfections that aesthetically affect the device, but not its usage."""
Z = '0. The device is new.'
A = 'A. Is like new (without visual damage)'
B = 'B. Is in really good condition (small visual damage in difficult places to spot)'
C = 'C. Is in good condition (small visual damage in parts that are easy to spot, not screens)'
D = 'D. Is acceptable (visual damage in visible parts, not ¬screens)'
E = 'E. Is unacceptable (considerable visual damage that can affect usage)'
class Functionality(Enum):
A = 'A. Everything works perfectly (buttons, and in case of screens there are no scratches)'
B = 'B. There is a button difficult to press or a small scratch in an edge of a screen'
C = 'C. A non-important button (or similar) doesn\'t work; screen has multiple scratches in edges'
D = 'D. Multiple buttons don\'t work; screen has visual damage resulting in uncomfortable usage'
class Bios(Enum):
A = 'A. If by pressing a key you could access a boot menu with the network boot'
B = 'B. You had to get into the BIOS, and in less than 5 steps you could set the network boot'
C = 'C. Like B, but with more than 5 steps'
D = 'D. Like B or C, but you had to unlock the BIOS (i.e. by removing the battery)'
E = 'E. The device could not be booted through the network.'
class Snapshot(JoinedTableMixin, EventWithOneDevice):
uuid = Column(UUID(as_uuid=True), nullable=False, unique=True)
version = Column(Unicode, nullable=False)
snapshot_software = Column(DBEnum(SoftwareType), nullable=False)
appearance = Column(DBEnum(Appearance), nullable=False)
appearance_score = Column(SmallInteger, nullable=False)
functionality = Column(DBEnum(Functionality), nullable=False)
functionality_score = Column(SmallInteger, check_range('functionality_score', min=0, max=5),
nullable=False)
labelling = Column(Boolean, nullable=False)
bios = Column(DBEnum(Bios), nullable=False)
condition = Column(SmallInteger, check_range('condition', min=0, max=5), nullable=False)
elapsed = Column(Interval, nullable=False)
install_name = Column(Unicode)
install_elapsed = Column(Interval)
install_success = Column(Boolean)
inventory_elapsed = Column(Interval)
uuid = Column(UUID(as_uuid=True), nullable=False, unique=True) # type: UUID
version = Column(Unicode(STR_SM_SIZE), nullable=False) # type: str
software = Column(DBEnum(SoftwareType), nullable=False) # type: SoftwareType
appearance = Column(DBEnum(Appearance), nullable=False) # type: Appearance
appearance_score = Column(SmallInteger,
check_range('appearance_score', -3, 5),
nullable=False) # type: int
functionality = Column(DBEnum(Functionality), nullable=False) # type: Functionality
functionality_score = Column(SmallInteger,
check_range('functionality_score', min=-3, max=5),
nullable=False) # 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
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
@validates('components')
def validate_components_only_workbench(self, _, components):
if self.software != SoftwareType.Workbench:
if components:
raise ValueError('Only Snapshots from Workbench can have components.')
return components
class SnapshotRequest(db.Model):
@ -202,11 +188,6 @@ class Test(JoinedTableMixin, EventWithOneDevice):
snapshot = relationship(Snapshot, backref=backref('tests', lazy=True, cascade=CASCADE_OWN))
class TestHardDriveLength(Enum):
Short = 'Short'
Extended = 'Extended'
class TestHardDrive(Test):
length = Column(DBEnum(TestHardDriveLength), nullable=False) # todo from type
status = Column(Unicode(STR_SIZE), nullable=False)

0
ereuse_devicehub/resources/event/remove/__init__.py
View File

7
ereuse_devicehub/resources/event/remove/views.py
View File

@ -1,7 +0,0 @@
from teal.resource import View
class Remove(View):
def post(self):
"""Removes a component from a computer."""
pass

171
ereuse_devicehub/resources/event/schemas.py Normal file
View File

@ -0,0 +1,171 @@
from flask import current_app as app
from marshmallow import ValidationError, validates_schema
from marshmallow.fields import Boolean, DateTime, Integer, Nested, String, TimeDelta, UUID
from marshmallow.validate import Length, Range
from marshmallow_enum import EnumField
from ereuse_devicehub.marshmallow import NestedOn
from ereuse_devicehub.resources.device.schemas import Component, Device
from ereuse_devicehub.resources.event.enums import Appearance, Bios, Functionality, Orientation, \
SoftwareType, StepTypes, TestHardDriveLength
from ereuse_devicehub.resources.models import STR_BIG_SIZE, STR_SIZE
from ereuse_devicehub.resources.schemas import Thing
from ereuse_devicehub.resources.user.schemas import User
from teal.marshmallow import Color, Version
from teal.resource import Schema
class Event(Thing):
id = Integer(dump_only=True)
title = String(default='',
validate=Length(STR_BIG_SIZE),
description='A name or title for the event. Used when searching for events.')
date = DateTime('iso', description='When this event happened. '
'Leave it blank if it is happening now. '
'This is used when creating events retroactively.')
secured = Boolean(default=False,
description='Can we ensure the info in this event is totally correct?'
'Devicehub will automatically set this too for some events,'
'for example in snapshots if it could detect the ids of the'
'hardware without margin of doubt.')
incidence = Boolean(default=False,
description='Was something wrong in this event?')
snapshot = Nested('Snapshot', dump_only=True, only='id')
description = String(default='', description='A comment about the event.')
components = Nested(Component, dump_only=True, only='id', many=True)
class EventWithOneDevice(Event):
device = Nested(Device, only='id')
class EventWithMultipleDevices(Event):
device = Nested(Device, many=True, only='id')
class Add(EventWithOneDevice):
pass
class Remove(EventWithOneDevice):
pass
class Allocate(EventWithMultipleDevices):
to = Nested(User, only='id',
description='The user the devices are allocated to.')
organization = String(validate=Length(STR_SIZE),
description='The organization where the user was when this happened.')
class Deallocate(EventWithMultipleDevices):
from_rel = Nested(User, only='id',
data_key='from',
description='The user where the devices are not allocated to anymore.')
organization = String(validate=Length(STR_SIZE),
description='The organization where the user was when this happened.')
class EraseBasic(EventWithOneDevice):
starting_time = DateTime(required=True, data_key='startingTime')
ending_time = DateTime(required=True, data_key='endingTime')
secure_random_steps = Integer(validate=Range(min=0), required=True,
data_key='secureRandomSteps')
success = Boolean(required=True)
clean_with_zeros = Boolean(required=True, data_key='cleanWithZeros')
class EraseSectors(EraseBasic):
pass
class Step(Schema):
id = Integer(dump_only=True)
type = EnumField(StepTypes, required=True)
starting_time = DateTime(required=True, data_key='startingTime')
ending_time = DateTime(required=True, data_key='endingTime')
secure_random_steps = Integer(validate=Range(min=0),
required=True,
data_key='secureRandomSteps')
success = Boolean(required=True)
clean_with_zeros = Boolean(required=True, data_key='cleanWithZeros')
class Condition(Schema):
appearance = EnumField(Appearance,
required=True,
description='Grades the imperfections that aesthetically '
'affect the device, but not its usage.')
appearance_score = Integer(validate=Range(-3, 5), dump_only=True)
functionality = EnumField(Functionality,
required=True,
description='Grades the defects of a device that affect its usage.')
functionality_score = Integer(validate=Range(-3, 5),
dump_only=True,
data_key='functionalityScore')
labelling = Boolean(description='Sets if there are labels stuck that should be removed.')
bios = EnumField(Bios, description='How difficult it has been to set the bios to '
'boot from the network.')
general = Integer(dump_only=True,
validate=Range(0, 5),
description='The grade of the device.')
class Installation(Schema):
name = String(validate=Length(STR_BIG_SIZE),
required=True,
description='The name of the OS installed.')
elapsed = TimeDelta(precision=TimeDelta.SECONDS, required=True)
success = Boolean(required=True)
class Inventory(Schema):
elapsed = TimeDelta(precision=TimeDelta.SECONDS, required=True)
class Snapshot(EventWithOneDevice):
device = NestedOn(Device) # todo and when dumping?
components = NestedOn(Component, many=True)
uuid = UUID(required=True)
version = Version(required=True, description='The version of the SnapshotSoftware.')
software = EnumField(SoftwareType,
required=True,
description='The software that generated this Snapshot.')
condition = Nested(Condition, required=True)
install = Nested(Installation)
elapsed = TimeDelta(precision=TimeDelta.SECONDS, required=True)
inventory = Nested(Inventory)
color = Color(description='Main color of the device.')
orientation = EnumField(Orientation, description='Is the device main stand wider or larger?')
@validates_schema
def validate_workbench_version(self, data: dict):
if data['software'] == SoftwareType.Workbench:
if data['version'] < app.config['MIN_WORKBENCH']:
raise ValidationError(
'Min. supported Workbench version is {}'.format(app.config['MIN_WORKBENCH']),
field_names=['version']
)
@validates_schema
def validate_components_only_workbench(self, data: dict):
if data['software'] != SoftwareType.Workbench:
if data['components'] is not None:
raise ValidationError('Only Workbench can add component info',
field_names=['components'])
class Test(EventWithOneDevice):
elapsed = TimeDelta(precision=TimeDelta.SECONDS, required=True)
success = Boolean(required=True)
class TestHardDrive(Test):
length = EnumField(TestHardDriveLength, required=True)
status = String(validate=Length(max=STR_SIZE), required=True)
lifetime = TimeDelta(precision=TimeDelta.DAYS, required=True)
first_error = Integer()
class StressTest(Test):
pass

7
ereuse_devicehub/resources/event/snapshot/__init__.py
View File

@ -1,7 +0,0 @@
from ereuse_devicehub.resources.event import EventDef
from ereuse_devicehub.resources.event.snapshot.views import SnapshotView
class SnapshotDef(EventDef):
VIEW = SnapshotView
SCHEMA = None

17
ereuse_devicehub/resources/event/snapshot/views.py
View File

@ -1,17 +0,0 @@
from teal.resource import View
class SnapshotView(View):
def post(self):
"""Creates a Snapshot."""
return super().post()
def delete(self, id):
"""Deletes a Snapshot"""
return super().delete(id)
def patch(self, id):
"""Modifies a Snapshot"""
return super().patch(id)

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

@ -1,7 +1,43 @@
from distutils.version import StrictVersion
from flask import request
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.device.sync import Sync
from ereuse_devicehub.resources.event.enums import SoftwareType
from ereuse_devicehub.resources.event.models import Event, Snapshot, TestHardDrive
from teal.resource import View
class EventView(View):
def one(self, id):
def one(self, id: int):
"""Gets one event."""
return super().one(id)
return Event.query.filter_by(id=id).one()
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()?
# noinspection PyArgumentList
c = snapshot.components if snapshot.software == SoftwareType.Workbench else None
snapshot.device, snapshot.components, snapshot.events = Sync.run(snapshot.device, c)
db.session.add(snapshot)
# transform it back
return self.schema.jsonify(snapshot)
class TestHardDriveView(View):
def post(self):
t = request.get_json() # type: dict
# noinspection PyArgumentList
test = TestHardDrive(snapshot_id=t.pop('snapshot'), device_id=t.pop('device'), **t)
return test
class StressTestView(View):
def post(self):
t = request.get_json() # type: dict

0
ereuse_devicehub/resources/model.py → ereuse_devicehub/resources/models.py
View File

10
ereuse_devicehub/resources/schema.py
View File

@ -1,10 +0,0 @@
from marshmallow.fields import DateTime, List, Str, URL, Nested
from teal.resource import Schema
class Thing(Schema):
url = URL(dump_only=True, description='The URL of the resource.')
same_as = List(URL(dump_only=True), dump_only=True)
updated = DateTime('iso', dump_only=True)
created = DateTime('iso', dump_only=True)
author = Nested('User', only='id', dump_only=True)

25
ereuse_devicehub/resources/schemas.py Normal file
View File

@ -0,0 +1,25 @@
from enum import Enum
from marshmallow.fields import DateTime, List, Nested, URL, String
from teal.resource import Schema
class UnitCodes(Enum):
mbyte = '4L'
mbps = 'E20'
mhz = 'MHZ'
gbyte = 'E34'
ghz = 'A86'
bit = 'A99'
kgm = 'KGM'
m = 'MTR'
class Thing(Schema):
type = String(description='Only required when it is nested.')
url = URL(dump_only=True, description='The URL of the resource.')
same_as = List(URL(dump_only=True), dump_only=True, data_key='sameAs')
updated = DateTime('iso', dump_only=True)
created = DateTime('iso', dump_only=True)
author = Nested('User', only='id', dump_only=True)

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

@ -0,0 +1,36 @@
from click import argument, option
from ereuse_devicehub import devicehub
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.user.schemas import User as UserS
from ereuse_devicehub.resources.user.views import UserView, login
from teal.resource import Converters, Resource
class UserDef(Resource):
SCHEMA = UserS
VIEW = UserView
ID_CONVERTER = Converters.uid
AUTH = True
def __init__(self, app: 'devicehub.Devicehub', import_name=__package__, static_folder=None,
static_url_path=None, template_folder=None, url_prefix=None, subdomain=None,
url_defaults=None, root_path=None):
cli_commands = ((self.create_user, 'user create'),)
super().__init__(app, import_name, static_folder, static_url_path, template_folder,
url_prefix, subdomain, url_defaults, root_path, cli_commands)
self.add_url_rule('/login', view_func=login, methods={'POST'})
@argument('email')
@option('--password', prompt=True, hide_input=True, confirmation_prompt=True)
def create_user(self, email: str, password: str) -> dict:
"""
Creates an user.
"""
with self.app.test_request_context():
self.schema.load({'email': email, 'password': password})
user = User(email=email, password=password)
db.session.add(user)
db.session.commit()
return user.dump()

5
ereuse_devicehub/resources/user/exceptions.py Normal file
View File

@ -0,0 +1,5 @@
from werkzeug.exceptions import Unauthorized
class WrongCredentials(Unauthorized):
description = 'There is not an user with the matching username/password'

9
ereuse_devicehub/resources/user/model.py
View File

@ -1,9 +0,0 @@
from sqlalchemy import BigInteger, Column, Sequence
from sqlalchemy_utils import EmailType
from ereuse_devicehub.resources.model import Thing
class User(Thing):
id = Column(BigInteger, Sequence('user_seq'), primary_key=True)
email = Column(EmailType, nullable=False)

26
ereuse_devicehub/resources/user/models.py Normal file
View File

@ -0,0 +1,26 @@
from uuid import uuid4
from flask import current_app
from sqlalchemy import Column, Unicode
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy_utils import EmailType, PasswordType
from ereuse_devicehub.resources.models import STR_SIZE, Thing
class User(Thing):
__table_args__ = {'schema': 'common'}
id = Column(UUID(as_uuid=True), default=uuid4, primary_key=True)
email = Column(EmailType, nullable=False, unique=True)
password = Column(PasswordType(max_length=STR_SIZE,
onload=lambda **kwargs: dict(
schemes=current_app.config['PASSWORD_SCHEMES'],
**kwargs
)))
"""
Password field.
From `here <https://sqlalchemy-utils.readthedocs.io/en/latest/
data_types.html#module-sqlalchemy_utils.types.password>`_
"""
name = Column(Unicode(length=STR_SIZE))
token = Column(UUID(as_uuid=True), default=uuid4, unique=True)

22
ereuse_devicehub/resources/user/schemas.py Normal file
View File

@ -0,0 +1,22 @@
from base64 import b64encode
from marshmallow import pre_dump
from marshmallow.fields import Email, String, UUID
from ereuse_devicehub.resources.schemas import Thing
class User(Thing):
id = UUID(dump_only=True)
email = Email(required=True)
password = String(load_only=True, required=True)
token = String(dump_only=True,
description='Use this token in an Authorization header to access the app.'
'The token can change overtime.')
@pre_dump
def base64encode_token(self, data: dict):
"""Encodes the token to base64 so clients don't have to."""
# framework needs ':' at the end
data['token'] = b64encode(str.encode(str(data['token']) + ':'))
return data

23
ereuse_devicehub/resources/user/views.py Normal file
View File

@ -0,0 +1,23 @@
from uuid import UUID
from flask import current_app as app, request
from ereuse_devicehub.resources.user.exceptions import WrongCredentials
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.user.schemas import User as UserS
from teal.resource import View
class UserView(View):
def one(self, id: UUID):
return self.schema.jsonify(User.query.filter_by(id=id).one())
def login():
user_s = app.resources['User'].schema # type: UserS
u = user_s.load(request.get_json(), partial=('email', 'password'))
user = User.query.filter_by(email=u['email']).one_or_none()
if user and user.password == u['password']:
return user_s.jsonify(user)
else:
raise WrongCredentials()

58
tests/conftest.py
View File

@ -1,31 +1,67 @@
import pytest
import json as stdlib_json
from pathlib import Path
from ereuse_devicehub.client import Client
import pytest
import yaml
from ereuse_devicehub.client import Client, UserClient
from ereuse_devicehub.config import DevicehubConfig
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.user.models import User
class TestConfig(DevicehubConfig):
SQLALCHEMY_DATABASE_URI = 'postgresql://localhost/dh_test'
SQLALCHEMY_BINDS = {
'common': 'postgresql://localhost/dh_test_common'
}
SCHEMA = 'test'
TESTING = True
@pytest.fixture()
@pytest.fixture(scope='module')
def config():
return TestConfig()
@pytest.fixture(scope='module')
def _app(config: TestConfig) -> Devicehub:
return Devicehub(config=config, db=db)
@pytest.fixture()
def app(config: TestConfig) -> Devicehub:
app = Devicehub(config=config, db=db)
db.create_all(app=app)
yield app
db.drop_all(app=app)
def app(request, _app: Devicehub) -> Devicehub:
db.drop_all(app=_app) # In case the test before was killed
db.create_all(app=_app)
# More robust than 'yield'
request.addfinalizer(lambda *args, **kw: db.drop_all(app=_app))
return _app
@pytest.fixture()
def client(app: Devicehub) -> Client:
return app.test_client()
@pytest.fixture()
def user(app: Devicehub) -> UserClient:
"""Gets a client with a logged-in dummy user."""
with app.app_context():
user = create_user()
client = UserClient(application=app,
response_wrapper=app.response_class,
email=user.email,
password='foo')
client.user, _ = client.login(client.email, client.password)
return client
def create_user(email='foo@foo.com', password='foo') -> User:
user = User(email=email, password=password)
db.session.add(user)
db.session.commit()
return user
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:
return yaml.load(f)

24
tests/files/basic.snapshot.yaml Normal file
View File

@ -0,0 +1,24 @@
uuid: 'f5efd26e-8754-46bc-87bf-fbccc39d60d9'
type: 'Snapshot'
version: '11.0'
software: 'Workbench'
condition:
appearance: 'A'
functionality: 'B'
labelling: True
bios: 'B'
elapsed: 4
device:
type: 'Microtower'
serialNumber: 'd1s'
model: 'd1ml'
manufacturer: 'd1mr'
components:
- type: 'GraphicCard'
serialNumber: 'gc1s'
model: 'gc1ml'
manufacturer: 'gc1mr'
- type: 'RamModule'
serialNumber: 'rm1s'
model: 'rm1ml'
manufacturer: 'rm1mr'

36
tests/test_auth.py Normal file
View File

@ -0,0 +1,36 @@
from uuid import uuid4
import pytest
from werkzeug.exceptions import Unauthorized
from ereuse_devicehub.client import UserClient, Client
from ereuse_devicehub.devicehub import Devicehub
from tests.conftest import create_user
def test_authenticate_success(app: Devicehub):
"""Checks the authenticate method."""
with app.app_context():
user = create_user()
response_user = app.auth.authenticate(token=str(user.token))
assert response_user == user
def test_authenticate_error(app: Devicehub):
"""Tests the authenticate method with wrong token values."""
with app.app_context():
MESSAGE = 'Provide a suitable token.'
create_user()
# Token doesn't exist
with pytest.raises(Unauthorized, message=MESSAGE):
app.auth.authenticate(token=str(uuid4()))
# Wrong token format
with pytest.raises(Unauthorized, message=MESSAGE):
app.auth.authenticate(token='this is a wrong uuid')
def test_auth_view(user: UserClient, client: Client):
"""Tests authentication at endpoint / view."""
user.get(res='User', item=user.user['id'], status=200)
client.get(res='User', item=user.user['id'], status=Unauthorized)
client.get(res='User', item=user.user['id'], token='wrong token', status=Unauthorized)

16
tests/test_basic.py
View File

@ -1,12 +1,14 @@
from datetime import datetime, timedelta
from uuid import uuid4
import pytest
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.device.models import Desktop, NetworkAdapter
from ereuse_devicehub.resources.event.models import Snapshot, SoftwareType, Appearance, \
Functionality, Bios
from ereuse_devicehub.resources.user.model import User
def test_dependencies():
with pytest.raises(ImportError):
# Simplejson has a different signature than stdlib json
# should be fixed though
# noinspection PyUnresolvedReferences
import simplejson
# noinspection PyArgumentList

14
tests/test_device.py
View File

@ -1,9 +1,13 @@
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.device.models import Desktop, GraphicCard, NetworkAdapter, Device
from ereuse_devicehub.resources.device.models import Desktop, Device, GraphicCard, NetworkAdapter
from ereuse_devicehub.resources.device.schemas import Device as DeviceS
def test_device_model(app: Devicehub):
"""
Tests that the correctness of the device model and its relationships.
"""
with app.test_request_context():
pc = Desktop(model='p1mo', manufacturer='p1ma', serial_number='p1s')
pc.components = components = [
@ -38,3 +42,11 @@ def test_device_model(app: Devicehub):
assert NetworkAdapter.query.first() is not None, 'We removed the network adaptor'
assert gcard.id == 3, 'We should still hold a reference to a zombie graphic card'
assert GraphicCard.query.first() is None, 'We should have deleted it –it was inside the pc'
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.dump({'id': 1})

48
tests/test_event.py
View File

@ -1,48 +0,0 @@
from datetime import datetime, timedelta
from uuid import uuid4
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.device.models import Microtower, Device
from ereuse_devicehub.resources.event.models import Snapshot, SoftwareType, Appearance, \
Functionality, Bios, SnapshotRequest, TestHardDrive, StressTest
from ereuse_devicehub.resources.user.model import User
# noinspection PyArgumentList
def test_event_model(app: Devicehub):
"""
Tests creating a Snapshot with its relationships ensuring correct
DB mapping.
"""
with app.test_request_context():
user = User(email='foo@bar.com')
device = Microtower(serial_number='a1')
snapshot = Snapshot(uuid=uuid4(),
date=datetime.now(),
version='1.0',
snapshot_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'})
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

61
tests/test_snapshot.py Normal file
View File

@ -0,0 +1,61 @@
from datetime import datetime, timedelta
from uuid import uuid4
from ereuse_devicehub.client import UserClient
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.device.models import Device, Microtower
from ereuse_devicehub.resources.event.models import Appearance, Bios, Functionality, Snapshot, \
SnapshotRequest, SoftwareType
from ereuse_devicehub.resources.user.models import User
from tests.conftest import file
def test_snapshot_model(app: Devicehub):
"""
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'})
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):
with app.app_context():
s = file('basic.snapshot')
app.resources['Snapshot'].schema.load(s)
def test_snapshot_post(user: UserClient):
"""Tests the post snapshot endpoint (validation, etc)."""
s = file('basic.snapshot')
snapshot, _ = user.post(s, res=Snapshot.__name__)

82
tests/test_user.py Normal file
View File

@ -0,0 +1,82 @@
from base64 import b64decode
from uuid import UUID
from sqlalchemy_utils import Password
from werkzeug.exceptions import NotFound, Unauthorized, UnprocessableEntity
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.models import User
from tests.conftest import create_user
def test_create_user_method(app: Devicehub):
"""
Tests creating an user through the main method.
This method checks that the token is correct, too.
"""
with app.app_context():
user_def = app.resources['User'] # type: UserDef
u = user_def.create_user(email='foo@foo.com', password='foo')
user = User.query.filter_by(id=u['id']).one() # type: User
assert user.email == 'foo@foo.com'
assert isinstance(user.token, UUID)
assert User.query.filter_by(email='foo@foo.com').one() == user
def test_create_user_email_insensitive(app: Devicehub):
"""Ensures email is case insensitive."""
with app.app_context():
user = User(email='FOO@foo.com')
db.session.add(user)
# We search in case insensitive manner
u1 = User.query.filter_by(email='foo@foo.com').one()
assert u1 == user
assert u1.email == 'FOO@foo.com'
def test_hash_password(app: Devicehub):
"""Tests correct password hashing and equaling."""
with app.app_context():
user = create_user()
assert isinstance(user.password, Password)
assert user.password == 'foo'
def test_login_success(client: Client, app: Devicehub):
"""
Tests successfully performing login.
This checks that:
- User is returned.
- User has token.
- User has not the password.
"""
with app.app_context():
create_user()
user, _ = client.post({'email': 'foo@foo.com', 'password': 'foo'},
uri='/users/login',
status=200)
assert user['email'] == 'foo@foo.com'
assert UUID(b64decode(user['token'].encode()).decode()[:-1])
assert 'password' not in user
def test_login_failure(client: Client, app: Devicehub):
"""Tests performing wrong login."""
# Wrong password
with app.app_context():
create_user()
client.post({'email': 'foo@foo.com', 'password': 'wrong pass'},
uri='/users/login',
status=Unauthorized)
# Wrong URI
client.post({}, uri='/wrong-uri', status=NotFound)
# Malformed data
client.post({}, uri='/users/login', status=UnprocessableEntity)
client.post({'email': 'this is not an email', 'password': 'nope'},
uri='/users/login',
status=UnprocessableEntity)