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

Integrate with rate

This commit is contained in:
Xavier Bustamante Talavera 2018-07-14 16:41:22 +02:00
parent 8f5b93739a
commit c1a3b23d8b
25 changed files with 974 additions and 391 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
docs/events.rst
View File

@ -2,7 +2,7 @@ Events
######
.. toctree::
:maxdepth: 4
:maxdepth: 4
event-diagram
@ -12,8 +12,9 @@ Rate
Devicehub generates an rating for a device taking into consideration the
visual, functional, and performance.
.. todo:: add performance as a result of component fusion + general tests in `here <https://
github.com/eReuse/Rdevicescore/blob/master/img/input_process_output.png>`_.
.. todo:: add performance as a result of component fusion + general
tests in `here <https://github.com/eReuse/Rdevicescore/blob/master/
img/input_process_output.png>`_.
A Workflow is as follows:
@ -27,28 +28,29 @@ A Workflow is as follows:
3. Devicehub aggregates different rates and computes a final score for
the device by performing a new ``AggregateRating`` event.
There are two **types** of ``Rate``: ``WorkbenchRate`` and
``PhotoboxRate``. Moreover, each rate can have different **versions**,
or different revisions of the algorithm used to compute the final score,
and Devicehub generates a rate event for **each** version. So, if
an agent fulfills a ``WorkbenchRate`` and there are 3 versions, Devicehub
generates 3 ``WorkbenchRate``. Devicehub understands that only one
version is the **official** and it will generate an ``AggregateRating``
only from the **official** version.
There are three **types** of ``Rate``: ``WorkbenchRate``,
``AppRate``, and ``PhotoboxRate``. ``WorkbenchRate`` can have different
**software** algorithms, and each software algorithm can have several
**versions**. So, we have 3 dimensions for ``WorkbenchRate``:
type, software, version.
.. todo:: we should be able to disable a version without destroying code
In the future, Devicehub will be able to use different and independent
algorithms to calculate a ``Rate`` (not only changed by versions).
Devicehub generates a rate event for each software and version. So,
if an agent fulfills a ``WorkbenchRate`` and there are 2 software
algorithms and each has two versions, Devicehub will generate 4 rates.
Devicehub understands that only one software and version are the
**oficial** (set in the settings of each inventory),
and it will generate an ``AggregateRating`` for only the official
versions. At the same time, ``Price`` only computes the price of
the **oficial** version.
The technical Workflow in Devicehub is as follows:
1. In **T1**, the user performs a ``Snapshot`` by processing the device
through the Workbench. From the benchmarks and the visual and
functional ratings the user does in the device, the system generates
a ``WorkbenchRate``. With only this information,
the system generates an ``AggregateRating``, which is the event
that the user will see in the web.
many ``WorkbenchRate`` (as many as software and versions defined).
With only this information, the system generates an ``AggregateRating``,
which is the event that the user will see in the web.
2. In **T2**, the user takes pictures from the device through the
Photobox, and DeviceHub crates an ``ImageSet`` with multiple
``Image`` with information from the photobox.
@ -72,6 +74,17 @@ The same ``ImageSet`` can be rated multiple times, generating a new
.. todo:: which info does photobox provide for each picture?
Price
*****
Price states a selling price for the device, but not necessariliy the
final price this was sold (which is set in the Sell event).
Devicehub automatically computes a price from ``AggregateRating``
events. As in a **Rate**, price can have **software** and **version**,
and there is an **official** price that is used to automatically
compute the price from an ``AggregateRating``. Only the official price
is computed from an ``AggregateRating``.
Snapshot
********
The Snapshot sets the physical information of the device (S/N, model...)
@ -175,10 +188,10 @@ There are four events for getting rid of devices:
been recovered under a new product.
.. note:: For usability purposes, users might not directly perform
``Dispose``, but this could automatically be done when
``Dispose``, but this could automatically be done when
performing ``ToDispose`` + ``Receive`` to a ``RecyclingCenter``.
.. todo:: Ensure that ``Dispose`` is a ``Trade`` event. An Org could
``Sell`` or ``Donate`` a device with the objective of disposing them.
``Sell`` or ``Donate`` a device with the objective of disposing them.
Is ``Dispose`` ok, or do we want to keep that extra ``Sell`` or
``Donate`` event? Could dispose be a synonym of any of those?

33
ereuse_devicehub/config.py
View File

@ -4,18 +4,21 @@ from typing import Set
from ereuse_devicehub.resources.device import CellphoneDef, ComponentDef, ComputerDef, \
ComputerMonitorDef, DataStorageDef, DesktopDef, DeviceDef, DisplayDef, GraphicCardDef, \
HardDriveDef, LaptopDef, MobileDef, MonitorDef, MotherboardDef, NetworkAdapterDef, \
ProcessorDef, RamModuleDef, ServerDef, SmartphoneDef, SolidStateDriveDef, TabletDef, \
TelevisionSetDef, SoundCardDef
ProcessorDef, RamModuleDef, ServerDef, SmartphoneDef, SolidStateDriveDef, SoundCardDef, \
TabletDef, TelevisionSetDef
from ereuse_devicehub.resources.enums import PriceSoftware, RatingSoftware
from ereuse_devicehub.resources.event import AddDef, AggregateRateDef, AppRateDef, \
BenchmarkDataStorageDef, BenchmarkDef, BenchmarkProcessorDef, BenchmarkProcessorSysbenchDef, \
BenchmarkRamSysbenchDef, BenchmarkWithRateDef, EraseBasicDef, EraseSectorsDef, EventDef, \
InstallDef, PhotoboxSystemRateDef, PhotoboxUserDef, RateDef, RemoveDef, SnapshotDef, StepDef, \
StepRandomDef, StepZeroDef, StressTestDef, TestDataStorageDef, TestDef, WorkbenchRateDef
BenchmarkRamSysbenchDef, BenchmarkWithRateDef, EraseBasicDef, EraseSectorsDef, EreusePriceDef, \
EventDef, InstallDef, PhotoboxSystemRateDef, PhotoboxUserDef, PriceDef, RateDef, RemoveDef, \
SnapshotDef, StepDef, StepRandomDef, StepZeroDef, StressTestDef, TestDataStorageDef, TestDef, \
WorkbenchRateDef
from ereuse_devicehub.resources.inventory import InventoryDef
from ereuse_devicehub.resources.tag import TagDef
from ereuse_devicehub.resources.user import OrganizationDef, UserDef
from teal.auth import TokenAuth
from teal.config import Config
from teal.currency import Currency
class DevicehubConfig(Config):
@ -27,16 +30,18 @@ class DevicehubConfig(Config):
UserDef,
OrganizationDef, TagDef, EventDef, AddDef, RemoveDef, EraseBasicDef, EraseSectorsDef,
StepDef, StepZeroDef, StepRandomDef, RateDef, AggregateRateDef, WorkbenchRateDef,
PhotoboxUserDef, PhotoboxSystemRateDef, InstallDef, SnapshotDef, TestDef,
PhotoboxUserDef, PhotoboxSystemRateDef, PriceDef, EreusePriceDef,
InstallDef, SnapshotDef, TestDef,
TestDataStorageDef, StressTestDef, WorkbenchRateDef, InventoryDef, BenchmarkDef,
BenchmarkDataStorageDef, BenchmarkWithRateDef, AppRateDef, BenchmarkProcessorDef,
BenchmarkProcessorSysbenchDef, BenchmarkRamSysbenchDef
}
PASSWORD_SCHEMES = {'pbkdf2_sha256'} # type: Set[str]
SQLALCHEMY_DATABASE_URI = 'postgresql://dhub:ereuse@localhost/devicehub' # type: str
SCHEMA = 'dhub'
MIN_WORKBENCH = StrictVersion('11.0a1') # type: StrictVersion
"""
the minimum algorithm_version of ereuse.org workbench that this devicehub
the minimum version of ereuse.org workbench that this devicehub
accepts. we recommend not changing this value.
"""
ORGANIZATION_NAME = None # type: str
@ -55,6 +60,20 @@ class DevicehubConfig(Config):
}
API_DOC_CLASS_DISCRIMINATOR = 'type'
WORKBENCH_RATE_SOFTWARE = RatingSoftware.ECost
WORKBENCH_RATE_VERSION = StrictVersion('1.0')
PHOTOBOX_RATE_SOFTWARE = RatingSoftware.ECost
PHOTOBOX_RATE_VERSION = StrictVersion('1.0')
"""
Official versions for WorkbenchRate and PhotoboxRate
"""
PRICE_SOFTWARE = PriceSoftware.Ereuse
PRICE_VERSION = StrictVersion('1.0')
PRICE_CURRENCY = Currency.EUR
"""
Official versions
"""
def __init__(self, db: str = None) -> None:
if not self.ORGANIZATION_NAME or not self.ORGANIZATION_TAX_ID:
raise ValueError('You need to set the main organization parameters.')

10
ereuse_devicehub/db.py
View File

@ -1,3 +1,11 @@
from teal.db import SQLAlchemy
from teal.db import SQLAlchemy as _SQLAlchemy
class SQLAlchemy(_SQLAlchemy):
def drop_all(self, bind='__all__', app=None):
"""A faster nuke-like option to drop everything."""
self.drop_schema()
self.drop_schema(schema='common')
db = SQLAlchemy(session_options={"autoflush": False})

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

@ -2,15 +2,15 @@ from contextlib import suppress
from itertools import groupby
from typing import Iterable, Set
from sqlalchemy import inspect
from sqlalchemy.exc import IntegrityError
from sqlalchemy.util import OrderedSet
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 Remove
from ereuse_devicehub.resources.tag.model import Tag
from sqlalchemy import inspect
from sqlalchemy.exc import IntegrityError
from sqlalchemy.util import OrderedSet
from teal.db import ResourceNotFound
from teal.marshmallow import ValidationError
@ -48,7 +48,7 @@ class Sync:
:return: A tuple of:
1. The device from the database (with an ID) whose
``components`` field contain the db algorithm_version
``components`` field contain the db version
of the passed-in components.
2. A list of Add / Remove (not yet added to session).
"""
@ -124,7 +124,7 @@ class Sync:
This method tries to get an existing device using the HID
or one of the tags, and...
- if it already exists it returns a "local synced algorithm_version"
- if it already exists it returns a "local synced version"
–the same ``device`` you passed-in but with updated values
from the database. In this case we do not
"touch" any of its values on the DB.

25
ereuse_devicehub/resources/enums.py
View File

@ -5,7 +5,7 @@ from typing import Union
@unique
class SnapshotSoftware(Enum):
"""The algorithm_software used to perform the Snapshot."""
"""The software used to perform the Snapshot."""
Workbench = 'Workbench'
AndroidApp = 'AndroidApp'
Web = 'Web'
@ -14,8 +14,16 @@ class SnapshotSoftware(Enum):
@unique
class RatingSoftware(Enum):
"""The algorithm_software used to compute the Score."""
Ereuse = 'Ereuse'
"""The software used to compute the Score."""
ECost = 'ECost'
"""
The eReuse.org rate algorithm that focuses maximizing refurbishment
of devices in general, specially penalizing very low and very high
devices in order to stimulate medium-range devices.
This model is cost-oriented.
"""
EMarket = 'EMarket'
RATE_POSITIVE = 0, 10
@ -48,13 +56,18 @@ class RatingRange(IntEnum):
return cls.HIGH
@unique
class PriceSoftware(Enum):
Ereuse = 'Ereuse'
@unique
class AggregateRatingVersions(Enum):
v1 = StrictVersion('1.0')
"""
This algorithm_version is set to aggregate :class:`ereuse_devicehub.resources.
event.models.WorkbenchRate` algorithm_version X and :class:`ereuse_devicehub.
resources.event.models.PhotoboxRate` algorithm_version Y.
This version is set to aggregate :class:`ereuse_devicehub.resources.
event.models.WorkbenchRate` version X and :class:`ereuse_devicehub.
resources.event.models.PhotoboxRate` version Y.
"""

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

@ -3,8 +3,8 @@ from typing import Callable, Iterable, Tuple
from ereuse_devicehub.resources.device.sync import Sync
from ereuse_devicehub.resources.event.schemas import Add, AggregateRate, AppRate, Benchmark, \
BenchmarkDataStorage, BenchmarkProcessor, BenchmarkProcessorSysbench, BenchmarkRamSysbench, \
BenchmarkWithRate, EraseBasic, EraseSectors, Event, Install, PhotoboxSystemRate, \
PhotoboxUserRate, Rate, Remove, Snapshot, Step, StepRandom, StepZero, StressTest, Test, \
BenchmarkWithRate, EraseBasic, EraseSectors, EreusePrice, Event, Install, PhotoboxSystemRate, \
PhotoboxUserRate, Price, Rate, Remove, Snapshot, Step, StepRandom, StepZero, StressTest, Test, \
TestDataStorage, WorkbenchRate
from ereuse_devicehub.resources.event.views import EventView, SnapshotView
from teal.resource import Converters, Resource
@ -82,6 +82,16 @@ class AppRateDef(RateDef):
SCHEMA = AppRate
class PriceDef(EventDef):
VIEW = None
SCHEMA = Price
class EreusePriceDef(EventDef):
VIEW = None
SCHEMA = EreusePrice
class InstallDef(EventDef):
VIEW = None
SCHEMA = Install

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

@ -3,23 +3,26 @@ from datetime import timedelta
from typing import Set, Union
from uuid import uuid4
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.device.models import Component, Computer, DataStorage, Device
from ereuse_devicehub.resources.enums import AppearanceRange, BOX_RATE_3, BOX_RATE_5, Bios, \
FunctionalityRange, RATE_NEGATIVE, RATE_POSITIVE, RatingRange, RatingSoftware, \
SnapshotExpectedEvents, SnapshotSoftware, TestHardDriveLength
from ereuse_devicehub.resources.image.models import Image
from ereuse_devicehub.resources.models import STR_BIG_SIZE, STR_SIZE, STR_SM_SIZE, Thing
from ereuse_devicehub.resources.user.models import User
from flask import g
from flask import current_app as app, g
from sqlalchemy import BigInteger, Boolean, CheckConstraint, Column, DateTime, Enum as DBEnum, \
Float, ForeignKey, Interval, JSON, SmallInteger, Unicode, event
Float, ForeignKey, Interval, JSON, SmallInteger, Unicode, event, orm
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.orderinglist import ordering_list
from sqlalchemy.orm import backref, relationship, validates
from sqlalchemy.orm.events import AttributeEvents as Events
from sqlalchemy.util import OrderedSet
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.device.models import Component, Computer, DataStorage, Desktop, \
Device, Laptop, Server
from ereuse_devicehub.resources.enums import AppearanceRange, BOX_RATE_3, BOX_RATE_5, Bios, \
FunctionalityRange, PriceSoftware, RATE_NEGATIVE, RATE_POSITIVE, RatingRange, RatingSoftware, \
SnapshotExpectedEvents, SnapshotSoftware, TestHardDriveLength
from ereuse_devicehub.resources.image.models import Image
from ereuse_devicehub.resources.models import STR_BIG_SIZE, STR_SIZE, STR_SM_SIZE, Thing
from ereuse_devicehub.resources.user.models import User
from teal.currency import Currency
from teal.db import ArrayOfEnum, CASCADE, CASCADE_OWN, INHERIT_COND, POLYMORPHIC_ID, \
POLYMORPHIC_ON, StrictVersionType, check_range
@ -279,8 +282,8 @@ class SnapshotRequest(db.Model):
class Rate(JoinedTableMixin, EventWithOneDevice):
rating = Column(Float(decimal_return_scale=2), check_range('rating', *RATE_POSITIVE))
algorithm_software = Column(DBEnum(RatingSoftware), nullable=False)
algorithm_version = Column(StrictVersionType, nullable=False)
software = Column(DBEnum(RatingSoftware))
version = Column(StrictVersionType)
appearance = Column(Float(decimal_return_scale=2), check_range('appearance', *RATE_NEGATIVE))
functionality = Column(Float(decimal_return_scale=2),
check_range('functionality', *RATE_NEGATIVE))
@ -349,6 +352,17 @@ class WorkbenchRate(ManualRate):
check_range('graphic_card', *RATE_POSITIVE))
bios = Column(DBEnum(Bios))
# todo ensure for WorkbenchRate version and software are not None when inserting them
def ratings(self) -> Set['WorkbenchRate']:
"""
Computes all the possible rates taking this rating as a model.
Returns a set of ratings, including this one, which is mutated.
"""
from ereuse_rate.main import main
return main(self, **app.config.get_namespace('WORKBENCH_RATE_'))
class AppRate(ManualRate):
pass
@ -387,6 +401,102 @@ class PhotoboxSystemRate(PhotoboxRate):
id = Column(UUID(as_uuid=True), ForeignKey(PhotoboxRate.id), primary_key=True)
class Price(JoinedTableMixin, EventWithOneDevice):
currency = Column(DBEnum(Currency), nullable=False)
price = Column(Float(decimal_return_scale=2), check_range('price', 0), nullable=False)
software = Column(DBEnum(PriceSoftware))
version = Column(StrictVersionType)
rating_id = Column(UUID(as_uuid=True), ForeignKey(AggregateRate.id))
rating = relationship(AggregateRate,
backref=backref('price',
lazy=True,
cascade=CASCADE_OWN,
uselist=False),
primaryjoin=AggregateRate.id == rating_id)
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self.currency = self.currency or app.config['PRICE_CURRENCY']
class EreusePrice(Price):
"""A Price class that auto-computes its amount by"""
MULTIPLIER = {
Desktop: 20,
Laptop: 30
}
class Type:
def __init__(self, percentage, price) -> None:
# see https://stackoverflow.com/a/29651462 for the - 0.005
self.amount = round(price * percentage - 0.005, 2)
self.percentage = round(percentage - 0.005, 2)
class Service:
REFURBISHER, PLATFORM, RETAILER = 0, 1, 2
STANDARD, WARRANTY2 = 'STD', 'WR2'
SCHEMA = {
Desktop: {
RatingRange.HIGH: {
STANDARD: (0.35125, 0.204375, 0.444375),
WARRANTY2: (0.47425, 0.275875, 0.599875)
},
RatingRange.MEDIUM: {
STANDARD: (0.385, 0.2558333333, 0.3591666667),
WARRANTY2: (0.539, 0.3581666667, 0.5028333333)
},
RatingRange.LOW: {
STANDARD: (0.5025, 0.30875, 0.18875),
},
},
Laptop: {
RatingRange.HIGH: {
STANDARD: (0.3469230769, 0.195, 0.4580769231),
WARRANTY2: (0.4522307692, 0.2632307692, 0.6345384615)
},
RatingRange.MEDIUM: {
STANDARD: (0.382, 0.1735, 0.4445),
WARRANTY2: (0.5108, 0.2429, 0.6463)
},
RatingRange.LOW: {
STANDARD: (0.4528571429, 0.2264285714, 0.3207142857),
}
}
}
SCHEMA[Server] = SCHEMA[Desktop]
def __init__(self, device, rating_range, role, price) -> None:
cls = device.__class__ if device.__class__ != Server else Desktop
rate = self.SCHEMA[cls][rating_range]
self.standard = EreusePrice.Type(rate['STD'][role], price)
self.warranty2 = EreusePrice.Type(rate['WR2'][role], price)
def __init__(self, rating: AggregateRate, **kwargs) -> None:
if rating.rating_range == RatingRange.VERY_LOW:
raise ValueError('Cannot compute price for Range.VERY_LOW')
self.price = round(rating.rating * self.MULTIPLIER[rating.device.__class__], 2)
super().__init__(rating=rating, device=rating.device, **kwargs)
self._compute()
self.software = self.software or app.config['PRICE_SOFTWARE']
self.version = self.version or app.config['PRICE_VERSION']
@orm.reconstructor
def _compute(self):
"""
Calculates eReuse.org prices when initializing the
instance from the price and other properties.
"""
self.refurbisher = self._service(self.Service.REFURBISHER)
self.retailer = self._service(self.Service.RETAILER)
self.platform = self._service(self.Service.PLATFORM)
self.warranty2 = round(self.refurbisher.warranty2.amount
+ self.retailer.warranty2.amount
+ self.platform.warranty2.amount, 2)
def _service(self, role):
return self.Service(self.device, self.rating.rating_range, role, self.price)
class Test(JoinedTableMixin, EventWithOneDevice):
elapsed = Column(Interval, nullable=False)
@ -474,6 +584,9 @@ class BenchmarkRamSysbench(BenchmarkWithRate):
# Listeners
# Listeners validate values and keep relationships synced
# The following listeners avoids setting values to events that
# do not make sense. For example, EraseBasic to a graphic card.
@event.listens_for(TestDataStorage.device, Events.set.__name__, propagate=True)
@event.listens_for(Install.device, Events.set.__name__, propagate=True)
@event.listens_for(EraseBasic.device, Events.set.__name__, propagate=True)

35
ereuse_devicehub/resources/event/models.pyi
View File

@ -1,14 +1,15 @@
from datetime import datetime, timedelta
from distutils.version import StrictVersion
from typing import List, Set
from typing import Dict, List, Set
from uuid import UUID
from sqlalchemy import Column
from sqlalchemy.orm import relationship
from sqlalchemy_utils import Currency
from ereuse_devicehub.resources.device.models import Component, Computer, Device
from ereuse_devicehub.resources.enums import AppearanceRange, Bios, FunctionalityRange, \
RatingSoftware, SnapshotExpectedEvents, SnapshotSoftware, TestHardDriveLength
PriceSoftware, RatingSoftware, SnapshotExpectedEvents, SnapshotSoftware, TestHardDriveLength
from ereuse_devicehub.resources.image.models import Image
from ereuse_devicehub.resources.models import Thing
from ereuse_devicehub.resources.user import User
@ -127,8 +128,8 @@ class Rate(EventWithOneDevice):
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self.rating = ... # type: float
self.algorithm_software = ... # type: RatingSoftware
self.algorithm_version = ... # type: StrictVersion
self.software = ... # type: RatingSoftware
self.version = ... # type: StrictVersion
self.appearance = ... # type: float
self.functionality = ... # type: float
self.rating_range = ... # type: str
@ -144,6 +145,7 @@ class AggregateRate(Rate):
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self.ratings = ... # type: Set[IndividualRate]
self.price = ... # type: Price
class ManualRate(IndividualRate):
@ -193,6 +195,31 @@ class PhotoboxSystemRate(PhotoboxRate):
pass
class Price(EventWithOneDevice):
currency = ... # type: Column
price = ... # type: Column
software = ... # type: Column
version = ... # type: Column
rating_id = ... # type: Column
rating = ... # type: relationship
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self.currency = ... # type: Currency
self.price = ... # type: float
self.software = ... # type: PriceSoftware
self.version = ... # type: StrictVersion
self.rating_id = ... # type: UUID
self.rating = ... # type: AggregateRate
class EreusePrice(Price):
MULTIPLIER = ... # type: Dict
def __init__(self, rating: AggregateRate, **kwargs) -> None:
super().__init__(**kwargs)
class Test(EventWithOneDevice):
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)

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

@ -1,5 +1,5 @@
from flask import current_app as app
from marshmallow import ValidationError, validates_schema
from marshmallow import Schema as MarshmallowSchema, ValidationError, validates_schema
from marshmallow.fields import Boolean, DateTime, Float, Integer, List, Nested, String, TimeDelta, \
UUID
from marshmallow.validate import Length, Range
@ -7,11 +7,13 @@ from marshmallow.validate import Length, Range
from ereuse_devicehub.marshmallow import NestedOn
from ereuse_devicehub.resources.device.schemas import Component, Device
from ereuse_devicehub.resources.enums import AppearanceRange, Bios, FunctionalityRange, \
RATE_POSITIVE, RatingSoftware, SnapshotExpectedEvents, SnapshotSoftware, TestHardDriveLength
PriceSoftware, RATE_POSITIVE, RatingSoftware, SnapshotExpectedEvents, SnapshotSoftware, \
TestHardDriveLength
from ereuse_devicehub.resources.event import models as m
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.currency import Currency
from teal.marshmallow import EnumField, Version
from teal.resource import Schema
@ -91,13 +93,11 @@ class Rate(EventWithOneDevice):
dump_only=True,
data_key='ratingValue',
description='The rating for the content.')
algorithm_software = EnumField(RatingSoftware,
software = EnumField(RatingSoftware,
dump_only=True,
data_key='algorithmSoftware',
description='The algorithm used to produce this rating.')
algorithm_version = Version(dump_only=True,
data_key='algorithmVersion',
description='The algorithm_version of the algorithm_software.')
version = Version(dump_only=True,
description='The version of the software.')
appearance = Integer(validate=Range(-3, 5), dump_only=True)
functionality = Integer(validate=Range(-3, 5),
dump_only=True,
@ -141,7 +141,7 @@ class ManualRate(IndividualRate):
functionality_range = EnumField(FunctionalityRange,
required=True,
data_key='functionalityRange',
description='Grades the defects of a device that affect its usage.')
description='Grades the defects of a device affecting usage.')
labelling = Boolean(description='Sets if there are labels stuck that should be removed.')
@ -158,6 +158,29 @@ class WorkbenchRate(ManualRate):
'boot from the network.')
class Price(EventWithOneDevice):
currency = EnumField(Currency, required=True)
price = Float(required=True)
software = EnumField(PriceSoftware, dump_only=True)
version = Version(dump_only=True)
rating = NestedOn(AggregateRate, dump_only=True)
class EreusePrice(Price):
class Service(MarshmallowSchema):
class Type(MarshmallowSchema):
amount = Float()
percentage = Float()
standard = Nested(Type)
warranty2 = Nested(Type)
warranty2 = Float()
refurbisher = Nested(Service)
retailer = Nested(Service)
platform = Nested(Service)
class Install(EventWithOneDevice):
name = String(validate=Length(min=4, max=STR_BIG_SIZE),
required=True,
@ -198,7 +221,7 @@ class Snapshot(EventWithOneDevice):
if data['software'] == SnapshotSoftware.Workbench:
if data['version'] < app.config['MIN_WORKBENCH']:
raise ValidationError(
'Min. supported Workbench algorithm_version is '
'Min. supported Workbench version is '
'{}'.format(app.config['MIN_WORKBENCH']),
field_names=['version']
)

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

@ -1,3 +1,4 @@
from contextlib import suppress
from distutils.version import StrictVersion
from typing import List
from uuid import UUID
@ -7,8 +8,8 @@ from sqlalchemy.util import OrderedSet
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.device.models import Component, Computer
from ereuse_devicehub.resources.enums import RatingSoftware, SnapshotSoftware
from ereuse_devicehub.resources.event.models import Event, ManualRate, Snapshot, WorkbenchRate
from ereuse_devicehub.resources.enums import SnapshotSoftware
from ereuse_devicehub.resources.event.models import Event, Snapshot, WorkbenchRate
from teal.resource import View
@ -51,21 +52,11 @@ class SnapshotView(View):
assert all(not c.events_one for c in components) if components else True
db_device, remove_events = self.resource_def.sync.run(device, components)
snapshot.device = db_device
snapshot.events |= remove_events | events_device
snapshot.events |= remove_events | events_device # Set events to snapshot
# commit will change the order of the components by what
# the DB wants. Let's get a copy of the list so we preserve order
ordered_components = OrderedSet(x for x in snapshot.components)
for event in events_device:
if isinstance(event, ManualRate):
event.algorithm_software = RatingSoftware.Ereuse
event.algorithm_version = StrictVersion('1.0')
if isinstance(event, WorkbenchRate):
# todo process workbench rate
event.data_storage = 2
event.graphic_card = 4
event.processor = 1
# Add the new events to the db-existing devices and components
db_device.events_one |= events_device
if components:
@ -73,6 +64,13 @@ class SnapshotView(View):
component.events_one |= events
snapshot.events |= events
# Compute ratings
with suppress(StopIteration):
# todo are we sure we want to have snapshots without rates?
snapshot.events |= next(
e.ratings() for e in events_device if isinstance(e, WorkbenchRate)
)
db.session.add(snapshot)
db.session.commit()
# todo we are setting snapshot dirty again with this components but

5
ereuse_devicehub/resources/models.py
View File

@ -18,3 +18,8 @@ class Thing(db.Model):
created.comment = """
When Devicehub created this.
"""
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
if not self.created:
self.created = datetime.utcnow()

5
setup.py
View File

@ -35,7 +35,7 @@ setup(
long_description=long_description,
long_description_content_type='text/markdown',
install_requires=[
'teal>=0.2.0a6',
'teal>=0.2.0a8',
'marshmallow_enum',
'ereuse-utils[Naming]>=0.4b1',
'psycopg2-binary',
@ -46,7 +46,8 @@ setup(
'click-spinner',
'sqlalchemy-utils[password, color, babel]',
'PyYAML',
'python-stdnum'
'python-stdnum',
'ereuse-rate==0.0.2'
],
extras_require={
'docs': [

17
tests/conftest.py
View File

@ -2,6 +2,7 @@ from pathlib import Path
import pytest
import yaml
from sqlalchemy.exc import ProgrammingError
from ereuse_devicehub.client import Client, UserClient
from ereuse_devicehub.config import DevicehubConfig
@ -31,10 +32,20 @@ def _app(config: TestConfig) -> Devicehub:
@pytest.fixture()
def app(request, _app: Devicehub) -> Devicehub:
with _app.app_context():
_app.init_db()
# More robust than 'yield'
request.addfinalizer(lambda *args, **kw: db.drop_all(app=_app))
def _drop(*args, **kwargs):
with _app.app_context():
db.drop_all()
with _app.app_context():
try:
_app.init_db()
except ProgrammingError:
print('Database was not correctly emptied. Re-empty and re-installing...')
_drop()
_app.init_db()
request.addfinalizer(_drop)
return _app

2
tests/files/2-second-device-with-components-of-first.snapshot.yaml
View File

@ -2,7 +2,7 @@ device:
manufacturer: p2m
serialNumber: p2s
model: p2
type: Computer
type: Desktop
chassis: Microtower
components:
- manufacturer: p2c1m

11
tests/files/basic.snapshot.yaml
View File

@ -4,7 +4,7 @@ version: '11.0'
software: Workbench
elapsed: 4
device:
type: Computer
type: Desktop
chassis: Microtower
serialNumber: d1s
model: d1ml
@ -24,3 +24,12 @@ components:
serialNumber: rm1s
model: rm1ml
manufacturer: rm1mr
speed: 1333
- type: Processor
serialNumber: p1s
model: p1ml
manufacturer: p1mr
speed: 1.6
events:
- type: BenchmarkProcessor
rate: 2410

156
tests/files/erase-sectors-2-hdd.snapshot.yaml Normal file
View File

@ -0,0 +1,156 @@
{
"version": "11.0a3",
"device": {
"serialNumber": null,
"manufacturer": null,
"model": null,
"type": "Desktop",
"events": [],
"chassis": "Tower"
},
"elapsed": 7631,
"software": "Workbench",
"type": "Snapshot",
"closed": false,
"uuid": "5387668a-8d21-4053-a1ac-36efb97fc3ea",
"expectedEvents": [
"TestDataStorage",
"EraseBasic"
],
"components": [
{
"serialNumber": null,
"threads": 2,
"manufacturer": "Intel Corp.",
"address": 64,
"model": "Intel Core i3-2100 CPU @ 3.10GHz",
"type": "Processor",
"events": [],
"cores": 2,
"speed": 1.6071410000000002
},
{
"manufacturer": "Intel Corporation",
"model": "6 Series/C200 Series Chipset Family High Definition Audio Controller",
"type": "SoundCard",
"events": [],
"serialNumber": null
},
{
"serialNumber": "8F17943",
"size": 4096,
"manufacturer": "Kingston",
"format": "DIMM",
"model": "9905403-038.A00LF",
"type": "RamModule",
"events": [],
"interface": "DDR3",
"speed": 1333.0
},
{
"manufacturer": "Realtek Semiconductor Co., Ltd.",
"model": "RTL8111/8168/8411 PCI Express Gigabit Ethernet Controller",
"type": "NetworkAdapter",
"events": [],
"serialNumber": "f4:6d:04:12:9b:85",
"speed": 1000
},
{
"serialNumber": "WD-WCAV29008961",
"size": 305245,
"manufacturer": "Western Digital",
"model": "WDC WD3200AAJS-2",
"type": "HardDrive",
"events": [
{
"endTime": "2018-07-13T11:54:55.100581",
"steps": [
{
"endTime": "2018-07-13T11:54:55.096491",
"type": "StepRandom",
"error": false,
"startTime": "2018-07-13T10:52:45.092981"
}
],
"type": "EraseBasic",
"error": false,
"zeros": false,
"startTime": "2018-07-13T10:52:45.092612"
},
{
"lifetime": 24658,
"assessment": false,
"elapsed": 131,
"length": "Short",
"offlineUncorrectable": 1,
"error": true,
"currentPendingSectorCount": 1,
"powerCycleCount": 1253,
"reallocatedSectorCount": 6,
"type": "TestDataStorage",
"status": "Completed: read failure"
}
],
"interface": "ATA"
},
{
"serialNumber": "WD-WCAV27984668",
"size": 305245,
"manufacturer": "Western Digital",
"model": "WDC WD3200AAJS-0",
"type": "HardDrive",
"events": [
{
"endTime": "2018-07-13T12:55:47.331586",
"steps": [
{
"endTime": "2018-07-13T12:55:47.326835",
"type": "StepRandom",
"error": false,
"startTime": "2018-07-13T11:54:55.100925"
}
],
"type": "EraseBasic",
"error": false,
"zeros": false,
"startTime": "2018-07-13T11:54:55.100667"
},
{
"lifetime": 21979,
"assessment": true,
"elapsed": 115,
"length": "Short",
"offlineUncorrectable": 0,
"error": false,
"currentPendingSectorCount": 0,
"powerCycleCount": 1956,
"reallocatedSectorCount": 0,
"type": "TestDataStorage",
"status": "Completed without error"
}
],
"interface": "ATA"
},
{
"serialNumber": null,
"manufacturer": "Intel Corporation",
"model": "2nd Generation Core Processor Family Integrated Graphics Controller",
"type": "GraphicCard",
"events": [],
"memory": 256.0
},
{
"pcmcia": 0,
"serial": 1,
"manufacturer": "ASUSTeK Computer INC.",
"model": "P8H61-M LE",
"type": "Motherboard",
"events": [],
"slots": 2,
"usb": 2,
"firewire": 0,
"serialNumber": "109192430003459"
}
],
"date": "2018-07-13T10:48:36.738398"
}

10
tests/files/erase-sectors.snapshot.yaml
View File

@ -4,7 +4,7 @@ version: '11.0'
software: Workbench
elapsed: 4
device:
type: Computer
type: Desktop
chassis: Microtower
serialNumber: pc1s
model: pc1ml
@ -28,10 +28,10 @@ components:
error: False
startTime: 2018-06-01T08:16:00
endTime: 2018-06-01T09:17:00
- type: GraphicCard
serialNumber: gc1s
model: gc1ml
manufacturer: gc1mr
- type: Processor
serialNumber: p1s
model: p1ml
manufacturer: p1mr
- type: RamModule
serialNumber: rm1s
model: rm1ml

155
tests/files/real-eee-1001pxd.snapshot.11.yaml Normal file
View File

@ -0,0 +1,155 @@
{
"components": [
{
"type": "NetworkAdapter",
"model": "AR9285 Wireless Network Adapter",
"serialNumber": "74:2f:68:8b:fd:c8",
"manufacturer": "Qualcomm Atheros",
"events": []
},
{
"type": "NetworkAdapter",
"model": "AR8152 v2.0 Fast Ethernet",
"serialNumber": "14:da:e9:42:f6:7c",
"manufacturer": "Qualcomm Atheros",
"speed": 100,
"events": []
},
{
"type": "Processor",
"cores": 1,
"address": 64,
"model": "Intel Atom CPU N455 @ 1.66GHz",
"serialNumber": null,
"manufacturer": "Intel Corp.",
"speed": 1.667,
"events": [
{
"type": "BenchmarkProcessorSysbench",
"rate": 164.0803,
"elapsed": 164
},
{
"type": "BenchmarkProcessor",
"rate": 6666.24,
"elapsed": 0
}
]
},
{
"type": "GraphicCard",
"model": "Atom Processor D4xx/D5xx/N4xx/N5xx Integrated Graphics Controller",
"serialNumber": null,
"memory": 256.0,
"manufacturer": "Intel Corporation",
"events": []
},
{
"type": "SoundCard",
"model": "NM10/ICH7 Family High Definition Audio Controller",
"serialNumber": null,
"manufacturer": "Intel Corporation",
"events": []
},
{
"type": "SoundCard",
"model": "USB 2.0 UVC VGA WebCam",
"serialNumber": "0x0001",
"manufacturer": "Azurewave",
"events": []
},
{
"type": "RamModule",
"format": "DIMM",
"model": null,
"size": 1024,
"interface": "DDR2",
"serialNumber": null,
"manufacturer": null,
"speed": 667.0,
"events": []
},
{
"type": "HardDrive",
"model": "HTS54322",
"size": 238475,
"interface": "ATA",
"serialNumber": "E2024242CV86HJ",
"manufacturer": "Hitachi",
"events": [
{
"type": "BenchmarkDataStorage",
"elapsed": 16,
"writeSpeed": 21.8,
"readSpeed": 66.2
},
{
"type": "TestDataStorage",
"length": "Short",
"elapsed": 2,
"error": true,
"status": "Unspecified Error. Self-test not started."
},
{
"type": "EraseBasic",
"steps": [
{
"type": "StepRandom",
"startTime": "2018-07-03T09:15:22.257059",
"error": false,
"endTime": "2018-07-03T10:32:11.843190"
}
],
"startTime": "2018-07-03T09:15:22.256074",
"error": false,
"zeros": false,
"endTime": "2018-07-03T10:32:11.848455"
}
]
},
{
"type": "Motherboard",
"serial": 1,
"firewire": 0,
"model": "1001PXD",
"slots": 2,
"pcmcia": 0,
"serialNumber": "Eee0123456789",
"usb": 5,
"manufacturer": "ASUSTeK Computer INC.",
"events": []
}
],
"elapsed": 4875,
"uuid": "c058e8d2-fb92-47cb-a4b7-522b75561135",
"version": "11.0a2",
"type": "Snapshot",
"software": "Workbench",
"date": "2018-07-03T09:10:57.034598",
"device": {
"type": "Laptop",
"model": "1001PXD",
"serialNumber": "B8OAAS048286",
"manufacturer": "ASUSTeK Computer INC.",
"chassis": "Netbook",
"events": [
{
"type": "BenchmarkRamSysbench",
"rate": 15.7188,
"elapsed": 16
},
{
"type": "StressTest",
"error": false,
"elapsed": 60
}
]
},
"expectedEvents": [
"Benchmark",
"SmartTest",
"StressTest",
"EraseBasic"
],
"closed": false
}

2
tests/test_basic.py
View File

@ -35,4 +35,4 @@ def test_api_docs(client: Client):
'scheme': 'basic',
'name': 'Authorization'
}
assert 52 == len(docs['definitions'])
assert 54 == len(docs['definitions'])

7
tests/test_dummy.py Normal file
View File

@ -0,0 +1,7 @@
from ereuse_devicehub.devicehub import Devicehub
def test_dummy(_app: Devicehub):
"""Tests the dummy cli command."""
runner = _app.test_cli_runner()
runner.invoke(args=['dummy', '--yes'], catch_exceptions=False)

13
tests/test_event.py
View File

@ -1,14 +1,15 @@
from datetime import datetime, timedelta
import pytest
from flask import g
from sqlalchemy.util import OrderedSet
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.device.models import Computer, Device, GraphicCard, HardDrive, \
from ereuse_devicehub.resources.device.models import Desktop, Device, GraphicCard, HardDrive, \
RamModule, SolidStateDrive
from ereuse_devicehub.resources.enums import TestHardDriveLength
from ereuse_devicehub.resources.event.models import BenchmarkDataStorage, EraseBasic, EraseSectors, \
EventWithOneDevice, Install, Ready, StepRandom, StepZero, StressTest, TestDataStorage
from flask import g
from sqlalchemy.util import OrderedSet
from tests.conftest import create_user
@ -117,7 +118,7 @@ def test_install():
@pytest.mark.usefixtures('auth_app_context')
def test_update_components_event_one():
computer = Computer(serial_number='sn1', model='ml1', manufacturer='mr1')
computer = Desktop(serial_number='sn1', model='ml1', manufacturer='mr1')
hdd = HardDrive(serial_number='foo', manufacturer='bar', model='foo-bar')
computer.components.add(hdd)
@ -142,7 +143,7 @@ def test_update_components_event_one():
@pytest.mark.usefixtures('auth_app_context')
def test_update_components_event_multiple():
computer = Computer(serial_number='sn1', model='ml1', manufacturer='mr1')
computer = Desktop(serial_number='sn1', model='ml1', manufacturer='mr1')
hdd = HardDrive(serial_number='foo', manufacturer='bar', model='foo-bar')
computer.components.add(hdd)
@ -168,7 +169,7 @@ def test_update_components_event_multiple():
@pytest.mark.usefixtures('auth_app_context')
def test_update_parent():
computer = Computer(serial_number='sn1', model='ml1', manufacturer='mr1')
computer = Desktop(serial_number='sn1', model='ml1', manufacturer='mr1')
hdd = HardDrive(serial_number='foo', manufacturer='bar', model='foo-bar')
computer.components.add(hdd)

6
tests/test_price.py Normal file
View File

@ -0,0 +1,6 @@
import pytest
@pytest.mark.xfail(reason='Just needs to do the test')
def test_price_no_data_storage():
pass

16
tests/test_rate.py
View File

@ -3,7 +3,7 @@ from distutils.version import StrictVersion
import pytest
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.device.models import Computer
from ereuse_devicehub.resources.device.models import Computer, Desktop
from ereuse_devicehub.resources.enums import Bios, ComputerChassis, ImageMimeTypes, Orientation, \
RatingSoftware
from ereuse_devicehub.resources.event.models import PhotoboxRate, WorkbenchRate
@ -11,31 +11,31 @@ from ereuse_devicehub.resources.image.models import Image, ImageList
@pytest.mark.usefixtures('auth_app_context')
def test_workbench_rate():
def test_workbench_rate_db():
rate = WorkbenchRate(processor=0.1,
ram=1.0,
bios=Bios.A,
labelling=False,
graphic_card=0.1,
data_storage=4.1,
algorithm_software=RatingSoftware.Ereuse,
algorithm_version=StrictVersion('1.0'),
software=RatingSoftware.ECost,
version=StrictVersion('1.0'),
device=Computer(serial_number='24', chassis=ComputerChassis.Tower))
db.session.add(rate)
db.session.commit()
@pytest.mark.usefixtures('auth_app_context')
def test_photobox_rate():
pc = Computer(serial_number='24', chassis=ComputerChassis.Tower)
def test_photobox_rate_db():
pc = Desktop(serial_number='24', chassis=ComputerChassis.Tower)
image = Image(name='foo',
content=b'123',
file_format=ImageMimeTypes.jpg,
orientation=Orientation.Horizontal,
image_list=ImageList(device=pc))
rate = PhotoboxRate(image=image,
algorithm_software=RatingSoftware.Ereuse,
algorithm_version=StrictVersion('1.0'),
software=RatingSoftware.ECost,
version=StrictVersion('1.0'),
device=pc)
db.session.add(rate)
db.session.commit()

564
tests/test_snapshot.py
View File

@ -8,18 +8,297 @@ 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 import models as m
from ereuse_devicehub.resources.device.exceptions import NeedsId
from ereuse_devicehub.resources.device.models import Computer, Device
from ereuse_devicehub.resources.device.sync import MismatchBetweenTagsAndHid
from ereuse_devicehub.resources.enums import Bios, RatingSoftware, SnapshotSoftware, \
ComputerChassis
from ereuse_devicehub.resources.event.models import Event, Snapshot, SnapshotRequest, \
WorkbenchRate
from ereuse_devicehub.resources.enums import Bios, ComputerChassis, RatingSoftware, \
SnapshotSoftware
from ereuse_devicehub.resources.event.models import AggregateRate, BenchmarkProcessor, \
EraseSectors, Event, Snapshot, SnapshotRequest, WorkbenchRate
from ereuse_devicehub.resources.tag import Tag
from ereuse_devicehub.resources.user.models import User
from tests.conftest import file
@pytest.mark.usefixtures('auth_app_context')
def test_snapshot_model():
"""
Tests creating a Snapshot with its relationships ensuring correct
DB mapping.
"""
device = m.Desktop(serial_number='a1', chassis=ComputerChassis.Tower)
# noinspection PyArgumentList
snapshot = Snapshot(uuid=uuid4(),
date=datetime.now(),
version='1.0',
software=SnapshotSoftware.DesktopApp,
elapsed=timedelta(seconds=25))
snapshot.device = device
snapshot.request = SnapshotRequest(request={'foo': 'bar'})
snapshot.events.add(WorkbenchRate(processor=0.1,
ram=1.0,
bios=Bios.A,
labelling=False,
graphic_card=0.1,
data_storage=4.1,
software=RatingSoftware.ECost,
version=StrictVersion('1.0'),
device=device))
db.session.add(snapshot)
db.session.commit()
device = m.Desktop.query.one() # type: m.Desktop
e1, e2 = device.events
assert isinstance(e1, Snapshot), 'Creation order must be preserved: 1. snapshot, 2. WR'
assert isinstance(e2, WorkbenchRate)
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 m.Desktop.query.one_or_none() is None
assert m.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), data correctness,
and relationship correctness.
"""
snapshot = snapshot_and_check(user, file('basic.snapshot'),
event_types=(
WorkbenchRate.t,
AggregateRate.t,
BenchmarkProcessor.t
),
perform_second_snapshot=False)
assert snapshot['software'] == 'Workbench'
assert snapshot['version'] == '11.0'
assert snapshot['uuid'] == 'f5efd26e-8754-46bc-87bf-fbccc39d60d9'
assert snapshot['elapsed'] == 4
assert snapshot['author']['id'] == user.user['id']
assert 'events' not in snapshot['device']
assert 'author' not in snapshot['device']
device, _ = user.get(res=m.Device, item=snapshot['device']['id'])
assert snapshot['components'] == device['components']
assert tuple(c['type'] for c in snapshot['components']) == (m.GraphicCard.t, m.RamModule.t,
m.Processor.t)
rate = next(e for e in snapshot['events'] if e['type'] == WorkbenchRate.t)
rate, _ = user.get(res=Event, item=rate['id'])
assert rate['device']['id'] == snapshot['device']['id']
assert rate['components'] == snapshot['components']
assert rate['snapshot']['id'] == snapshot['id']
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(
(
e['type'],
[c['serialNumber'] for c in e['components']]
)
for e in user.get_many(res=Event, resources=events, key='id')
)
# We add the first device (2 times). The distribution of components
# (represented with their S/N) should be:
# PC 1: p1c1s, p1c2s, p1c3s. PC 2: ø
s1 = file('1-device-with-components.snapshot')
snapshot1 = snapshot_and_check(user, s1, perform_second_snapshot=False)
pc1_id = snapshot1['device']['id']
pc1, _ = user.get(res=m.Device, item=pc1_id)
# Parent contains components
assert tuple(c['serialNumber'] for c in pc1['components']) == ('p1c1s', 'p1c2s', 'p1c3s')
# Components contain parent
assert all(c['parent'] == pc1_id for c in pc1['components'])
# pc has Snapshot as event
assert len(pc1['events']) == 1
assert pc1['events'][0]['type'] == Snapshot.t
# p1c1s has Snapshot
p1c1s, _ = user.get(res=m.Device, item=pc1['components'][0]['id'])
assert tuple(e['type'] for e in p1c1s['events']) == ('Snapshot',)
# We register a new device
# It has the processor of the first one (p1c2s)
# PC 1: p1c1s, p1c3s. PC 2: p2c1s, p1c2s
# Events PC1: Snapshot, Remove. PC2: Snapshot
s2 = file('2-second-device-with-components-of-first.snapshot')
# num_events = 2 = Remove, Add
snapshot2 = snapshot_and_check(user, s2, event_types=('Remove',),
perform_second_snapshot=False)
pc2_id = snapshot2['device']['id']
pc1, _ = user.get(res=m.Device, item=pc1_id)
pc2, _ = user.get(res=m.Device, item=pc2_id)
# PC1
assert tuple(c['serialNumber'] for c in pc1['components']) == ('p1c1s', 'p1c3s')
assert all(c['parent'] == pc1_id for c in pc1['components'])
assert tuple(e['type'] for e in pc1['events']) == ('Snapshot', 'Remove')
# PC2
assert tuple(c['serialNumber'] for c in pc2['components']) == ('p1c2s', 'p2c1s')
assert all(c['parent'] == pc2_id for c in pc2['components'])
assert tuple(e['type'] for e in pc2['events']) == ('Snapshot',)
# p1c2s has two Snapshots, a Remove and an Add
p1c2s, _ = user.get(res=m.Device, item=pc2['components'][0]['id'])
assert tuple(e['type'] for e in p1c2s['events']) == ('Snapshot', 'Snapshot', 'Remove')
# We register the first device again, but removing motherboard
# and moving processor from the second device to the first.
# We have created 1 Remove (from PC2's processor back to PC1)
# PC 0: p1c2s, p1c3s. PC 1: p2c1s
s3 = file('3-first-device-but-removing-motherboard-and-adding-processor-from-2.snapshot')
snapshot_and_check(user, s3, ('Remove',), perform_second_snapshot=False)
pc1, _ = user.get(res=m.Device, item=pc1_id)
pc2, _ = user.get(res=m.Device, item=pc2_id)
# PC1
assert {c['serialNumber'] for c in pc1['components']} == {'p1c2s', 'p1c3s'}
assert all(c['parent'] == pc1_id for c in pc1['components'])
assert tuple(get_events_info(pc1['events'])) == (
# id, type, components, snapshot
('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',)
assert all(c['parent'] == pc2_id for c in pc2['components'])
assert tuple(e['type'] for e in pc2['events']) == (
'Snapshot', # Second Snapshot
'Remove' # the processor we added in 2.
)
# p1c2s has Snapshot, Remove and Add
p1c2s, _ = user.get(res=m.Device, item=pc1['components'][0]['id'])
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,
# adding a graphic card and adding a new component
s4 = file('4-first-device-but-removing-processor.snapshot-and-adding-graphic-card')
snapshot_and_check(user, s4, perform_second_snapshot=False)
pc1, _ = user.get(res=m.Device, item=pc1_id)
pc2, _ = user.get(res=m.Device, item=pc2_id)
# PC 0: p1c3s, p1c4s. PC1: p2c1s
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] == ('Snapshot', ['p1c3s', 'p1c4s'])
# PC2
# 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=m.Device)
s['device']['id'] = 1 # Assign the ID of the placeholder
user.post(s, res=Snapshot)
def test_snapshot_mismatch_id():
"""Tests uploading a device with an ID from another device."""
# Note that this won't happen as in this new version
# the ID is not used in the Snapshot process
pass
def test_snapshot_tag_inner_tag(tag_id: str, user: UserClient, app: Devicehub):
"""Tests a posting Snapshot with a local tag."""
b = file('basic.snapshot')
b['device']['tags'] = [{'type': 'Tag', 'id': tag_id}]
snapshot_and_check(user, b,
event_types=(WorkbenchRate.t, AggregateRate.t, BenchmarkProcessor.t))
with app.app_context():
tag, *_ = Tag.query.all() # type: Tag
assert tag.device_id == 1, 'Tag should be linked to the first device'
def test_snapshot_tag_inner_tag_mismatch_between_tags_and_hid(user: UserClient, tag_id: str):
"""Ensures one device cannot 'steal' the tag from another one."""
pc1 = file('basic.snapshot')
pc1['device']['tags'] = [{'type': 'Tag', 'id': tag_id}]
user.post(pc1, res=Snapshot)
pc2 = file('1-device-with-components.snapshot')
user.post(pc2, res=Snapshot) # PC2 uploads well
pc2['device']['tags'] = [{'type': 'Tag', 'id': tag_id}] # Set tag from pc1 to pc2
user.post(pc2, res=Snapshot, status=MismatchBetweenTagsAndHid)
def test_erase(user: UserClient):
"""Tests a Snapshot with EraseSectors."""
s = file('erase-sectors.snapshot')
snapshot = snapshot_and_check(user, s, (EraseSectors.t,), perform_second_snapshot=True)
storage, *_ = snapshot['components']
assert storage['type'] == 'SolidStateDrive', 'Components must be ordered by input order'
storage, _ = user.get(res=m.Device, item=storage['id']) # Let's get storage events too
# order: creation time descending
erasure1, _snapshot1, erasure2, _snapshot2 = storage['events']
assert erasure1['type'] == erasure2['type'] == 'EraseSectors'
assert _snapshot1['type'] == _snapshot2['type'] == 'Snapshot'
assert snapshot == user.get(res=Event, item=_snapshot2['id'])[0]
erasure, _ = user.get(res=Event, item=erasure1['id'])
assert len(erasure['steps']) == 2
assert erasure['steps'][0]['startTime'] == '2018-06-01T08:15:00+00:00'
assert erasure['steps'][0]['endTime'] == '2018-06-01T09:16:00+00:00'
assert erasure['steps'][1]['startTime'] == '2018-06-01T08:16:00+00:00'
assert erasure['steps'][1]['endTime'] == '2018-06-01T09:17:00+00:00'
assert erasure['device']['id'] == storage['id']
for step in erasure['steps']:
assert step['type'] == 'StepZero'
assert step['error'] is False
assert 'num' not in step
def test_snapshot_computer_monitor(user: UserClient):
s = file('computer-monitor.snapshot')
snapshot_and_check(user, s, event_types=('AppRate',))
def test_snapshot_components_none():
"""
Tests that a snapshot without components does not
remove them from the computer.
"""
# todo test
pass
def test_snapshot_components_empty():
"""
Tests that a snapshot whose components are an empty list remove
all its components.
"""
def assert_similar_device(device1: dict, device2: dict):
"""
Like :class:`ereuse_devicehub.resources.device.models.Device.
@ -60,7 +339,8 @@ def snapshot_and_check(user: UserClient,
:return: The last resulting snapshot.
"""
snapshot, _ = user.post(res=Snapshot, data=input_snapshot)
assert tuple(e['type'] for e in snapshot['events']) == event_types
assert all(e['type'] in event_types for e in snapshot['events'])
assert len(snapshot['events']) == len(event_types)
# Ensure there is no Remove event after the first Add
found_add = False
for event in snapshot['events']:
@ -80,275 +360,3 @@ def snapshot_and_check(user: UserClient,
return snapshot_and_check(user, input_snapshot, event_types, perform_second_snapshot=False)
else:
return snapshot
@pytest.mark.usefixtures('auth_app_context')
def test_snapshot_model():
"""
Tests creating a Snapshot with its relationships ensuring correct
DB mapping.
"""
device = Computer(serial_number='a1', chassis=ComputerChassis.Tower)
# noinspection PyArgumentList
snapshot = Snapshot(uuid=uuid4(),
date=datetime.now(),
version='1.0',
software=SnapshotSoftware.DesktopApp,
elapsed=timedelta(seconds=25))
snapshot.device = device
snapshot.request = SnapshotRequest(request={'foo': 'bar'})
snapshot.events.add(WorkbenchRate(processor=0.1,
ram=1.0,
bios=Bios.A,
labelling=False,
graphic_card=0.1,
data_storage=4.1,
algorithm_software=RatingSoftware.Ereuse,
algorithm_version=StrictVersion('1.0'),
device=device))
db.session.add(snapshot)
db.session.commit()
device = Computer.query.one() # type: Computer
e1, e2 = device.events
assert isinstance(e1, Snapshot), 'Creation order must be preserved: 1. snapshot, 2. WR'
assert isinstance(e2, WorkbenchRate)
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 Computer.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), data correctness,
and relationship correctness.
"""
snapshot = snapshot_and_check(user, file('basic.snapshot'),
event_types=('WorkbenchRate',),
perform_second_snapshot=False)
assert snapshot['software'] == 'Workbench'
assert snapshot['version'] == '11.0'
assert snapshot['uuid'] == 'f5efd26e-8754-46bc-87bf-fbccc39d60d9'
assert snapshot['elapsed'] == 4
assert snapshot['author']['id'] == user.user['id']
assert 'events' not in snapshot['device']
assert 'author' not in snapshot['device']
device, _ = user.get(res=Device, item=snapshot['device']['id'])
assert snapshot['components'] == device['components']
assert tuple(c['type'] for c in snapshot['components']) == ('GraphicCard', 'RamModule')
rate, _ = user.get(res=Event, item=snapshot['events'][0]['id'])
assert rate['device']['id'] == snapshot['device']['id']
assert rate['components'] == snapshot['components']
assert rate['snapshot']['id'] == snapshot['id']
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(
(
e['type'],
[c['serialNumber'] for c in e['components']]
)
for e in user.get_many(res=Event, resources=events, key='id')
)
# We add the first device (2 times). The distribution of components
# (represented with their S/N) should be:
# PC 1: p1c1s, p1c2s, p1c3s. PC 2: ø
s1 = file('1-device-with-components.snapshot')
snapshot1 = snapshot_and_check(user, s1, perform_second_snapshot=False)
pc1_id = snapshot1['device']['id']
pc1, _ = user.get(res=Device, item=pc1_id)
# Parent contains components
assert tuple(c['serialNumber'] for c in pc1['components']) == ('p1c1s', 'p1c2s', 'p1c3s')
# Components contain parent
assert all(c['parent'] == pc1_id for c in pc1['components'])
# pc has Snapshot as event
assert len(pc1['events']) == 1
assert pc1['events'][0]['type'] == Snapshot.t
# p1c1s has Snapshot
p1c1s, _ = user.get(res=Device, item=pc1['components'][0]['id'])
assert tuple(e['type'] for e in p1c1s['events']) == ('Snapshot',)
# We register a new device
# It has the processor of the first one (p1c2s)
# PC 1: p1c1s, p1c3s. PC 2: p2c1s, p1c2s
# Events PC1: Snapshot, Remove. PC2: Snapshot
s2 = file('2-second-device-with-components-of-first.snapshot')
# num_events = 2 = Remove, Add
snapshot2 = snapshot_and_check(user, s2, event_types=('Remove',),
perform_second_snapshot=False)
pc2_id = snapshot2['device']['id']
pc1, _ = user.get(res=Device, item=pc1_id)
pc2, _ = user.get(res=Device, item=pc2_id)
# PC1
assert tuple(c['serialNumber'] for c in pc1['components']) == ('p1c1s', 'p1c3s')
assert all(c['parent'] == pc1_id for c in pc1['components'])
assert tuple(e['type'] for e in pc1['events']) == ('Snapshot', 'Remove')
# PC2
assert tuple(c['serialNumber'] for c in pc2['components']) == ('p1c2s', 'p2c1s')
assert all(c['parent'] == pc2_id for c in pc2['components'])
assert tuple(e['type'] for e in pc2['events']) == ('Snapshot',)
# p1c2s has two Snapshots, a Remove and an Add
p1c2s, _ = user.get(res=Device, item=pc2['components'][0]['id'])
assert tuple(e['type'] for e in p1c2s['events']) == ('Snapshot', 'Snapshot', 'Remove')
# We register the first device again, but removing motherboard
# and moving processor from the second device to the first.
# We have created 1 Remove (from PC2's processor back to PC1)
# PC 0: p1c2s, p1c3s. PC 1: p2c1s
s3 = file('3-first-device-but-removing-motherboard-and-adding-processor-from-2.snapshot')
snapshot_and_check(user, s3, ('Remove',), perform_second_snapshot=False)
pc1, _ = user.get(res=Device, item=pc1_id)
pc2, _ = user.get(res=Device, item=pc2_id)
# PC1
assert {c['serialNumber'] for c in pc1['components']} == {'p1c2s', 'p1c3s'}
assert all(c['parent'] == pc1_id for c in pc1['components'])
assert tuple(get_events_info(pc1['events'])) == (
# id, type, components, snapshot
('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',)
assert all(c['parent'] == pc2_id for c in pc2['components'])
assert tuple(e['type'] for e in pc2['events']) == (
'Snapshot', # Second Snapshot
'Remove' # the processor we added in 2.
)
# p1c2s has Snapshot, Remove and Add
p1c2s, _ = user.get(res=Device, item=pc1['components'][0]['id'])
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,
# adding a graphic card and adding a new component
s4 = file('4-first-device-but-removing-processor.snapshot-and-adding-graphic-card')
snapshot_and_check(user, s4, perform_second_snapshot=False)
pc1, _ = user.get(res=Device, item=pc1_id)
pc2, _ = user.get(res=Device, item=pc2_id)
# PC 0: p1c3s, p1c4s. PC1: p2c1s
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] == ('Snapshot', ['p1c3s', 'p1c4s'])
# PC2
# 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)
def test_snapshot_mismatch_id():
"""Tests uploading a device with an ID from another device."""
# Note that this won't happen as in this new algorithm_version
# the ID is not used in the Snapshot process
pass
def test_snapshot_tag_inner_tag(tag_id: str, user: UserClient, app: Devicehub):
"""Tests a posting Snapshot with a local tag."""
b = file('basic.snapshot')
b['device']['tags'] = [{'type': 'Tag', 'id': tag_id}]
snapshot_and_check(user, b, event_types=('WorkbenchRate',))
with app.app_context():
tag, *_ = Tag.query.all() # type: Tag
assert tag.device_id == 1, 'Tag should be linked to the first device'
def test_snapshot_tag_inner_tag_mismatch_between_tags_and_hid(user: UserClient, tag_id: str):
"""Ensures one device cannot 'steal' the tag from another one."""
pc1 = file('basic.snapshot')
pc1['device']['tags'] = [{'type': 'Tag', 'id': tag_id}]
user.post(pc1, res=Snapshot)
pc2 = file('1-device-with-components.snapshot')
user.post(pc2, res=Snapshot) # PC2 uploads well
pc2['device']['tags'] = [{'type': 'Tag', 'id': tag_id}] # Set tag from pc1 to pc2
user.post(pc2, res=Snapshot, status=MismatchBetweenTagsAndHid)
def test_erase(user: UserClient):
"""Tests a Snapshot with EraseSectors."""
s = file('erase-sectors.snapshot')
snapshot = snapshot_and_check(user, s, ('EraseSectors',), perform_second_snapshot=True)
storage, *_ = snapshot['components']
assert storage['type'] == 'SolidStateDrive', 'Components must be ordered by input order'
storage, _ = user.get(res=Device, item=storage['id']) # Let's get storage events too
# order: creation time descending
_snapshot1, erasure1, _snapshot2, erasure2 = storage['events']
assert erasure1['type'] == erasure2['type'] == 'EraseSectors'
assert _snapshot1['type'] == _snapshot2['type'] == 'Snapshot'
assert snapshot == user.get(res=Event, item=_snapshot2['id'])[0]
erasure, _ = user.get(res=Event, item=erasure1['id'])
assert len(erasure['steps']) == 2
assert erasure['steps'][0]['startTime'] == '2018-06-01T08:15:00+00:00'
assert erasure['steps'][0]['endTime'] == '2018-06-01T09:16:00+00:00'
assert erasure['steps'][1]['startTime'] == '2018-06-01T08:16:00+00:00'
assert erasure['steps'][1]['endTime'] == '2018-06-01T09:17:00+00:00'
assert erasure['device']['id'] == storage['id']
for step in erasure['steps']:
assert step['type'] == 'StepZero'
assert step['error'] is False
assert 'num' not in step
def test_snapshot_computer_monitor(user: UserClient):
s = file('computer-monitor.snapshot')
snapshot_and_check(user, s, event_types=('AppRate',))
def test_snapshot_components_none():
"""
Tests that a snapshot without components does not
remove them from the computer.
"""
# todo test
pass
def test_snapshot_components_empty():
"""
Tests that a snapshot whose components are an empty list remove
all its components.
"""

4
tests/test_tag.py
View File

@ -5,7 +5,7 @@ from sqlalchemy.exc import IntegrityError
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 Computer
from ereuse_devicehub.resources.device.models import Desktop
from ereuse_devicehub.resources.enums import ComputerChassis
from ereuse_devicehub.resources.tag import Tag
from ereuse_devicehub.resources.tag.view import CannotCreateETag, TagNotLinked
@ -87,7 +87,7 @@ def test_tag_get_device_from_tag_endpoint(app: Devicehub, user: UserClient):
with app.app_context():
# Create a pc with a tag
tag = Tag(id='foo-bar')
pc = Computer(serial_number='sn1', chassis=ComputerChassis.Tower)
pc = Desktop(serial_number='sn1', chassis=ComputerChassis.Tower)
pc.tags.add(tag)
db.session.add(pc)
db.session.commit()