Merge pull request #365 from eReuse/dependencies/pip-tools

Manage dependencies using pip-tools
This commit is contained in:
Santiago L 2022-09-29 13:18:27 +02:00 committed by GitHub
commit 29f716ab76
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 604 additions and 224 deletions

View File

@ -1,10 +1,8 @@
name: Selenium name: Selenium
on: on:
push:
branches: [master, testing]
pull_request: pull_request:
branches: [master, testing] types: [ready_for_review, review_requested]
jobs: jobs:
build: build:

View File

@ -25,3 +25,7 @@ repos:
entry: npm run babel entry: npm run babel
language: node language: node
files: ^ereuse_devicehub/static/js/main_inventory.js files: ^ereuse_devicehub/static/js/main_inventory.js
- repo: https://github.com/jazzband/pip-tools
rev: 6.8.0
hooks:
- id: pip-compile

View File

@ -30,3 +30,29 @@ pre-commit install
Do this: `device_detail.html` Do this: `device_detail.html`
Don't do this: `DeviceDetail.html`, `Device-detail.html` Don't do this: `DeviceDetail.html`, `Device-detail.html`
## Adding a new dependency to the project
This project tracks its packages using pip-tools, it could be installed by running:
```
pip install pip-tools
```
Whenever you need to install a new package using pip install <package-name>:
1. Put the package name into `requirements.in` instead.
```
# requirements.in
...
new_package
```
2. Compile the requirements
```
pip-compile requirements.in --output-file=requirements.txt
```
3. Then install upgraded dependencies:
```
pip install -U -r requirements.txt
```

41
requirements.in Normal file
View File

@ -0,0 +1,41 @@
alembic==1.4.2
atomicwrites==1.4.0
click-spinner==0.1.8
colorama==0.3.9
colour==0.1.5
ereuse-utils[naming,test,session,cli]==0.4.0b50
Flask-Cors==3.0.10
Flask-Login==0.5.0
Flask-WTF==1.0.0
flask-weasyprint==0.4
hashids==1.2.0
more-itertools==8.12.0
passlib==1.7.1
phonenumbers==8.9.11
psycopg2-binary==2.8.3
pyjwt==2.4.0
python-decouple==3.3
python-dotenv==0.14.0
python-stdnum==1.9
pyyaml==5.4
requests==2.27.1
requests-mock==1.5.2
requests-toolbelt==0.9.1
sortedcontainers==2.1.0
sqlalchemy-citext==1.3.post0
sqlalchemy-utils==0.33.11
teal==0.2.0a38
tqdm==4.32.2
# workbench json parsing dependencies
pint==0.9
py-dmidecode==0.1.0
pandas==1.3.5
numpy==1.22.0 # pandas dependency
odfpy==1.4.1 # pandas dependency
xlrd==2.0.1 # pandas dependency
openpyxl==3.0.10 # pandas dependency
et_xmlfile==1.1.0 # pandas dependency
# manual dependency
marshmallow-enum==1.4.1

View File

@ -1,53 +1,226 @@
#
# This file is autogenerated by pip-compile with python 3.8
# To update, run:
#
# pip-compile --output-file=requirements.txt requirements.in
#
alembic==1.4.2 alembic==1.4.2
# via -r requirements.in
anytree==2.4.3 anytree==2.4.3
# via teal
apispec==0.39.0 apispec==0.39.0
# via teal
atomicwrites==1.4.0
# via -r requirements.in
boltons==18.0.1 boltons==18.0.1
# via
# ereuse-utils
# teal
cairocffi==1.4.0
# via
# cairosvg
# weasyprint
cairosvg==2.5.2
# via weasyprint
certifi==2022.9.24
# via requests
cffi==1.15.1
# via
# cairocffi
# weasyprint
charset-normalizer==2.0.12
# via requests
click==6.7 click==6.7
# via
# ereuse-utils
# flask
click-spinner==0.1.8 click-spinner==0.1.8
# via
# -r requirements.in
# teal
colorama==0.3.9 colorama==0.3.9
# via
# -r requirements.in
# ereuse-utils
colour==0.1.5 colour==0.1.5
ereuse-utils[naming,test,session,cli]==0.4.0b50 # via
Flask==1.0.2 # -r requirements.in
Flask-Cors==3.0.10 # sqlalchemy-utils
Flask-Login==0.5.0 cssselect2==0.7.0
Flask-SQLAlchemy==2.3.2 # via
Flask-WTF==1.0.0 # cairosvg
# weasyprint
defusedxml==0.7.1
# via
# cairosvg
# odfpy
ereuse-utils[cli,naming,session,test]==0.4.0b50
# via
# -r requirements.in
# teal
et-xmlfile==1.1.0
# via
# -r requirements.in
# openpyxl
flask==1.0.2
# via
# ereuse-utils
# flask-cors
# flask-login
# flask-sqlalchemy
# flask-weasyprint
# flask-wtf
# teal
flask-cors==3.0.10
# via
# -r requirements.in
# teal
flask-login==0.5.0
# via -r requirements.in
flask-sqlalchemy==2.5.1
# via teal
flask-weasyprint==0.4
# via -r requirements.in
flask-wtf==1.0.0
# via -r requirements.in
hashids==1.2.0 hashids==1.2.0
# via -r requirements.in
html5lib==1.1
# via weasyprint
idna==3.4
# via requests
inflection==0.3.1 inflection==0.3.1
# via ereuse-utils
itsdangerous==2.0.1 itsdangerous==2.0.1
# lock Jinja2 version because it's the latest compatible with Flask 1.0.X # via
# see related info on https://github.com/pallets/jinja/issues/1628 # flask
Jinja2==3.0.3 # flask-wtf
jinja2==3.0.3
# via flask
mako==1.2.3
# via alembic
markupsafe==2.1.1
# via
# jinja2
# mako
# wtforms
marshmallow==3.0.0b11 marshmallow==3.0.0b11
# via
# marshmallow-enum
# teal
# webargs
marshmallow-enum==1.4.1 marshmallow-enum==1.4.1
passlib==1.7.1 # via -r requirements.in
phonenumbers==8.9.11 more-itertools==8.12.0
pytest==3.7.2 # via -r requirements.in
pytest-runner==4.2 numpy==1.22.0
python-dateutil==2.7.3 # via
python-stdnum==1.9 # -r requirements.in
PyYAML==5.4 # pandas
requests[security]==2.27.1 odfpy==1.4.1
requests-mock==1.5.2 # via -r requirements.in
SQLAlchemy==1.3.24 openpyxl==3.0.10
SQLAlchemy-Utils==0.33.11 # via -r requirements.in
teal==0.2.0a38
webargs==5.5.3
Werkzeug==0.15.5
sqlalchemy-citext==1.3.post0
flask-weasyprint==0.5
weasyprint==44
psycopg2-binary==2.8.3
sortedcontainers==2.1.0
tqdm==4.32.2
python-decouple==3.3
python-dotenv==0.14.0
pyjwt==2.4.0
pint==0.9
py-dmidecode==0.1.0
pandas==1.3.5 pandas==1.3.5
numpy==1.22.0 # pandas dependency # via -r requirements.in
odfpy==1.4.1 # pandas dependency passlib==1.7.1
xlrd==2.0.1 # pandas dependency # via
openpyxl==3.0.10 # pandas dependency # -r requirements.in
et_xmlfile==1.1.0 # pandas dependency # sqlalchemy-utils
phonenumbers==8.9.11
# via
# -r requirements.in
# sqlalchemy-utils
pillow==9.2.0
# via cairosvg
pint==0.9
# via -r requirements.in
psycopg2-binary==2.8.3
# via -r requirements.in
py-dmidecode==0.1.0
# via -r requirements.in
pycparser==2.21
# via cffi
pyjwt==2.4.0
# via -r requirements.in
pyphen==0.13.0
# via weasyprint
python-dateutil==2.7.3
# via
# alembic
# pandas
python-decouple==3.3
# via -r requirements.in
python-dotenv==0.14.0
# via -r requirements.in
python-editor==1.0.4
# via alembic
python-stdnum==1.9
# via -r requirements.in
pytz==2022.2.1
# via pandas
pyyaml==5.4
# via
# -r requirements.in
# apispec
requests==2.27.1
# via
# -r requirements.in
# requests-mock
# requests-toolbelt
requests-mock==1.5.2
# via -r requirements.in
requests-toolbelt==0.9.1
# via
# -r requirements.in
# ereuse-utils
six==1.16.0
# via
# anytree
# flask-cors
# html5lib
# python-dateutil
# requests-mock
# sqlalchemy-utils
sortedcontainers==2.1.0
# via -r requirements.in
sqlalchemy==1.3.24
# via
# alembic
# flask-sqlalchemy
# sqlalchemy-citext
# sqlalchemy-utils
sqlalchemy-citext==1.3.post0
# via -r requirements.in
sqlalchemy-utils[color,password,phone]==0.33.11
# via
# -r requirements.in
# teal
teal==0.2.0a38
# via -r requirements.in
tinycss2==1.1.1
# via
# cairosvg
# cssselect2
# weasyprint
tqdm==4.32.2
# via
# -r requirements.in
# ereuse-utils
urllib3==1.26.12
# via requests
weasyprint==44
# via flask-weasyprint
webargs==5.5.3
# via teal
webencodings==0.5.1
# via
# cssselect2
# html5lib
# tinycss2
werkzeug==2.0.3
# via flask
wtforms==3.0.1
# via flask-wtf
xlrd==2.0.1
# via -r requirements.in

View File

@ -1,5 +1,4 @@
import copy import copy
import ipaddress
import json import json
import os import os
import shutil import shutil
@ -7,7 +6,7 @@ from datetime import datetime, timedelta
from decimal import Decimal from decimal import Decimal
from io import BytesIO from io import BytesIO
from json.decoder import JSONDecodeError from json.decoder import JSONDecodeError
from typing import Tuple, Type from typing import Tuple
import pytest import pytest
from dateutil.tz import tzutc from dateutil.tz import tzutc
@ -15,7 +14,7 @@ from flask import current_app as app
from flask import g from flask import g
from pytest import raises from pytest import raises
from sqlalchemy.util import OrderedSet from sqlalchemy.util import OrderedSet
from teal.enums import Currency, Subdivision from teal.enums import Currency
from ereuse_devicehub.client import Client, UserClient from ereuse_devicehub.client import Client, UserClient
from ereuse_devicehub.db import db from ereuse_devicehub.db import db
@ -82,11 +81,7 @@ 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( with pytest.raises(TypeError):
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( device=GraphicCard(
serial_number='foo', manufacturer='bar', model='foo-bar' serial_number='foo', manufacturer='bar', model='foo-bar'
@ -94,6 +89,10 @@ def test_validate_device_data_storage():
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
@ -292,9 +291,7 @@ def test_generic_action(
for ams in [models.Recycling, models.Use, models.Refurbish, models.Management] for ams in [models.Recycling, models.Use, models.Refurbish, models.Management]
), ),
) )
def test_simple_status_actions( def test_simple_status_actions(action_model: models.Action, user2: UserClient):
action_model: models.Action, user2: UserClient
):
"""Simple test of status action.""" """Simple test of status action."""
user = user2 user = user2
snap, _ = user.post(file('basic.snapshot'), res=models.Snapshot) snap, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
@ -554,7 +551,9 @@ def test_status_without_lot(action_model: models.Action, user: UserClient):
for ams in [models.Recycling, models.Use, models.Refurbish, models.Management] for ams in [models.Recycling, models.Use, models.Refurbish, models.Management]
), ),
) )
def test_status_in_temporary_lot(action_model: models.Action, user: UserClient, app: Devicehub): def test_status_in_temporary_lot(
action_model: models.Action, user: UserClient, app: Devicehub
):
"""Test of status actions for devices in a temporary lot.""" """Test of status actions for devices in a temporary lot."""
snap, _ = user.post(file('basic.snapshot'), res=models.Snapshot) snap, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
abstract = Device.query.filter_by(id=snap['device']['id']).first() abstract = Device.query.filter_by(id=snap['device']['id']).first()
@ -727,7 +726,7 @@ def test_live_without_TestDataStorage(user: UserClient, client: Client, app: Dev
acer = file('acer.happy.battery.snapshot') acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot) snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id'] device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=device_id).one()
post_request = { post_request = {
"transaction": "ccc", "transaction": "ccc",
"name": "John", "name": "John",
@ -767,7 +766,7 @@ def test_live_without_hdd_1(user: UserClient, client: Client, app: Devicehub):
acer = file('acer.happy.battery.snapshot') acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot) snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id'] device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=device_id).one()
post_request = { post_request = {
"transaction": "ccc", "transaction": "ccc",
"name": "John", "name": "John",
@ -803,7 +802,7 @@ def test_live_without_hdd_2(user: UserClient, client: Client, app: Devicehub):
acer['components'] = components acer['components'] = components
snapshot, _ = user.post(json_encode(acer), res=models.Snapshot) snapshot, _ = user.post(json_encode(acer), res=models.Snapshot)
device_id = snapshot['device']['id'] device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=device_id).one()
post_request = { post_request = {
"transaction": "ccc", "transaction": "ccc",
"name": "John", "name": "John",
@ -838,7 +837,7 @@ def test_live_without_hdd_3(user: UserClient, client: Client, app: Devicehub):
acer['components'] = components acer['components'] = components
snapshot, _ = user.post(json_encode(acer), res=models.Snapshot) snapshot, _ = user.post(json_encode(acer), res=models.Snapshot)
device_id = snapshot['device']['id'] device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=device_id).one()
post_request = { post_request = {
"transaction": "ccc", "transaction": "ccc",
"name": "John", "name": "John",
@ -875,7 +874,7 @@ def test_live_with_hdd_with_old_time(user: UserClient, client: Client, app: Devi
acer = file('acer.happy.battery.snapshot') acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot) snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id'] device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=device_id).one()
post_request = { post_request = {
"transaction": "ccc", "transaction": "ccc",
"name": "John", "name": "John",
@ -1022,7 +1021,7 @@ def test_allocate(user: UserClient):
allocate, _ = user.post(res=models.Allocate, data=post_request) allocate, _ = user.post(res=models.Allocate, data=post_request)
# Normal allocate # Normal allocate
device, _ = user.get(res=Device, item=devicehub_id) device, _ = user.get(res=Device, item=devicehub_id)
assert device['allocated'] == True assert device['allocated'] is True
action = [a for a in device['actions'] if a['type'] == 'Allocate'][0] action = [a for a in device['actions'] if a['type'] == 'Allocate'][0]
assert action['transaction'] == allocate['transaction'] assert action['transaction'] == allocate['transaction']
assert action['finalUserCode'] == allocate['finalUserCode'] assert action['finalUserCode'] == allocate['finalUserCode']
@ -1097,11 +1096,11 @@ def test_deallocate(user: UserClient):
user.post(res=models.Allocate, data=post_allocate) user.post(res=models.Allocate, data=post_allocate)
device, _ = user.get(res=Device, item=devicehub_id) device, _ = user.get(res=Device, item=devicehub_id)
assert device['allocated'] == True assert device['allocated'] is True
deallocate, _ = user.post(res=models.Deallocate, data=post_deallocate) deallocate, _ = user.post(res=models.Deallocate, data=post_deallocate)
assert deallocate['startTime'] == post_deallocate['startTime'] assert deallocate['startTime'] == post_deallocate['startTime']
assert deallocate['devices'][0]['id'] == device_id assert deallocate['devices'][0]['id'] == device_id
assert deallocate['devices'][0]['allocated'] == False assert deallocate['devices'][0]['allocated'] is False
res, _ = user.post(res=models.Deallocate, data=post_deallocate, status=422) res, _ = user.post(res=models.Deallocate, data=post_deallocate, status=422)
assert res['code'] == 422 assert res['code'] == 422
assert res['type'] == 'ValidationError' assert res['type'] == 'ValidationError'
@ -1204,8 +1203,8 @@ def test_offer_without_to(user: UserClient):
users = [ac.user for ac in trade.acceptances] users = [ac.user for ac in trade.acceptances]
assert trade.user_to == device.owner assert trade.user_to == device.owner
assert request_post['code'].lower() in device.owner.email assert request_post['code'].lower() in device.owner.email
assert device.owner.active == False assert device.owner.active is False
assert device.owner.phantom == True assert device.owner.phantom is True
assert trade.user_to in users assert trade.user_to in users
assert trade.user_from in users assert trade.user_from in users
assert device.owner.email != user.email assert device.owner.email != user.email
@ -1283,8 +1282,8 @@ def test_offer_without_from(user: UserClient, user2: UserClient):
phantom_user = trade.user_from phantom_user = trade.user_from
assert request_post['code'].lower() in phantom_user.email assert request_post['code'].lower() in phantom_user.email
assert phantom_user.active == False assert phantom_user.active is False
assert phantom_user.phantom == True assert phantom_user.phantom is True
# assert trade.confirm_transfer # assert trade.confirm_transfer
users = [ac.user for ac in trade.acceptances] users = [ac.user for ac in trade.acceptances]
@ -1778,12 +1777,12 @@ def test_confirmRevoke(user: UserClient, user2: UserClient):
assert device_10 in trade.devices assert device_10 in trade.devices
assert len(trade.devices) == 10 assert len(trade.devices) == 10
# the SCRAP confirms the revoke action # TODO??? the SCRAP confirms the revoke action
request_confirm_revoke = { # request_confirm_revoke = {
'type': 'ConfirmRevoke', # 'type': 'ConfirmRevoke',
'action': device_10.actions[-2].id, # 'action': device_10.actions[-2].id,
'devices': [snap10['device']['id']], # 'devices': [snap10['device']['id']],
} # }
# check validation error # check validation error
# user2.post(res=models.Action, data=request_confirm_revoke, status=422) # user2.post(res=models.Action, data=request_confirm_revoke, status=422)
@ -2796,7 +2795,7 @@ def test_action_web_erase(user: UserClient, client: Client):
) )
assert "alert alert-info" in response assert "alert alert-info" in response
assert "100% coincidence." in response assert "100% coincidence." in response
assert not "alert alert-danger" in response assert "alert alert-danger" not in response
@pytest.mark.mvp @pytest.mark.mvp
@ -2805,7 +2804,7 @@ def test_moveOnDocument(user: UserClient, user2: UserClient):
lotIn, _ = user.post({'name': 'MyLotIn'}, res=Lot) lotIn, _ = user.post({'name': 'MyLotIn'}, res=Lot)
lotOut, _ = user.post({'name': 'MyLotOut'}, res=Lot) lotOut, _ = user.post({'name': 'MyLotOut'}, res=Lot)
url = ( url = (
'http://www.ereuse.org/apapaapaapaapaapaapaapaapaapaapapaapaapaapaapaapaapaapaapaapapaapaapaapaapaapaapaapaapaapaaaa', 'http://www.ereuse.org/apapaapaapaapaapaapaapaapaapaapapaapaapaapaapaapaapaapaapaapapaapaapaapaapaapaapaapaap',
) )
request_post1 = { request_post1 = {
'filename': 'test.pdf', 'filename': 'test.pdf',
@ -2887,7 +2886,7 @@ def test_delete_devices(user: UserClient):
assert action_delete.t == 'Delete' assert action_delete.t == 'Delete'
assert str(action_delete.id) == action['id'] assert str(action_delete.id) == action['id']
assert db_device.active == False assert db_device.active is False
# Check use of filter from frontend # Check use of filter from frontend
url = '/devices/?filter={"type":["Computer"]}' url = '/devices/?filter={"type":["Computer"]}'
@ -2953,7 +2952,6 @@ def test_delete_devices_permitions(user: UserClient, user2: UserClient):
file_snap = file('1-device-with-components.snapshot') file_snap = file('1-device-with-components.snapshot')
snap, _ = user.post(file_snap, res=models.Snapshot) snap, _ = user.post(file_snap, res=models.Snapshot)
device = Device.query.filter_by(id=snap['device']['id']).one()
request = { request = {
'type': 'Delete', 'type': 'Delete',
@ -2973,7 +2971,7 @@ def test_moveOnDocument_bug168(user: UserClient, user2: UserClient):
lotIn, _ = user.post({'name': 'MyLotIn'}, res=Lot) lotIn, _ = user.post({'name': 'MyLotIn'}, res=Lot)
lotOut, _ = user.post({'name': 'MyLotOut'}, res=Lot) lotOut, _ = user.post({'name': 'MyLotOut'}, res=Lot)
url = ( url = (
'http://www.ereuse.org/apapaapaapaapaapaapaapaapaapaapapaapaapaapaapaapaapaapaapaapapaapaapaapaapaapaapaapaapaapaaaa', 'http://www.ereuse.org/apapaapaapaapaapaapaapaapaapaapapaapaapaapaapaapaapaapaapaapapaapaapaapaapaapaapaapaap',
) )
request_post1 = { request_post1 = {
'filename': 'test.pdf', 'filename': 'test.pdf',
@ -3014,13 +3012,12 @@ def test_moveOnDocument_bug168(user: UserClient, user2: UserClient):
'container_to': id_hash, 'container_to': id_hash,
'description': description, 'description': description,
} }
doc, _ = user.post(res=models.Action, data=request_moveOn) user.post(res=models.Action, data=request_moveOn)
trade = models.Trade.query.one()
trade_document1 = TradeDocument.query.filter_by(id=tradedocument_from['id']).one() trade_document1 = TradeDocument.query.filter_by(id=tradedocument_from['id']).one()
trade_document2 = TradeDocument.query.filter_by(id=tradedocument_to['id']).one() trade_document2 = TradeDocument.query.filter_by(id=tradedocument_to['id']).one()
assert trade_document1.total_weight == 150.0 assert trade_document1.total_weight == 150.0
assert trade_document2.total_weight == 4.0 assert trade_document2.total_weight == 4.0
assert trade_document1.trading == 'Confirm' assert trade_document1.trading == 'Confirm'
assert trade_document2.trading == None assert trade_document2.trading is None
tradedocument, _ = user.delete(res=TradeDocument, item=tradedocument_to['id']) tradedocument, _ = user.delete(res=TradeDocument, item=tradedocument_to['id'])

View File

@ -21,14 +21,22 @@ def test_authenticate_success(app: Devicehub):
def test_authenticate_error(app: Devicehub): def test_authenticate_error(app: Devicehub):
"""Tests the authenticate method with wrong token values.""" """Tests the authenticate method with wrong token values."""
with app.app_context(): with app.app_context():
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('Provide a suitable token.')
@pytest.mark.mvp
def test_authenticate_error_malformed_token(app: Devicehub):
"""Tests the authenticate method with malformed token."""
with app.app_context():
create_user()
# 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('Provide a suitable token.')
@pytest.mark.mvp @pytest.mark.mvp
@ -36,4 +44,6 @@ def test_auth_view(user: UserClient, client: Client):
"""Tests authentication at endpoint / view.""" """Tests authentication at endpoint / view."""
user.get(res='User', item=user.user['id'], status=200) user.get(res='User', item=user.user['id'], status=200)
client.get(res='User', item=user.user['id'], status=Unauthorized) client.get(res='User', item=user.user['id'], status=Unauthorized)
client.get(res='User', item=user.user['id'], token='wrong token', status=Unauthorized) client.get(
res='User', item=user.user['id'], token='wrong token', status=Unauthorized
)

View File

@ -1,13 +1,12 @@
import copy import copy
import datetime import datetime
import pytest
from uuid import UUID from uuid import UUID
from flask import g
import pytest
from colour import Color from colour import Color
from ereuse_utils.naming import Naming from ereuse_utils.naming import Naming
from ereuse_utils.test import ANY from ereuse_utils.test import ANY
from flask import g
from pytest import raises from pytest import raises
from sqlalchemy.util import OrderedSet from sqlalchemy.util import OrderedSet
from teal.db import ResourceNotFound from teal.db import ResourceNotFound
@ -20,26 +19,35 @@ from ereuse_devicehub.resources.action import models as m
from ereuse_devicehub.resources.action.models import Remove, TestConnectivity from ereuse_devicehub.resources.action.models import Remove, TestConnectivity
from ereuse_devicehub.resources.agent.models import Person from ereuse_devicehub.resources.agent.models import Person
from ereuse_devicehub.resources.device import models as d from ereuse_devicehub.resources.device import models as d
from ereuse_devicehub.resources.device.exceptions import NeedsId
from ereuse_devicehub.resources.device.schemas import Device as DeviceS from ereuse_devicehub.resources.device.schemas import Device as DeviceS
from ereuse_devicehub.resources.device.sync import MismatchBetweenTags, MismatchBetweenTagsAndHid, \ from ereuse_devicehub.resources.device.sync import (
Sync MismatchBetweenTags,
from ereuse_devicehub.resources.enums import ComputerChassis, DisplayTech, Severity, \ MismatchBetweenTagsAndHid,
SnapshotSoftware, TransferState Sync,
)
from ereuse_devicehub.resources.enums import (
ComputerChassis,
DisplayTech,
Severity,
SnapshotSoftware,
TransferState,
)
from ereuse_devicehub.resources.tag.model import Tag from ereuse_devicehub.resources.tag.model import Tag
from ereuse_devicehub.resources.user import User from ereuse_devicehub.resources.user import User
from tests import conftest from tests import conftest
from tests.conftest import file, yaml2json, json_encode from tests.conftest import file, json_encode, yaml2json
@pytest.mark.mvp @pytest.mark.mvp
@pytest.mark.usefixtures(conftest.auth_app_context.__name__) @pytest.mark.usefixtures(conftest.auth_app_context.__name__)
def test_device_model(): def test_device_model():
"""Tests that the correctness of the device model and its relationships.""" """Tests that the correctness of the device model and its relationships."""
pc = d.Desktop(model='p1mo', pc = d.Desktop(
model='p1mo',
manufacturer='p1ma', manufacturer='p1ma',
serial_number='p1s', serial_number='p1s',
chassis=ComputerChassis.Tower) chassis=ComputerChassis.Tower,
)
net = d.NetworkAdapter(model='c1mo', manufacturer='c1ma', serial_number='c1s') net = d.NetworkAdapter(model='c1mo', manufacturer='c1ma', serial_number='c1s')
graphic = d.GraphicCard(model='c2mo', manufacturer='c2ma', memory=1500) graphic = d.GraphicCard(model='c2mo', manufacturer='c2ma', memory=1500)
pc.components.add(net) pc.components.add(net)
@ -55,7 +63,9 @@ def test_device_model():
# Removing a component from pc doesn't delete the component # Removing a component from pc doesn't delete the component
pc.components.remove(net) pc.components.remove(net)
db.session.commit() db.session.commit()
pc = d.Device.query.filter_by(id=pc.id).first() # this is the same as querying for d.Desktop directly pc = d.Device.query.filter_by(
id=pc.id
).first() # this is the same as querying for d.Desktop directly
assert pc.components == {graphic} assert pc.components == {graphic}
network_adapter = d.NetworkAdapter.query.one() network_adapter = d.NetworkAdapter.query.one()
assert network_adapter not in pc.components assert network_adapter not in pc.components
@ -72,7 +82,9 @@ def test_device_model():
assert network_adapter.id == 4 assert network_adapter.id == 4
assert d.NetworkAdapter.query.first() is not None, 'We removed the network adaptor' assert d.NetworkAdapter.query.first() is not None, 'We removed the network adaptor'
assert gcard.id == 5, 'We should still hold a reference to a zombie graphic card' assert gcard.id == 5, 'We should still hold a reference to a zombie graphic card'
assert d.GraphicCard.query.first() is None, 'We should have deleted it it was inside the pc' assert (
d.GraphicCard.query.first() is None
), 'We should have deleted it it was inside the pc'
@pytest.mark.xfail(reason='Test not developed') @pytest.mark.xfail(reason='Test not developed')
@ -92,21 +104,25 @@ def test_device_schema():
@pytest.mark.mvp @pytest.mark.mvp
@pytest.mark.usefixtures(conftest.auth_app_context.__name__) @pytest.mark.usefixtures(conftest.auth_app_context.__name__)
def test_physical_properties(): def test_physical_properties():
c = d.Motherboard(slots=2, c = d.Motherboard(
slots=2,
usb=3, usb=3,
serial_number='sn', serial_number='sn',
model='ml', model='ml',
manufacturer='mr', manufacturer='mr',
width=2.0, width=2.0,
color=Color()) color=Color(),
pc = d.Desktop(chassis=ComputerChassis.Tower, )
pc = d.Desktop(
chassis=ComputerChassis.Tower,
model='foo', model='foo',
manufacturer='bar', manufacturer='bar',
serial_number='foo-bar', serial_number='foo-bar',
weight=2.8, weight=2.8,
width=1.4, width=1.4,
height=2.1, height=2.1,
color=Color('LightSeaGreen')) color=Color('LightSeaGreen'),
)
pc.components.add(c) pc.components.add(c)
db.session.add(pc) db.session.add(pc)
db.session.commit() db.session.commit()
@ -122,7 +138,7 @@ def test_physical_properties():
'manufacturer': 'mr', 'manufacturer': 'mr',
'bios_date': None, 'bios_date': None,
'ram_max_size': None, 'ram_max_size': None,
'ram_slots': None 'ram_slots': None,
} }
assert pc.physical_properties == { assert pc.physical_properties == {
'chassis': ComputerChassis.Tower, 'chassis': ComputerChassis.Tower,
@ -132,7 +148,7 @@ def test_physical_properties():
'receiver_id': None, 'receiver_id': None,
'serial_number': 'foo-bar', 'serial_number': 'foo-bar',
'part_number': None, 'part_number': None,
'transfer_state': TransferState.Initial 'transfer_state': TransferState.Initial,
} }
@ -142,14 +158,19 @@ def test_component_similar_one():
user = User.query.filter().first() user = User.query.filter().first()
snapshot = yaml2json('pc-components.db') snapshot = yaml2json('pc-components.db')
pc = snapshot['device'] pc = snapshot['device']
snapshot['components'][0]['serial_number'] = snapshot['components'][1]['serial_number'] = None snapshot['components'][0]['serial_number'] = snapshot['components'][1][
pc = d.Desktop(**pc, components=OrderedSet(d.Component(**c) for c in snapshot['components'])) 'serial_number'
] = None
pc = d.Desktop(
**pc, components=OrderedSet(d.Component(**c) for c in snapshot['components'])
)
component1, component2 = pc.components # type: d.Component component1, component2 = pc.components # type: d.Component
db.session.add(pc) db.session.add(pc)
db.session.flush() db.session.flush()
# Let's create a new component named 'A' similar to 1 # Let's create a new component named 'A' similar to 1
componentA = d.Component(model=component1.model, manufacturer=component1.manufacturer, componentA = d.Component(
owner_id=user.id) model=component1.model, manufacturer=component1.manufacturer, owner_id=user.id
)
similar_to_a = componentA.similar_one(pc, set()) similar_to_a = componentA.similar_one(pc, set())
assert similar_to_a == component1 assert similar_to_a == component1
# d.Component B does not have the same model # d.Component B does not have the same model
@ -175,9 +196,11 @@ def test_add_remove():
pc = d.Desktop(**pc, components=OrderedSet([c1, c2])) pc = d.Desktop(**pc, components=OrderedSet([c1, c2]))
db.session.add(pc) db.session.add(pc)
c3 = d.Component(serial_number='nc1', owner_id=user.id) c3 = d.Component(serial_number='nc1', owner_id=user.id)
pc2 = d.Desktop(serial_number='s2', pc2 = d.Desktop(
serial_number='s2',
components=OrderedSet([c3]), components=OrderedSet([c3]),
chassis=ComputerChassis.Microtower) chassis=ComputerChassis.Microtower,
)
c4 = d.Component(serial_number='c4s', owner_id=user.id) c4 = d.Component(serial_number='c4s', owner_id=user.id)
db.session.add(pc2) db.session.add(pc2)
db.session.add(c4) db.session.add(c4)
@ -201,7 +224,9 @@ def test_sync_run_components_empty():
remove all the components from the device. remove all the components from the device.
""" """
s = yaml2json('pc-components.db') s = yaml2json('pc-components.db')
pc = d.Desktop(**s['device'], components=OrderedSet(d.Component(**c) for c in s['components'])) pc = d.Desktop(
**s['device'], components=OrderedSet(d.Component(**c) for c in s['components'])
)
db.session.add(pc) db.session.add(pc)
db.session.commit() db.session.commit()
@ -219,7 +244,9 @@ def test_sync_run_components_none():
keep all the components from the device. keep all the components from the device.
""" """
s = yaml2json('pc-components.db') s = yaml2json('pc-components.db')
pc = d.Desktop(**s['device'], components=OrderedSet(d.Component(**c) for c in s['components'])) pc = d.Desktop(
**s['device'], components=OrderedSet(d.Component(**c) for c in s['components'])
)
db.session.add(pc) db.session.add(pc)
db.session.commit() db.session.commit()
@ -249,7 +276,8 @@ def test_sync_execute_register_desktop_existing_no_tag():
db.session.commit() db.session.commit()
pc = d.Desktop( pc = d.Desktop(
**yaml2json('pc-components.db')['device']) # Create a new transient non-db object **yaml2json('pc-components.db')['device']
) # Create a new transient non-db object
# 1: device exists on DB # 1: device exists on DB
db_pc = Sync().execute_register(pc) db_pc = Sync().execute_register(pc)
pc.amount = 0 pc.amount = 0
@ -285,7 +313,9 @@ def test_sync_execute_register_desktop_tag_not_linked():
db.session.commit() db.session.commit()
# Create a new transient non-db object # Create a new transient non-db object
pc = d.Desktop(**yaml2json('pc-components.db')['device'], tags=OrderedSet([Tag(id='foo')])) pc = d.Desktop(
**yaml2json('pc-components.db')['device'], tags=OrderedSet([Tag(id='foo')])
)
returned_pc = Sync().execute_register(pc) returned_pc = Sync().execute_register(pc)
assert returned_pc == pc assert returned_pc == pc
assert tag.device == pc, 'Tag has to be linked' assert tag.device == pc, 'Tag has to be linked'
@ -326,7 +356,9 @@ def test_sync_execute_register_tag_does_not_exist():
Tags have to be created before trying to link them through a Snapshot. Tags have to be created before trying to link them through a Snapshot.
""" """
user = User.query.filter().first() user = User.query.filter().first()
pc = d.Desktop(**yaml2json('pc-components.db')['device'], tags=OrderedSet([Tag('foo')])) pc = d.Desktop(
**yaml2json('pc-components.db')['device'], tags=OrderedSet([Tag('foo')])
)
pc.owner_id = user.id pc.owner_id = user.id
with raises(ResourceNotFound): with raises(ResourceNotFound):
Sync().execute_register(pc) Sync().execute_register(pc)
@ -345,7 +377,8 @@ def test_sync_execute_register_tag_linked_same_device():
db.session.commit() db.session.commit()
pc = d.Desktop( pc = d.Desktop(
**yaml2json('pc-components.db')['device']) # Create a new transient non-db object **yaml2json('pc-components.db')['device']
) # Create a new transient non-db object
pc.tags.add(Tag(id='foo')) pc.tags.add(Tag(id='foo'))
db_pc = Sync().execute_register(pc) db_pc = Sync().execute_register(pc)
assert db_pc.id == orig_pc.id assert db_pc.id == orig_pc.id
@ -369,7 +402,8 @@ def test_sync_execute_register_tag_linked_other_device_mismatch_between_tags():
db.session.commit() db.session.commit()
pc1 = d.Desktop( pc1 = d.Desktop(
**yaml2json('pc-components.db')['device']) # Create a new transient non-db object **yaml2json('pc-components.db')['device']
) # Create a new transient non-db object
pc1.tags.add(Tag(id='foo-1')) pc1.tags.add(Tag(id='foo-1'))
pc1.tags.add(Tag(id='foo-2')) pc1.tags.add(Tag(id='foo-2'))
with raises(MismatchBetweenTags): with raises(MismatchBetweenTags):
@ -393,7 +427,8 @@ def test_sync_execute_register_mismatch_between_tags_and_hid():
db.session.commit() db.session.commit()
pc1 = d.Desktop( pc1 = d.Desktop(
**yaml2json('pc-components.db')['device']) # Create a new transient non-db object **yaml2json('pc-components.db')['device']
) # Create a new transient non-db object
pc1.tags.add(Tag(id='foo-2')) pc1.tags.add(Tag(id='foo-2'))
with raises(MismatchBetweenTagsAndHid): with raises(MismatchBetweenTagsAndHid):
Sync().execute_register(pc1) Sync().execute_register(pc1)
@ -404,22 +439,36 @@ def test_sync_execute_register_mismatch_between_tags_and_hid():
def test_get_device(user: UserClient): def test_get_device(user: UserClient):
"""Checks GETting a d.Desktop with its components.""" """Checks GETting a d.Desktop with its components."""
g.user = User.query.one() g.user = User.query.one()
pc = d.Desktop(model='p1mo', pc = d.Desktop(
model='p1mo',
manufacturer='p1ma', manufacturer='p1ma',
serial_number='p1s', serial_number='p1s',
chassis=ComputerChassis.Tower, chassis=ComputerChassis.Tower,
owner_id=user.user['id']) owner_id=user.user['id'],
pc.components = OrderedSet([ )
d.NetworkAdapter(model='c1mo', manufacturer='c1ma', serial_number='c1s', pc.components = OrderedSet(
owner_id=user.user['id']), [
d.GraphicCard(model='c2mo', manufacturer='c2ma', memory=1500, owner_id=user.user['id']) d.NetworkAdapter(
]) model='c1mo',
manufacturer='c1ma',
serial_number='c1s',
owner_id=user.user['id'],
),
d.GraphicCard(
model='c2mo', manufacturer='c2ma', memory=1500, owner_id=user.user['id']
),
]
)
db.session.add(pc) db.session.add(pc)
# todo test is an abstract class. replace with another one # todo test is an abstract class. replace with another one
db.session.add(TestConnectivity(device=pc, db.session.add(
TestConnectivity(
device=pc,
severity=Severity.Info, severity=Severity.Info,
agent=Person(name='Timmy'), agent=Person(name='Timmy'),
author=User(email='bar@bar.com'))) author=User(email='bar@bar.com'),
)
)
db.session.commit() db.session.commit()
pc_api, _ = user.get(res=d.Device, item=pc.devicehub_id) pc_api, _ = user.get(res=d.Device, item=pc.devicehub_id)
assert len(pc_api['actions']) == 1 assert len(pc_api['actions']) == 1
@ -427,10 +476,14 @@ def test_get_device(user: UserClient):
assert pc_api['actions'][0]['device'] == pc.id assert pc_api['actions'][0]['device'] == pc.id
assert pc_api['actions'][0]['severity'] == 'Info' assert pc_api['actions'][0]['severity'] == 'Info'
assert UUID(pc_api['actions'][0]['author']) assert UUID(pc_api['actions'][0]['author'])
assert 'actions_components' not in pc_api, 'actions_components are internal use only' assert (
'actions_components' not in pc_api
), 'actions_components are internal use only'
assert 'actions_one' not in pc_api, 'they are internal use only' assert 'actions_one' not in pc_api, 'they are internal use only'
assert 'author' not in pc_api assert 'author' not in pc_api
assert tuple(c['id'] for c in pc_api['components']) == tuple(c.id for c in pc.components) assert tuple(c['id'] for c in pc_api['components']) == tuple(
c.id for c in pc.components
)
assert pc_api['hid'] == 'desktop-p1ma-p1mo-p1s' assert pc_api['hid'] == 'desktop-p1ma-p1mo-p1s'
assert pc_api['model'] == 'p1mo' assert pc_api['model'] == 'p1mo'
assert pc_api['manufacturer'] == 'p1ma' assert pc_api['manufacturer'] == 'p1ma'
@ -443,41 +496,59 @@ def test_get_device(user: UserClient):
def test_get_devices(app: Devicehub, user: UserClient): def test_get_devices(app: Devicehub, user: UserClient):
"""Checks GETting multiple devices.""" """Checks GETting multiple devices."""
g.user = User.query.one() g.user = User.query.one()
pc = d.Desktop(model='p1mo', pc = d.Desktop(
model='p1mo',
manufacturer='p1ma', manufacturer='p1ma',
serial_number='p1s', serial_number='p1s',
chassis=ComputerChassis.Tower, chassis=ComputerChassis.Tower,
owner_id=user.user['id']) owner_id=user.user['id'],
pc.components = OrderedSet([ )
d.NetworkAdapter(model='c1mo', manufacturer='c1ma', serial_number='c1s', pc.components = OrderedSet(
owner_id=user.user['id']), [
d.GraphicCard(model='c2mo', manufacturer='c2ma', memory=1500, d.NetworkAdapter(
owner_id=user.user['id']) model='c1mo',
]) manufacturer='c1ma',
pc1 = d.Desktop(model='p2mo', serial_number='c1s',
owner_id=user.user['id'],
),
d.GraphicCard(
model='c2mo', manufacturer='c2ma', memory=1500, owner_id=user.user['id']
),
]
)
pc1 = d.Desktop(
model='p2mo',
manufacturer='p2ma', manufacturer='p2ma',
serial_number='p2s', serial_number='p2s',
chassis=ComputerChassis.Tower, chassis=ComputerChassis.Tower,
owner_id=user.user['id']) owner_id=user.user['id'],
pc2 = d.Laptop(model='p3mo', )
pc2 = d.Laptop(
model='p3mo',
manufacturer='p3ma', manufacturer='p3ma',
serial_number='p3s', serial_number='p3s',
chassis=ComputerChassis.Netbook, chassis=ComputerChassis.Netbook,
owner_id=user.user['id']) owner_id=user.user['id'],
)
db.session.add_all((pc, pc1, pc2)) db.session.add_all((pc, pc1, pc2))
db.session.commit() db.session.commit()
devices, _ = user.get(res=d.Device) devices, _ = user.get(res=d.Device)
ids = (pc.id, pc1.id, pc2.id, pc.components[0].id, pc.components[1].id) ids = (pc.id, pc1.id, pc2.id, pc.components[0].id, pc.components[1].id)
assert tuple(dev['id'] for dev in devices['items']) == ids assert tuple(dev['id'] for dev in devices['items']) == ids
assert tuple(dev['type'] for dev in devices['items']) == ( assert tuple(dev['type'] for dev in devices['items']) == (
d.Desktop.t, d.Desktop.t, d.Laptop.t, d.NetworkAdapter.t, d.GraphicCard.t d.Desktop.t,
d.Desktop.t,
d.Laptop.t,
d.NetworkAdapter.t,
d.GraphicCard.t,
) )
@pytest.mark.mvp @pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__) @pytest.mark.usefixtures(conftest.app_context.__name__)
def test_get_device_permissions(app: Devicehub, user: UserClient, user2: UserClient, def test_get_device_permissions(
client: Client): app: Devicehub, user: UserClient, user2: UserClient, client: Client
):
"""Checks GETting a d.Desktop with its components.""" """Checks GETting a d.Desktop with its components."""
s, _ = user.post(file('asus-eee-1000h.snapshot.11'), res=m.Snapshot) s, _ = user.post(file('asus-eee-1000h.snapshot.11'), res=m.Snapshot)
@ -529,12 +600,12 @@ def test_get_devices_unassigned(user: UserClient):
assert len(devices['items']) == 2 assert len(devices['items']) == 2
from ereuse_devicehub.resources.lot.models import Lot from ereuse_devicehub.resources.lot.models import Lot
device_id = devices['items'][0]['id'] device_id = devices['items'][0]['id']
my_lot, _ = user.post(({'name': 'My_lot'}), res=Lot) my_lot, _ = user.post(({'name': 'My_lot'}), res=Lot)
lot, _ = user.post({}, lot, _ = user.post(
res=Lot, {}, res=Lot, item='{}/devices'.format(my_lot['id']), query=[('id', device_id)]
item='{}/devices'.format(my_lot['id']), )
query=[('id', device_id)])
lot = Lot.query.filter_by(id=lot['id']).one() lot = Lot.query.filter_by(id=lot['id']).one()
assert next(iter(lot.devices)).id == device_id assert next(iter(lot.devices)).id == device_id
@ -554,13 +625,15 @@ def test_get_devices_unassigned(user: UserClient):
@pytest.mark.mvp @pytest.mark.mvp
@pytest.mark.usefixtures(conftest.auth_app_context.__name__) @pytest.mark.usefixtures(conftest.auth_app_context.__name__)
def test_computer_monitor(): def test_computer_monitor():
m = d.ComputerMonitor(technology=DisplayTech.LCD, m = d.ComputerMonitor(
technology=DisplayTech.LCD,
manufacturer='foo', manufacturer='foo',
model='bar', model='bar',
serial_number='foo-bar', serial_number='foo-bar',
resolution_width=1920, resolution_width=1920,
resolution_height=1080, resolution_height=1080,
size=14.5) size=14.5,
)
db.session.add(m) db.session.add(m)
db.session.commit() db.session.commit()
@ -568,9 +641,11 @@ def test_computer_monitor():
@pytest.mark.mvp @pytest.mark.mvp
def test_manufacturer(user: UserClient): def test_manufacturer(user: UserClient):
m, r = user.get(res='Manufacturer', query=[('search', 'asus')]) m, r = user.get(res='Manufacturer', query=[('search', 'asus')])
assert m == {'items': [{'name': 'Asus', 'url': 'https://en.wikipedia.org/wiki/Asus'}]} assert m == {
'items': [{'name': 'Asus', 'url': 'https://en.wikipedia.org/wiki/Asus'}]
}
assert r.cache_control.public assert r.cache_control.public
assert r.expires > datetime.datetime.now() assert r.expires.timestamp() > datetime.datetime.now().timestamp()
@pytest.mark.mvp @pytest.mark.mvp
@ -591,12 +666,20 @@ def test_device_properties_format(app: Devicehub, user: UserClient):
assert format(pc, 's') == '(asustek computer inc.) S/N 94OAAQ021116' assert format(pc, 's') == '(asustek computer inc.) S/N 94OAAQ021116'
assert pc.ram_size == 1024 assert pc.ram_size == 1024
assert pc.data_storage_size == 152627 assert pc.data_storage_size == 152627
assert pc.graphic_card_model == 'mobile 945gse express integrated graphics controller' assert (
pc.graphic_card_model
== 'mobile 945gse express integrated graphics controller'
)
assert pc.processor_model == 'intel atom cpu n270 @ 1.60ghz' assert pc.processor_model == 'intel atom cpu n270 @ 1.60ghz'
net = next(c for c in pc.components if isinstance(c, d.NetworkAdapter)) net = next(c for c in pc.components if isinstance(c, d.NetworkAdapter))
assert format(net) == 'NetworkAdapter 5: model ar8121/ar8113/ar8114 ' \ assert (
format(net) == 'NetworkAdapter 5: model ar8121/ar8113/ar8114 '
'gigabit or fast ethernet, S/N 00:24:8c:7f:cf:2d' 'gigabit or fast ethernet, S/N 00:24:8c:7f:cf:2d'
assert format(net, 't') == 'NetworkAdapter ar8121/ar8113/ar8114 gigabit or fast ethernet' )
assert (
format(net, 't')
== 'NetworkAdapter ar8121/ar8113/ar8114 gigabit or fast ethernet'
)
assert format(net, 's') == 'qualcomm atheros 00:24:8C:7F:CF:2D 100 Mbps' assert format(net, 's') == 'qualcomm atheros 00:24:8C:7F:CF:2D 100 Mbps'
hdd = next(c for c in pc.components if isinstance(c, d.DataStorage)) hdd = next(c for c in pc.components if isinstance(c, d.DataStorage))
assert format(hdd) == 'HardDrive 10: model st9160310as, S/N 5sv4tqa6' assert format(hdd) == 'HardDrive 10: model st9160310as, S/N 5sv4tqa6'
@ -638,8 +721,12 @@ def test_networking_model(user: UserClient):
@pytest.mark.usefixtures(conftest.app_context.__name__) @pytest.mark.usefixtures(conftest.app_context.__name__)
def test_cooking_mixer(user: UserClient): def test_cooking_mixer(user: UserClient):
mixer = d.Mixer(serial_number='foo', model='bar', manufacturer='foobar', mixer = d.Mixer(
owner_id=user.user['id']) serial_number='foo',
model='bar',
manufacturer='foobar',
owner_id=user.user['id'],
)
db.session.add(mixer) db.session.add(mixer)
db.session.commit() db.session.commit()
@ -652,12 +739,12 @@ def test_cooking_mixer_api(user: UserClient):
'serialNumber': 'foo', 'serialNumber': 'foo',
'model': 'bar', 'model': 'bar',
'manufacturer': 'foobar', 'manufacturer': 'foobar',
'type': 'Mixer' 'type': 'Mixer',
}, },
'version': '11.0', 'version': '11.0',
'software': SnapshotSoftware.Web.name 'software': SnapshotSoftware.Web.name,
}, },
res=m.Snapshot res=m.Snapshot,
) )
mixer, _ = user.get(res=d.Device, item=snapshot['device']['id']) mixer, _ = user.get(res=d.Device, item=snapshot['device']['id'])
assert mixer['type'] == 'Mixer' assert mixer['type'] == 'Mixer'
@ -673,14 +760,19 @@ def test_hid_with_mac(app: Devicehub, user: UserClient):
pc, _ = user.get(res=d.Device, item=snap['device']['devicehubID']) pc, _ = user.get(res=d.Device, item=snap['device']['devicehubID'])
assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116' assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116'
pc = d.Device.query.filter_by(devicehub_id=snap['device']['devicehubID']).one() pc = d.Device.query.filter_by(devicehub_id=snap['device']['devicehubID']).one()
assert pc.placeholder.binding.hid == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d' assert (
pc.placeholder.binding.hid
== 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d'
)
@pytest.mark.mvp @pytest.mark.mvp
def test_hid_without_mac(app: Devicehub, user: UserClient): def test_hid_without_mac(app: Devicehub, user: UserClient):
"""Checks hid without mac.""" """Checks hid without mac."""
snapshot = yaml2json('asus-eee-1000h.snapshot.11') snapshot = yaml2json('asus-eee-1000h.snapshot.11')
snapshot['components'] = [c for c in snapshot['components'] if c['type'] != 'NetworkAdapter'] snapshot['components'] = [
c for c in snapshot['components'] if c['type'] != 'NetworkAdapter'
]
snap, _ = user.post(json_encode(snapshot), res=m.Snapshot) snap, _ = user.post(json_encode(snapshot), res=m.Snapshot)
pc, _ = user.get(res=d.Device, item=snap['device']['devicehubID']) pc, _ = user.get(res=d.Device, item=snap['device']['devicehubID'])
assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116' assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116'
@ -709,7 +801,10 @@ def test_hid_with_2networkadapters(app: Devicehub, user: UserClient):
devices, _ = user.get(res=d.Device) devices, _ = user.get(res=d.Device)
laptop = devices['items'][0] laptop = devices['items'][0]
assert laptop['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d' assert (
laptop['hid']
== 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d'
)
assert len([c for c in devices['items'] if c['type'] == 'Laptop']) == 2 assert len([c for c in devices['items'] if c['type'] == 'Laptop']) == 2
@ -726,14 +821,20 @@ def test_hid_with_2network_and_drop_no_mac_in_hid(app: Devicehub, user: UserClie
pc, _ = user.get(res=d.Device, item=snap['device']['devicehubID']) pc, _ = user.get(res=d.Device, item=snap['device']['devicehubID'])
assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116' assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116'
pc = d.Device.query.filter_by(devicehub_id=snap['device']['devicehubID']).one() pc = d.Device.query.filter_by(devicehub_id=snap['device']['devicehubID']).one()
assert pc.placeholder.binding.hid == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d' assert (
pc.placeholder.binding.hid
== 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d'
)
snapshot['uuid'] = 'd1b70cb8-8929-4f36-99b7-fe052cec0abb' snapshot['uuid'] = 'd1b70cb8-8929-4f36-99b7-fe052cec0abb'
snapshot['components'] = [c for c in snapshot['components'] if c != network] snapshot['components'] = [c for c in snapshot['components'] if c != network]
user.post(json_encode(snapshot), res=m.Snapshot) user.post(json_encode(snapshot), res=m.Snapshot)
devices, _ = user.get(res=d.Device) devices, _ = user.get(res=d.Device)
laptop = devices['items'][0] laptop = devices['items'][0]
assert pc.placeholder.binding.hid == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d' assert (
pc.placeholder.binding.hid
== 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d'
)
assert len([c for c in devices['items'] if c['type'] == 'Laptop']) == 2 assert len([c for c in devices['items'] if c['type'] == 'Laptop']) == 2
assert len([c for c in laptop['components'] if c['type'] == 'NetworkAdapter']) == 1 assert len([c for c in laptop['components'] if c['type'] == 'NetworkAdapter']) == 1
@ -752,7 +853,10 @@ def test_hid_with_2network_and_drop_mac_in_hid(app: Devicehub, user: UserClient)
pc, _ = user.get(res=d.Device, item=snap['device']['devicehubID']) pc, _ = user.get(res=d.Device, item=snap['device']['devicehubID'])
assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116' assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116'
pc = d.Device.query.filter_by(devicehub_id=snap['device']['devicehubID']).one() pc = d.Device.query.filter_by(devicehub_id=snap['device']['devicehubID']).one()
assert pc.placeholder.binding.hid == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d' assert (
pc.placeholder.binding.hid
== 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d'
)
# we drop the network card then is used for to build the hid # we drop the network card then is used for to build the hid
snapshot['uuid'] = 'd1b70cb8-8929-4f36-99b7-fe052cec0abb' snapshot['uuid'] = 'd1b70cb8-8929-4f36-99b7-fe052cec0abb'
@ -762,19 +866,25 @@ def test_hid_with_2network_and_drop_mac_in_hid(app: Devicehub, user: UserClient)
laptops = [c for c in devices['items'] if c['type'] == 'Laptop'] laptops = [c for c in devices['items'] if c['type'] == 'Laptop']
assert len(laptops) == 4 assert len(laptops) == 4
hids = [laptops[0]['hid'], laptops[2]['hid']] hids = [laptops[0]['hid'], laptops[2]['hid']]
proof_hid = ['laptop-asustek_computer_inc-1000h-94oaaq021116-a0:24:8c:7f:cf:2d', proof_hid = [
'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d'] 'laptop-asustek_computer_inc-1000h-94oaaq021116-a0:24:8c:7f:cf:2d',
'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d',
]
assert all([h in proof_hid for h in hids]) assert all([h in proof_hid for h in hids])
# we drop all network cards # we drop all network cards
snapshot['uuid'] = 'd1b70cb8-8929-4f36-99b7-fe052cec0abc' snapshot['uuid'] = 'd1b70cb8-8929-4f36-99b7-fe052cec0abc'
snapshot['components'] = [c for c in snapshot['components'] if c not in [network, network2]] snapshot['components'] = [
c for c in snapshot['components'] if c not in [network, network2]
]
user.post(json_encode(snapshot), res=m.Snapshot) user.post(json_encode(snapshot), res=m.Snapshot)
devices, _ = user.get(res=d.Device) devices, _ = user.get(res=d.Device)
laptops = [c for c in devices['items'] if c['type'] == 'Laptop'] laptops = [c for c in devices['items'] if c['type'] == 'Laptop']
assert len(laptops) == 4 assert len(laptops) == 4
hids = [laptops[0]['hid'], laptops[2]['hid']] hids = [laptops[0]['hid'], laptops[2]['hid']]
proof_hid = ['laptop-asustek_computer_inc-1000h-94oaaq021116-a0:24:8c:7f:cf:2d', proof_hid = [
'laptop-asustek_computer_inc-1000h-94oaaq021116-a0:24:8c:7f:cf:2d',
'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d', 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d',
'laptop-asustek_computer_inc-1000h-94oaaq021116'] 'laptop-asustek_computer_inc-1000h-94oaaq021116',
]
assert all([h in proof_hid for h in hids]) assert all([h in proof_hid for h in hids])

View File

@ -21,10 +21,12 @@ from tests.conftest import TestConfig
class NoExcCliRunner(click.testing.CliRunner): class NoExcCliRunner(click.testing.CliRunner):
"""Runner that interfaces with the Devicehub CLI.""" """Runner that interfaces with the Devicehub CLI."""
def invoke(self, *args, input=None, env=None, catch_exceptions=False, color=False, def invoke(
**extra): self, *args, input=None, env=None, catch_exceptions=False, color=False, **extra
r = super().invoke(ereuse_devicehub.cli.cli, ):
args, input, env, catch_exceptions, color, **extra) r = super().invoke(
ereuse_devicehub.cli.cli, args, input, env, catch_exceptions, color, **extra
)
assert r.exit_code == 0, 'CLI code {}: {}'.format(r.exit_code, r.output) assert r.exit_code == 0, 'CLI code {}: {}'.format(r.exit_code, r.output)
return r return r
@ -69,16 +71,26 @@ def test_inventory_create_delete_user(cli, tdb1, tdb2):
""" """
# Create first DB # Create first DB
cli.inv('tdb1') cli.inv('tdb1')
cli.invoke('inv', 'add', cli.invoke(
'-n', 'Test DB1', 'inv',
'-on', 'ACME DB1', 'add',
'-oi', 'acme-id', '-n',
'-tu', 'https://example.com', 'Test DB1',
'-tt', '3c66a6ad-22de-4db6-ac46-d8982522ec40', '-on',
'--common') 'ACME DB1',
'-oi',
'acme-id',
'-tu',
'https://example.com',
'-tt',
'3c66a6ad-22de-4db6-ac46-d8982522ec40',
'--common',
)
# Create an user for first DB # Create an user for first DB
cli.invoke('user', 'add', 'foo@foo.com', '-a', 'Foo', '-c', 'ES', '-p', 'Such password') cli.invoke(
'user', 'add', 'foo@foo.com', '-a', 'Foo', '-c', 'ES', '-p', 'Such password'
)
with tdb1.app_context(): with tdb1.app_context():
# There is a row for the inventory # There is a row for the inventory
@ -98,12 +110,20 @@ def test_inventory_create_delete_user(cli, tdb1, tdb2):
cli.inv('tdb2') cli.inv('tdb2')
# Create a second DB # Create a second DB
# Note how we don't create common anymore # Note how we don't create common anymore
cli.invoke('inv', 'add', cli.invoke(
'-n', 'Test DB2', 'inv',
'-on', 'ACME DB2', 'add',
'-oi', 'acme-id-2', '-n',
'-tu', 'https://example.com', 'Test DB2',
'-tt', 'fbad1c08-ffdc-4a61-be49-464962c186a8') '-on',
'ACME DB2',
'-oi',
'acme-id-2',
'-tu',
'https://example.com',
'-tt',
'fbad1c08-ffdc-4a61-be49-464962c186a8',
)
# Create an user for with access for both DB # Create an user for with access for both DB
cli.invoke('user', 'add', 'bar@bar.com', '-a', 'Bar', '-p', 'Wow password') cli.invoke('user', 'add', 'bar@bar.com', '-a', 'Bar', '-p', 'Wow password')
@ -144,5 +164,6 @@ 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.')