Compare commits

...
This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.

22 commits

Author SHA1 Message Date
Santiago L c94b5a648b Autoformat black & isort 2022-04-06 13:05:51 +02:00
Santiago L 186f6398c0 Fix issue calling create_table_from_selectable() on LotDeviceDescendants
Replace `None` by `Column()`

sqlalchemy_utils on a4154bd0809bc6bbf0c27d5f7c0f3f2872edd779
breaks using `None` as padding parameter needed to follow restriction:
"All selectables passed to CompoundSelect must have identical numbers of columns"
2022-04-06 13:05:31 +02:00
Santiago L ecafc9ea39 Fix requirements part 2 2022-04-06 13:03:15 +02:00
Santiago L d1f27cc8e7 Fix incompatible dependecies with teal requirements 2022-04-06 11:31:49 +02:00
Santiago L 47a167b947 Merge branch 'upgrade-dependencies' into teal3 2022-04-05 14:03:02 +02:00
Santiago L e77bdfdad7 Bump Werkzeug to version >= 2.0 2022-04-05 13:48:12 +02:00
Santiago L 7f83b670ca Bump to click 8.0 2022-04-05 13:48:12 +02:00
Santiago L 6e5f6e2879
Merge branch 'testing' into dependencies/teal 2022-04-05 13:43:10 +02:00
Santiago L 305ddec2b8 Bump Flask to version 2.x 2022-04-05 13:37:27 +02:00
Santiago L e5dada4c17 Bump teal to devel version
Test if upgraded requirements works (Flask 2.x ...)
2022-04-05 13:32:37 +02:00
Santiago L e185ba297b Move testing requirements to separate file
Created requirements-testing.txt
2022-02-11 12:53:19 +01:00
Santiago L 5cf6adb1c3 Drop Werkzeug because it's a Flask dependency 2022-02-11 12:49:10 +01:00
Santiago L d66c335118 Drop python-dotenv from requirements.txt (unused) 2022-02-11 12:28:18 +01:00
Santiago L 0b41b526fd Drop tqdm as requirement (added but never used) 2022-02-11 12:25:48 +01:00
Santiago L cb231a35f8 Merge branch 'testing' into upgrade-dependencies 2022-02-11 12:09:21 +01:00
Santiago L 09649fee29 Sort requirements alphabetically 2022-02-10 09:51:42 +01:00
Santiago L 74432031e4 Revert "Add pytest-xdist to run test in parallel"
This reverts commit 3a395bed80.
2022-02-09 09:54:57 +01:00
Santiago L 3a395bed80 Add pytest-xdist to run test in parallel
Disable coverage because it's not compatible with parallel execution
2022-02-09 09:42:44 +01:00
Santiago L 01b85661b9 Fix "message" parameter of pytest.raises (part 2) 2022-02-09 09:29:15 +01:00
Santiago L 4256a8ba81 Add missing dependency more-itertools 2022-02-08 14:53:44 +01:00
Santiago L 311369691f Fix "message" parameter of pytest.raises
Removed in version 5.0 of pytest
https://docs.pytest.org/en/7.0.x/deprecations.html#message-parameter-of-pytest-raises
2022-02-08 14:48:54 +01:00
Santiago L 8c8323308b Bump pytest to 7.0.0 and drop deprecated pytest-runner 2022-02-08 14:38:13 +01:00
9 changed files with 219 additions and 157 deletions

View file

@ -47,7 +47,7 @@ jobs:
sudo apt-get update -qy sudo apt-get update -qy
sudo apt-get -y install postgresql-client --no-install-recommends sudo apt-get -y install postgresql-client --no-install-recommends
python -m pip install --upgrade pip python -m pip install --upgrade pip
pip install flake8 pytest coverage pip install -r requirements-testing.txt
pip install -r requirements.txt pip install -r requirements.txt
- name: Prepare database - name: Prepare database

View file

@ -10,7 +10,7 @@ from sqlalchemy import TEXT
from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy_utils import LtreeType from sqlalchemy_utils import LtreeType
from sqlalchemy_utils.types.ltree import LQUERY from sqlalchemy_utils.types.ltree import LQUERY
from teal.db import CASCADE_OWN, UUIDLtree, check_range, IntEnum from teal.db import CASCADE_OWN, IntEnum, UUIDLtree, check_range
from teal.resource import url_for_resource from teal.resource import url_for_resource
from ereuse_devicehub.db import create_view, db, exp, f from ereuse_devicehub.db import create_view, db, exp, f
@ -21,70 +21,88 @@ from ereuse_devicehub.resources.user.models import User
class Lot(Thing): class Lot(Thing):
id = db.Column(UUID(as_uuid=True), primary_key=True) # uuid is generated on init by default id = db.Column(
UUID(as_uuid=True), primary_key=True
) # uuid is generated on init by default
name = db.Column(CIText(), nullable=False) name = db.Column(CIText(), nullable=False)
description = db.Column(CIText()) description = db.Column(CIText())
description.comment = """A comment about the lot.""" description.comment = """A comment about the lot."""
closed = db.Column(db.Boolean, default=False, nullable=False) closed = db.Column(db.Boolean, default=False, nullable=False)
closed.comment = """A closed lot cannot be modified anymore.""" closed.comment = """A closed lot cannot be modified anymore."""
devices = db.relationship(Device, devices = db.relationship(
backref=db.backref('lots', lazy=True, collection_class=set), Device,
secondary=lambda: LotDevice.__table__, backref=db.backref('lots', lazy=True, collection_class=set),
lazy=True, secondary=lambda: LotDevice.__table__,
collection_class=set) lazy=True,
collection_class=set,
)
"""The **children** devices that the lot has. """The **children** devices that the lot has.
Note that the lot can have more devices, if they are inside Note that the lot can have more devices, if they are inside
descendant lots. descendant lots.
""" """
parents = db.relationship(lambda: Lot, parents = db.relationship(
viewonly=True, lambda: Lot,
lazy=True, viewonly=True,
collection_class=set, lazy=True,
secondary=lambda: LotParent.__table__, collection_class=set,
primaryjoin=lambda: Lot.id == LotParent.child_id, secondary=lambda: LotParent.__table__,
secondaryjoin=lambda: LotParent.parent_id == Lot.id, primaryjoin=lambda: Lot.id == LotParent.child_id,
cascade='refresh-expire', # propagate changes outside ORM secondaryjoin=lambda: LotParent.parent_id == Lot.id,
backref=db.backref('children', cascade='refresh-expire', # propagate changes outside ORM
viewonly=True, backref=db.backref(
lazy=True, 'children',
cascade='refresh-expire', viewonly=True,
collection_class=set) lazy=True,
) cascade='refresh-expire',
collection_class=set,
),
)
"""The parent lots.""" """The parent lots."""
all_devices = db.relationship(Device, all_devices = db.relationship(
viewonly=True, Device,
lazy=True, viewonly=True,
collection_class=set, lazy=True,
secondary=lambda: LotDeviceDescendants.__table__, collection_class=set,
primaryjoin=lambda: Lot.id == LotDeviceDescendants.ancestor_lot_id, secondary=lambda: LotDeviceDescendants.__table__,
secondaryjoin=lambda: LotDeviceDescendants.device_id == Device.id) primaryjoin=lambda: Lot.id == LotDeviceDescendants.ancestor_lot_id,
secondaryjoin=lambda: LotDeviceDescendants.device_id == Device.id,
)
"""All devices, including components, inside this lot and its """All devices, including components, inside this lot and its
descendants. descendants.
""" """
amount = db.Column(db.Integer, check_range('amount', min=0, max=100), default=0) amount = db.Column(db.Integer, check_range('amount', min=0, max=100), default=0)
owner_id = db.Column(UUID(as_uuid=True), owner_id = db.Column(
db.ForeignKey(User.id), UUID(as_uuid=True),
nullable=False, db.ForeignKey(User.id),
default=lambda: g.user.id) nullable=False,
default=lambda: g.user.id,
)
owner = db.relationship(User, primaryjoin=owner_id == User.id) owner = db.relationship(User, primaryjoin=owner_id == User.id)
transfer_state = db.Column(IntEnum(TransferState), default=TransferState.Initial, nullable=False) transfer_state = db.Column(
IntEnum(TransferState), default=TransferState.Initial, nullable=False
)
transfer_state.comment = TransferState.__doc__ transfer_state.comment = TransferState.__doc__
receiver_address = db.Column(CIText(), receiver_address = db.Column(
db.ForeignKey(User.email), CIText(),
nullable=False, db.ForeignKey(User.email),
default=lambda: g.user.email) nullable=False,
default=lambda: g.user.email,
)
receiver = db.relationship(User, primaryjoin=receiver_address == User.email) receiver = db.relationship(User, primaryjoin=receiver_address == User.email)
def __init__(self, name: str, closed: bool = closed.default.arg, def __init__(
description: str = None) -> None: self, name: str, closed: bool = closed.default.arg, description: str = None
) -> None:
"""Initializes a lot """Initializes a lot
:param name: :param name:
:param closed: :param closed:
""" """
super().__init__(id=uuid.uuid4(), name=name, closed=closed, description=description) super().__init__(
id=uuid.uuid4(), name=name, closed=closed, description=description
)
Path(self) # Lots have always one edge per default. Path(self) # Lots have always one edge per default.
@property @property
@ -115,7 +133,9 @@ class Lot(Thing):
@classmethod @classmethod
def descendantsq(cls, id): def descendantsq(cls, id):
_id = UUIDLtree.convert(id) _id = UUIDLtree.convert(id)
return (cls.id == Path.lot_id) & Path.path.lquery(exp.cast('*.{}.*'.format(_id), LQUERY)) return (cls.id == Path.lot_id) & Path.path.lquery(
exp.cast('*.{}.*'.format(_id), LQUERY)
)
@classmethod @classmethod
def roots(cls): def roots(cls):
@ -176,13 +196,17 @@ class Lot(Thing):
if isinstance(child, Lot): if isinstance(child, Lot):
return Path.has_lot(self.id, child.id) return Path.has_lot(self.id, child.id)
elif isinstance(child, Device): elif isinstance(child, Device):
device = db.session.query(LotDeviceDescendants) \ device = (
.filter(LotDeviceDescendants.device_id == child.id) \ db.session.query(LotDeviceDescendants)
.filter(LotDeviceDescendants.ancestor_lot_id == self.id) \ .filter(LotDeviceDescendants.device_id == child.id)
.filter(LotDeviceDescendants.ancestor_lot_id == self.id)
.one_or_none() .one_or_none()
)
return device return device
else: else:
raise TypeError('Lot only contains devices and lots, not {}'.format(child.__class__)) raise TypeError(
'Lot only contains devices and lots, not {}'.format(child.__class__)
)
def __repr__(self) -> str: def __repr__(self) -> str:
return '<Lot {0.name} devices={0.devices!r}>'.format(self) return '<Lot {0.name} devices={0.devices!r}>'.format(self)
@ -192,35 +216,44 @@ class LotDevice(db.Model):
device_id = db.Column(db.BigInteger, db.ForeignKey(Device.id), primary_key=True) device_id = db.Column(db.BigInteger, db.ForeignKey(Device.id), primary_key=True)
lot_id = db.Column(UUID(as_uuid=True), db.ForeignKey(Lot.id), primary_key=True) lot_id = db.Column(UUID(as_uuid=True), db.ForeignKey(Lot.id), primary_key=True)
created = db.Column(db.DateTime, nullable=False, default=datetime.utcnow) created = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
author_id = db.Column(UUID(as_uuid=True), author_id = db.Column(
db.ForeignKey(User.id), UUID(as_uuid=True),
nullable=False, db.ForeignKey(User.id),
default=lambda: g.user.id) nullable=False,
default=lambda: g.user.id,
)
author = db.relationship(User, primaryjoin=author_id == User.id) author = db.relationship(User, primaryjoin=author_id == User.id)
author_id.comment = """The user that put the device in the lot.""" author_id.comment = """The user that put the device in the lot."""
class Path(db.Model): class Path(db.Model):
id = db.Column(db.UUID(as_uuid=True), id = db.Column(
primary_key=True, db.UUID(as_uuid=True),
server_default=db.text('gen_random_uuid()')) primary_key=True,
server_default=db.text('gen_random_uuid()'),
)
lot_id = db.Column(db.UUID(as_uuid=True), db.ForeignKey(Lot.id), nullable=False) lot_id = db.Column(db.UUID(as_uuid=True), db.ForeignKey(Lot.id), nullable=False)
lot = db.relationship(Lot, lot = db.relationship(
backref=db.backref('paths', Lot,
lazy=True, backref=db.backref(
collection_class=set, 'paths', lazy=True, collection_class=set, cascade=CASCADE_OWN
cascade=CASCADE_OWN), ),
primaryjoin=Lot.id == lot_id) primaryjoin=Lot.id == lot_id,
)
path = db.Column(LtreeType, nullable=False) path = db.Column(LtreeType, nullable=False)
created = db.Column(db.TIMESTAMP(timezone=True), server_default=db.text('CURRENT_TIMESTAMP')) created = db.Column(
db.TIMESTAMP(timezone=True), server_default=db.text('CURRENT_TIMESTAMP')
)
created.comment = """When Devicehub created this.""" created.comment = """When Devicehub created this."""
__table_args__ = ( __table_args__ = (
# dag.delete_edge needs to disable internally/temporarily the unique constraint # dag.delete_edge needs to disable internally/temporarily the unique constraint
db.UniqueConstraint(path, name='path_unique', deferrable=True, initially='immediate'), db.UniqueConstraint(
path, name='path_unique', deferrable=True, initially='immediate'
),
db.Index('path_gist', path, postgresql_using='gist'), db.Index('path_gist', path, postgresql_using='gist'),
db.Index('path_btree', path, postgresql_using='btree'), db.Index('path_btree', path, postgresql_using='btree'),
db.Index('lot_id_index', lot_id, postgresql_using='hash') db.Index('lot_id_index', lot_id, postgresql_using='hash'),
) )
def __init__(self, lot: Lot) -> None: def __init__(self, lot: Lot) -> None:
@ -243,7 +276,9 @@ class Path(db.Model):
child_id = UUIDLtree.convert(child_id) child_id = UUIDLtree.convert(child_id)
return bool( return bool(
db.session.execute( db.session.execute(
"SELECT 1 from path where path ~ '*.{}.*.{}.*'".format(parent_id, child_id) "SELECT 1 from path where path ~ '*.{}.*.{}.*'".format(
parent_id, child_id
)
).first() ).first()
) )
@ -263,47 +298,75 @@ class LotDeviceDescendants(db.Model):
"""Ancestor lot table.""" """Ancestor lot table."""
_desc = Lot.__table__.alias() _desc = Lot.__table__.alias()
"""Descendant lot table.""" """Descendant lot table."""
lot_device = _desc \ lot_device = _desc.join(LotDevice, _desc.c.id == LotDevice.lot_id).join(
.join(LotDevice, _desc.c.id == LotDevice.lot_id) \ Path, _desc.c.id == Path.lot_id
.join(Path, _desc.c.id == Path.lot_id) )
"""Join: Path -- Lot -- LotDevice""" """Join: Path -- Lot -- LotDevice"""
descendants = "path.path ~ (CAST('*.'|| replace(CAST({}.id as text), '-', '_') " \ descendants = (
"|| '.*' AS LQUERY))".format(_ancestor.name) "path.path ~ (CAST('*.'|| replace(CAST({}.id as text), '-', '_') "
"|| '.*' AS LQUERY))".format(_ancestor.name)
)
"""Query that gets the descendants of the ancestor lot.""" """Query that gets the descendants of the ancestor lot."""
devices = db.select([ devices = (
LotDevice.device_id, db.select(
_desc.c.id.label('parent_lot_id'), [
_ancestor.c.id.label('ancestor_lot_id'), LotDevice.device_id,
None _desc.c.id.label('parent_lot_id'),
]).select_from(_ancestor).select_from(lot_device).where(db.text(descendants)) _ancestor.c.id.label('ancestor_lot_id'),
db.column(
'padding'
), # foo column to have same nunber of columns on joined selects (union)
]
)
.select_from(_ancestor)
.select_from(lot_device)
.where(db.text(descendants))
)
# Components # Components
_parent_device = Device.__table__.alias(name='parent_device') _parent_device = Device.__table__.alias(name='parent_device')
"""The device that has the access to the lot.""" """The device that has the access to the lot."""
lot_device_component = lot_device \ lot_device_component = lot_device.join(
.join(_parent_device, _parent_device.c.id == LotDevice.device_id) \ _parent_device, _parent_device.c.id == LotDevice.device_id
.join(Component, _parent_device.c.id == Component.parent_id) ).join(Component, _parent_device.c.id == Component.parent_id)
"""Join: Path -- Lot -- LotDevice -- ParentDevice (Device) -- Component""" """Join: Path -- Lot -- LotDevice -- ParentDevice (Device) -- Component"""
components = db.select([ components = (
Component.id.label('device_id'), db.select(
_desc.c.id.label('parent_lot_id'), [
_ancestor.c.id.label('ancestor_lot_id'), Component.id.label('device_id'),
LotDevice.device_id.label('device_parent_id'), _desc.c.id.label('parent_lot_id'),
]).select_from(_ancestor).select_from(lot_device_component).where(db.text(descendants)) _ancestor.c.id.label('ancestor_lot_id'),
LotDevice.device_id.label('device_parent_id'),
]
)
.select_from(_ancestor)
.select_from(lot_device_component)
.where(db.text(descendants))
)
__table__ = create_view('lot_device_descendants', devices.union(components)) __table__ = create_view('lot_device_descendants', devices.union(components))
class LotParent(db.Model): class LotParent(db.Model):
i = f.index(Path.path, db.func.text2ltree(f.replace(exp.cast(Path.lot_id, TEXT), '-', '_'))) i = f.index(
Path.path, db.func.text2ltree(f.replace(exp.cast(Path.lot_id, TEXT), '-', '_'))
)
__table__ = create_view( __table__ = create_view(
'lot_parent', 'lot_parent',
db.select([ db.select(
Path.lot_id.label('child_id'), [
exp.cast(f.replace(exp.cast(f.subltree(Path.path, i - 1, i), TEXT), '_', '-'), Path.lot_id.label('child_id'),
UUID).label('parent_id') exp.cast(
]).select_from(Path).where(i > 0), f.replace(
exp.cast(f.subltree(Path.path, i - 1, i), TEXT), '_', '-'
),
UUID,
).label('parent_id'),
]
)
.select_from(Path)
.where(i > 0),
) )

3
pytest.ini Normal file
View file

@ -0,0 +1,3 @@
[pytest]
markers =
mvp: mark tests as required by MVP

4
requirements-testing.txt Normal file
View file

@ -0,0 +1,4 @@
coverage
flake8
pytest==7.0.0
requests-mock==1.5.2

View file

@ -1,45 +1,43 @@
alembic==1.4.2 alembic==1.4.2
anytree==2.4.3 anytree==2.4.3
apispec==0.39.0 apispec==5.1.1
boltons==18.0.1 boltons==18.0.1
click==6.7 click==8.0
click-spinner==0.1.8 click-spinner==0.1.8
colorama==0.3.9 colorama==0.3.9
colour==0.1.5 colour==0.1.5
ereuse-utils[naming,test,session,cli]==0.4.0b50 ereuse-utils[naming,test,session,cli]==0.4.0b50
Flask==1.0.2 Flask>=2.0
Flask-Cors==3.0.10 Flask-Cors==3.0.10
Flask-Login==0.5.0 Flask-Login==0.6.0
Flask-SQLAlchemy==2.3.2 Flask-SQLAlchemy==2.5.1
Flask-WTF==1.0.0 Flask-WTF==1.0.0
flask-WeasyPrint==0.5
hashids==1.2.0 hashids==1.2.0
inflection==0.3.1 inflection==0.3.1
# lock itsdangerous version until upgrade to Flask 2.x
itsdangerous==2.0.1 itsdangerous==2.0.1
# lock Jinja2 version because it's the latest compatible with Flask 1.0.X # lock Jinja2 version because it's the latest compatible with Flask 1.0.X
# see related info on https://github.com/pallets/jinja/issues/1628 # see related info on https://github.com/pallets/jinja/issues/1628
Jinja2==3.0.3 Jinja2==3.0.3
marshmallow==3.0.0b11 marshmallow==3.0.0b11
marshmallow-enum==1.4.1 marshmallow-enum==1.4.1
more-itertools==8.12.0
passlib==1.7.1 passlib==1.7.1
phonenumbers==8.9.11 phonenumbers==8.9.11
pytest==3.7.2 psycopg2-binary==2.8.3
pytest-runner==4.2 PyJWT==2.0.0a1
python-dateutil==2.7.3 python-dateutil==2.7.3
python-decouple==3.3
python-stdnum==1.9 python-stdnum==1.9
PyYAML==5.4 PyYAML==5.4
requests[security]==2.27.1 requests[security]==2.27.1
requests-mock==1.5.2
SQLAlchemy==1.3.24
SQLAlchemy-Utils==0.33.11
teal==0.2.0a38
webargs==5.5.3
Werkzeug==0.15.3
sqlalchemy-citext==1.3.post0
flask-weasyprint==0.5
weasyprint==44
psycopg2-binary==2.8.3
sortedcontainers==2.1.0 sortedcontainers==2.1.0
tqdm==4.32.2 SQLAlchemy==1.4.34
python-decouple==3.3 sqlalchemy-citext==1.3.post0
python-dotenv==0.14.0 SQLAlchemy-Utils==0.38.2
pyjwt==2.0.0a1 # teal under development version
-e git+https://github.com/eReuse/teal/@upgrade-dependencies#egg=teal
WeasyPrint==44
webargs==5.5.3
Werkzeug>=2.0

View file

@ -3,11 +3,6 @@ from setuptools import find_packages, setup
from ereuse_devicehub import __version__ from ereuse_devicehub import __version__
test_requires = [
'pytest',
'requests_mock'
]
setup( setup(
name='ereuse-devicehub', name='ereuse-devicehub',
version=__version__, version=__version__,
@ -52,17 +47,12 @@ setup(
'docs-auto': [ 'docs-auto': [
'sphinx-autobuild' 'sphinx-autobuild'
], ],
'test': test_requires
}, },
tests_require=test_requires,
entry_points={ entry_points={
'console_scripts': [ 'console_scripts': [
'dh = ereuse_devicehub.cli:cli' 'dh = ereuse_devicehub.cli:cli'
] ]
}, },
setup_requires=[
'pytest-runner'
],
classifiers=[ classifiers=[
'Development Status :: 2 - Pre-Alpha', 'Development Status :: 2 - Pre-Alpha',
'Environment :: Web Environment', 'Environment :: Web Environment',
@ -71,8 +61,7 @@ setup(
'License :: OSI Approved :: GNU Affero General Public License v3', 'License :: OSI Approved :: GNU Affero General Public License v3',
'Operating System :: OS Independent', 'Operating System :: OS Independent',
'Programming Language :: Python :: 3 :: Only', 'Programming Language :: Python :: 3 :: Only',
'Programming Language :: Python :: 3.5', 'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.6',
'Topic :: Internet :: WWW/HTTP :: Dynamic Content', 'Topic :: Internet :: WWW/HTTP :: Dynamic Content',
'Topic :: Internet :: WWW/HTTP :: WSGI :: Application', 'Topic :: Internet :: WWW/HTTP :: WSGI :: Application',
'Topic :: Software Development :: Libraries :: Python Modules', 'Topic :: Software Development :: Libraries :: Python Modules',

View file

@ -75,14 +75,14 @@ def test_erase_basic():
def test_validate_device_data_storage(): def test_validate_device_data_storage():
"""Checks the validation for data-storage-only actions works.""" """Checks the validation for data-storage-only actions works."""
# We can't set a GraphicCard # We can't set a GraphicCard
with pytest.raises(TypeError, with pytest.raises(TypeError):
message='EraseBasic.device must be a DataStorage '
'but you passed <GraphicCard None model=\'foo-bar\' S/N=\'foo\'>'):
models.EraseBasic( models.EraseBasic(
device=GraphicCard(serial_number='foo', manufacturer='bar', model='foo-bar'), device=GraphicCard(serial_number='foo', manufacturer='bar', model='foo-bar'),
clean_with_zeros=True, clean_with_zeros=True,
**conftest.T **conftest.T
) )
pytest.fail('EraseBasic.device must be a DataStorage '
'but you passed <GraphicCard None model=\'foo-bar\' S/N=\'foo\'>')
@pytest.mark.mvp @pytest.mark.mvp
@ -335,10 +335,10 @@ def test_outgoinlot_status_actions(action_model: models.Action, user: UserClient
@pytest.mark.parametrize('action_model', @pytest.mark.parametrize('action_model',
(pytest.param(ams, id=ams.__class__.__name__) (pytest.param(ams, id=ams.__class__.__name__)
for ams in [ for ams in [
models.Recycling, models.Recycling,
models.Use, models.Use,
models.Refurbish, models.Refurbish,
models.Management models.Management
])) ]))
def test_incominglot_status_actions(action_model: models.Action, user: UserClient, user2: UserClient): def test_incominglot_status_actions(action_model: models.Action, user: UserClient, user2: UserClient):
"""Test of status actions in outgoinlot.""" """Test of status actions in outgoinlot."""
@ -494,10 +494,10 @@ def test_recycling_container(user: UserClient):
@pytest.mark.parametrize('action_model', @pytest.mark.parametrize('action_model',
(pytest.param(ams, id=ams.__class__.__name__) (pytest.param(ams, id=ams.__class__.__name__)
for ams in [ for ams in [
models.Recycling, models.Recycling,
models.Use, models.Use,
models.Refurbish, models.Refurbish,
models.Management models.Management
])) ]))
def test_status_without_lot(action_model: models.Action, user: UserClient): def test_status_without_lot(action_model: models.Action, user: UserClient):
"""Test of status actions for devices without lot.""" """Test of status actions for devices without lot."""
@ -512,10 +512,10 @@ def test_status_without_lot(action_model: models.Action, user: UserClient):
@pytest.mark.parametrize('action_model', @pytest.mark.parametrize('action_model',
(pytest.param(ams, id=ams.__class__.__name__) (pytest.param(ams, id=ams.__class__.__name__)
for ams in [ for ams in [
models.Recycling, models.Recycling,
models.Use, models.Use,
models.Refurbish, models.Refurbish,
models.Management models.Management
])) ]))
def test_status_in_temporary_lot(action_model: models.Action, user: UserClient): def test_status_in_temporary_lot(action_model: models.Action, user: UserClient):
"""Test of status actions for devices in a temporary lot.""" """Test of status actions for devices in a temporary lot."""
@ -913,7 +913,7 @@ def test_allocate(user: UserClient):
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot) snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
device_id = snapshot['device']['id'] device_id = snapshot['device']['id']
devicehub_id = snapshot['device']['devicehubID'] devicehub_id = snapshot['device']['devicehubID']
post_request = {"transaction": "ccc", post_request = {"transaction": "ccc",
"finalUserCode": "aabbcc", "finalUserCode": "aabbcc",
"name": "John", "name": "John",
"severity": "Info", "severity": "Info",
@ -1638,7 +1638,7 @@ def test_confirmRevoke(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices) query=devices)
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',
@ -1659,8 +1659,8 @@ def test_confirmRevoke(user: UserClient, user2: UserClient):
'type': 'Confirm', 'type': 'Confirm',
'action': trade.id, 'action': trade.id,
'devices': [ 'devices': [
snap1['device']['id'], snap1['device']['id'],
snap2['device']['id'], snap2['device']['id'],
snap3['device']['id'], snap3['device']['id'],
snap4['device']['id'], snap4['device']['id'],
snap5['device']['id'], snap5['device']['id'],
@ -1677,7 +1677,7 @@ def test_confirmRevoke(user: UserClient, user2: UserClient):
assert trade.devices[-1].actions[-1].t == 'Confirm' assert trade.devices[-1].actions[-1].t == 'Confirm'
assert trade.devices[-1].actions[-1].user == trade.user_from assert trade.devices[-1].actions[-1].user == trade.user_from
# The manager remove one device of the lot and automaticaly # The manager remove one device of the lot and automaticaly
# is create one revoke action # is create one revoke action
device_10 = trade.devices[-1] device_10 = trade.devices[-1]
lot, _ = user.delete({}, lot, _ = user.delete({},
@ -1804,7 +1804,7 @@ def test_trade_case2(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices[:-1]) query=devices[:-1])
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',
@ -1926,7 +1926,7 @@ def test_trade_case4(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices[:-1]) query=devices[:-1])
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',
@ -1991,7 +1991,7 @@ def test_trade_case5(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices) query=devices)
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',
@ -2057,7 +2057,7 @@ def test_trade_case6(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices[:-1]) query=devices[:-1])
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',
@ -2125,7 +2125,7 @@ def test_trade_case7(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices) query=devices)
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',
@ -2192,7 +2192,7 @@ def test_trade_case8(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices) query=devices)
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',
@ -2266,7 +2266,7 @@ def test_trade_case9(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices[:-1]) query=devices[:-1])
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',
@ -2348,7 +2348,7 @@ def test_trade_case10(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices[:-1]) query=devices[:-1])
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',
@ -2435,7 +2435,7 @@ def test_trade_case11(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices) query=devices)
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',
@ -2507,7 +2507,7 @@ def test_trade_case12(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices) query=devices)
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',
@ -2584,7 +2584,7 @@ def test_trade_case13(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices[:-1]) query=devices[:-1])
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',
@ -2664,7 +2664,7 @@ def test_trade_case14(user: UserClient, user2: UserClient):
item='{}/devices'.format(lot['id']), item='{}/devices'.format(lot['id']),
query=devices[:-1]) query=devices[:-1])
# the manager shares the temporary lot with the SCRAP as an incoming lot # the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it # for the CRAP to confirm it
request_post = { request_post = {
'type': 'Trade', 'type': 'Trade',

View file

@ -24,11 +24,14 @@ def test_authenticate_error(app: Devicehub):
MESSAGE = 'Provide a suitable token.' MESSAGE = 'Provide a suitable token.'
create_user() create_user()
# Token doesn't exist # Token doesn't exist
with pytest.raises(Unauthorized, message=MESSAGE): with pytest.raises(Unauthorized):
app.auth.authenticate(token=str(uuid4())) app.auth.authenticate(token=str(uuid4()))
pytest.fail(MESSAGE)
# Wrong token format # Wrong token format
with pytest.raises(Unauthorized, message=MESSAGE): with pytest.raises(Unauthorized):
app.auth.authenticate(token='this is a wrong uuid') app.auth.authenticate(token='this is a wrong uuid')
pytest.fail(MESSAGE)
@pytest.mark.mvp @pytest.mark.mvp

View file

@ -144,5 +144,7 @@ def test_create_existing_inventory(cli, tdb1):
cli.invoke('inv', 'add', '--common') cli.invoke('inv', 'add', '--common')
with tdb1.app_context(): with tdb1.app_context():
assert db.has_schema('tdb1') assert db.has_schema('tdb1')
with pytest.raises(AssertionError, message='Schema tdb1 already exists.'):
with pytest.raises(AssertionError):
cli.invoke('inv', 'add', '--common') cli.invoke('inv', 'add', '--common')
pytest.fail('Schema tdb1 already exists.')