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

resolve conflict

This commit is contained in:
Cayo Puigdefabregas 2021-05-25 09:03:34 +02:00
parent c1008ae653 db8c7e6ce0
commit 1098ef773f
21 changed files with 332 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
ereuse_devicehub/auth.py
View File

@ -3,12 +3,17 @@ from teal.auth import TokenAuth
from teal.db import ResourceNotFound
from werkzeug.exceptions import Unauthorized
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.user.models import User, Session
class Auth(TokenAuth):
def authenticate(self, token: str, *args, **kw) -> User:
try:
return User.query.filter_by(token=token).one()
user = User.query.filter_by(token=token).first()
if user:
return user
ses = Session.query.filter_by(token=token).one()
return ses.user
except (ResourceNotFound, DataError):
raise Unauthorized('Provide a suitable token.')

1
ereuse_devicehub/config.py
View File

@ -37,6 +37,7 @@ class DevicehubConfig(Config):
DB_PASSWORD = config('DB_PASSWORD', 'ereuse')
DB_HOST = config('DB_HOST', 'localhost')
DB_DATABASE = config('DB_DATABASE', 'devicehub')
DB_SCHEMA = config('DB_SCHEMA', 'dbtest')
SQLALCHEMY_DATABASE_URI = 'postgresql://{user}:{pw}@{host}/{db}'.format(
user=DB_USER,
pw=DB_PASSWORD,

6
ereuse_devicehub/dummy/dummy.py
View File

@ -17,6 +17,8 @@ from ereuse_devicehub.resources.device.models import Device
from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.tag.model import Tag
from ereuse_devicehub.resources.user import User
from ereuse_devicehub.resources.user.models import Session
from ereuse_devicehub.resources.enums import SessionType
class Dummy:
@ -194,6 +196,10 @@ class Dummy:
user.individuals.add(Person(name=name))
db.session.add(user)
session_external = Session(user=user, type=SessionType.External)
session_internal = Session(user=user, type=SessionType.Internal)
db.session.add(session_internal)
db.session.add(session_external)
db.session.commit()
client = UserClient(self.app, user.email, password,

62
ereuse_devicehub/migrations/versions/21afd375a654_session_table.py Normal file
View File

@ -0,0 +1,62 @@
"""session_table
Revision ID: 21afd375a654
Revises: 6a2a939d5668
Create Date: 2021-04-13 11:18:27.720567
"""
from alembic import context
from alembic import op
from sqlalchemy.dialects import postgresql
import sqlalchemy as sa
import sqlalchemy_utils
import citext
import teal
from ereuse_devicehub.resources.enums import SessionType
# revision identifiers, used by Alembic.
revision = '21afd375a654'
down_revision = '398826453b39'
branch_labels = None
depends_on = None
comment_update = 'The last time Devicehub recorded a change for this thing.\n'
def get_inv():
INV = context.get_x_argument(as_dictionary=True).get('inventory')
if not INV:
raise ValueError("Inventory value is not specified")
return INV
def upgrade():
op.create_table('session',
sa.Column('updated', sa.TIMESTAMP(timezone=True),
server_default=sa.text('CURRENT_TIMESTAMP'),
nullable=False,
comment=comment_update),
sa.Column('created', sa.TIMESTAMP(timezone=True),
server_default=sa.text('CURRENT_TIMESTAMP'),
nullable=False, comment='When Devicehub created this.'),
sa.Column('id', sa.BigInteger(), nullable=False),
sa.Column('expired', sa.BigInteger(), nullable=True),
sa.Column('token', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('type', teal.db.IntEnum(SessionType), nullable=False),
sa.Column('user_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.ForeignKeyConstraint(['user_id'], ['common.user.id'], ),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('token'),
schema='common'
)
op.create_index(op.f('ix_session_created'), 'session', ['created'], unique=False, schema='common')
op.create_index(op.f('ix_session_updated'), 'session', ['updated'], unique=False, schema='common')
op.create_index(op.f('ix_session_token'), 'session', ['token'], unique=True, schema='common')
def downgrade():
op.drop_table('trade', schema=f'{get_inv()}')
op.drop_index(op.f('ix_session_created'), table_name='session', schema='common')
op.drop_index(op.f('ix_session_updated'), table_name='session', schema='common')
op.drop_index(op.f('ix_session_token'), table_name='session', schema='common')

37
ereuse_devicehub/resources/documents/documents.py
View File

@ -1,7 +1,8 @@
import csv
import datetime
import enum
import uuid
import datetime
import pathlib
from collections import OrderedDict
from io import StringIO
from typing import Callable, Iterable, Tuple
@ -18,7 +19,9 @@ from flask.json import jsonify
from teal.cache import cache
from teal.resource import Resource, View
from ereuse_devicehub import auth
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.enums import SessionType
from ereuse_devicehub.resources.action import models as evs
from ereuse_devicehub.resources.device import models as devs
from ereuse_devicehub.resources.deliverynote.models import Deliverynote
@ -253,7 +256,7 @@ class StampsView(View):
url = urlutils.URL(request.url)
url.normalize()
url.path_parts = url.path_parts[:-2] + ['check', '']
return url.to_text()
return url.to_text()
def get(self):
result = ('', '')
@ -314,6 +317,30 @@ class InternalStatsView(DeviceView):
return output
class WbConfDocumentView(DeviceView):
def get(self, wbtype: str):
if not wbtype.lower() in ['usodyrate', 'usodywipe']:
return jsonify('')
data = {'token': self.get_token(),
'host': app.config['DB_HOST'],
'inventory': app.config['DB_SCHEMA']
}
data['erase'] = False
# data['erase'] = True if wbtype == 'usodywipe' else False
env = flask.render_template('documents/wbSettings.ini', **data)
output = make_response(env)
output.headers['Content-Disposition'] = 'attachment; filename=settings.ini'
output.headers['Content-type'] = 'text/plain'
return output
def get_token(self):
tk = [s.token for s in g.user.sessions if s.type == SessionType.Internal][0]
token = auth.Auth.encode(tk)
return token
class DocumentDef(Resource):
__type__ = 'Document'
SCHEMA = None
@ -380,3 +407,9 @@ class DocumentDef(Resource):
auth=app.auth)
actions_view = app.auth.requires_auth(actions_view)
self.add_url_rule('/actions/', defaults=d, view_func=actions_view, methods=get)
wbconf_view = WbConfDocumentView.as_view('WbConfDocumentView',
definition=self,
auth=app.auth)
wbconf_view = app.auth.requires_auth(wbconf_view)
self.add_url_rule('/wbconf/<string:wbtype>', view_func=wbconf_view, methods=get)

17
ereuse_devicehub/resources/documents/templates/documents/wbSettings.ini Normal file
View File

@ -0,0 +1,17 @@
[settings]
DH_TOKEN="{{token}}"
DH_HOST="{{host}}"
DH_DATABASE="{{inventory}}"
DEVICEHUB_URL=https://${DB_HOST}/${DB_DATABASE}/
WB_BENCHMARK = False
WB_STRESS_TEST = 0
WB_SMART_TEST = ""
WB_ERASE = {{erase}}
WB_ERASE_STEPS = 1
WB_ERASE_LEADING_ZEROS = False
WB_DEBUG = True

21
ereuse_devicehub/resources/enums.py
View File

@ -393,3 +393,24 @@ class TransferState(IntEnum):
def __str__(self):
return self.name
@unique
class SessionType(IntEnum):
"""
Enumaration of types of sessions:
* Internal: permanent Session for internal user. This is used in usb of WorkBench.
* External: permanent Session for external users. This is used in usb of WorkBench.
* Session: This is used for keep more than one session in the app frontend.
Devicehub specially raises user awareness when an action
has a Severity of ``Warning`` or greater.
"""
Internal = 0
External = 1
Session = 2
def __str__(self):
return self.name

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

@ -7,7 +7,7 @@ from teal.resource import Converters, Resource
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.user import schemas
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.user.views import UserView, login
from ereuse_devicehub.resources.user.views import UserView, login, logout
class UserDef(Resource):
@ -23,6 +23,8 @@ class UserDef(Resource):
super().__init__(app, import_name, static_folder, static_url_path, template_folder,
url_prefix, subdomain, url_defaults, root_path, cli_commands)
self.add_url_rule('/login/', view_func=login, methods={'POST'})
logout_view = app.auth.requires_auth(logout)
self.add_url_rule('/logout/', view_func=logout_view, methods={'GET'})
@argument('email')
@option('-i', '--inventory',

19
ereuse_devicehub/resources/user/models.py
View File

@ -2,13 +2,15 @@ from uuid import uuid4
from citext import CIText
from flask import current_app as app
from sqlalchemy import Column, Boolean
from sqlalchemy import Column, Boolean, BigInteger, Sequence
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy_utils import EmailType, PasswordType
from teal.db import IntEnum
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.inventory.model import Inventory
from ereuse_devicehub.resources.models import STR_SIZE, Thing
from ereuse_devicehub.resources.enums import SessionType
class User(Thing):
@ -63,3 +65,18 @@ class UserInventory(db.Model):
__table_args__ = {'schema': 'common'}
user_id = db.Column(db.UUID(as_uuid=True), db.ForeignKey(User.id), primary_key=True)
inventory_id = db.Column(db.Unicode(), db.ForeignKey(Inventory.id), primary_key=True)
class Session(Thing):
__table_args__ = {'schema': 'common'}
id = Column(BigInteger, Sequence('device_seq'), primary_key=True)
expired = Column(BigInteger, default=0)
token = Column(UUID(as_uuid=True), default=uuid4, unique=True, nullable=False)
type = Column(IntEnum(SessionType), default=SessionType.Internal, nullable=False)
user_id = db.Column(db.UUID(as_uuid=True), db.ForeignKey(User.id))
user = db.relationship(User,
backref=db.backref('sessions', lazy=True, collection_class=set),
collection_class=set)
def __str__(self) -> str:
return '{0.token}'.format(self)

4
ereuse_devicehub/resources/user/schemas.py
View File

@ -9,6 +9,10 @@ from ereuse_devicehub.resources.inventory.schema import Inventory
from ereuse_devicehub.resources.schemas import Thing
class Session(Thing):
token = String(dump_only=True)
class User(Thing):
id = UUID(dump_only=True)
email = Email(required=True)

13
ereuse_devicehub/resources/user/views.py
View File

@ -1,11 +1,12 @@
from uuid import UUID
from uuid import UUID, uuid4
from flask import g, request
from flask.json import jsonify
from teal.resource import View
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.user.exceptions import WrongCredentials
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.user.schemas import User as UserS
class UserView(View):
@ -24,3 +25,11 @@ def login():
return schema_with_token.jsonify(user)
else:
raise WrongCredentials()
def logout():
# We use custom schema as we only want to parse a subset of user
g.user.token = uuid4()
db.session.add(g.user)
db.session.commit()
return jsonify('Ok')

6
tests/conftest.py
View File

@ -17,6 +17,8 @@ from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.agent.models import Person
from ereuse_devicehub.resources.tag import Tag
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.user.models import Session
from ereuse_devicehub.resources.enums import SessionType
STARTT = datetime(year=2000, month=1, day=1, hour=1)
"""A dummy starting time to use in tests."""
@ -111,7 +113,11 @@ def user2(app: Devicehub) -> UserClient:
def create_user(email='foo@foo.com', password='foo') -> User:
user = User(email=email, password=password)
user.individuals.add(Person(name='Timmy'))
session_external = Session(user=user, type=SessionType.External)
session_internal = Session(user=user, type=SessionType.Internal)
db.session.add(user)
db.session.add(session_internal)
db.session.add(session_external)
db.session.commit()
return user

2
tests/files/basic.csv
View File

File diff suppressed because one or more lines are too long

6
tests/files/proposal_extended_csv_report.csv
View File

File diff suppressed because one or more lines are too long

10
tests/test_action.py
View File

@ -387,7 +387,7 @@ def test_live_without_TestDataStorage(user: UserClient, client: Client, app: Dev
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
@ -420,7 +420,7 @@ def test_live_without_hdd_1(user: UserClient, client: Client, app: Devicehub):
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
@ -451,7 +451,7 @@ def test_live_without_hdd_2(user: UserClient, client: Client, app: Devicehub):
acer['components'] = components
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
@ -482,7 +482,7 @@ def test_live_without_hdd_3(user: UserClient, client: Client, app: Devicehub):
acer['components'] = components
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
@ -515,7 +515,7 @@ def test_live_with_hdd_with_old_time(user: UserClient, client: Client, app: Devi
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",

4
tests/test_basic.py
View File

@ -40,6 +40,7 @@ def test_api_docs(client: Client):
'/documents/erasures/',
'/documents/devices/',
'/documents/stamps/',
'/documents/wbconf/{wbtype}',
'/documents/internalstats/',
'/documents/stock/',
'/documents/check/',
@ -55,7 +56,8 @@ def test_api_docs(client: Client):
'/tags/',
'/tags/{tag_id}/device/{device_id}',
'/users/',
'/users/login/'
'/users/login/',
'/users/logout/',
# '/devices/{dev1_id}/merge/{dev2_id}',
# '/batteries/{dev1_id}/merge/{dev2_id}',
# '/bikes/{dev1_id}/merge/{dev2_id}',

133
tests/test_device.py
View File

@ -65,13 +65,13 @@ def test_device_model():
gcard = d.GraphicCard.query.one()
db.session.delete(pc)
db.session.flush()
assert pc.id == 1
assert pc.id == 3
assert d.Desktop.query.first() is None
db.session.commit()
assert d.Desktop.query.first() is None
assert network_adapter.id == 2
assert network_adapter.id == 4
assert d.NetworkAdapter.query.first() is not None, 'We removed the network adaptor'
assert gcard.id == 3, 'We should still hold a reference to a zombie graphic card'
assert 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'
@ -396,74 +396,73 @@ def test_sync_execute_register_mismatch_between_tags_and_hid():
@pytest.mark.mvp
def test_get_device(app: Devicehub, user: UserClient):
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_get_device(user: UserClient):
"""Checks GETting a d.Desktop with its components."""
with app.app_context():
pc = d.Desktop(model='p1mo',
manufacturer='p1ma',
serial_number='p1s',
chassis=ComputerChassis.Tower,
owner_id=user.user['id'])
pc.components = OrderedSet([
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)
# todo test is an abstract class. replace with another one
db.session.add(TestConnectivity(device=pc,
severity=Severity.Info,
agent=Person(name='Timmy'),
author=User(email='bar@bar.com')))
db.session.commit()
devicehub_id = pc.devicehub_id
pc, _ = user.get(res=d.Device, item=devicehub_id)
assert len(pc['actions']) == 1
assert pc['actions'][0]['type'] == 'TestConnectivity'
assert pc['actions'][0]['device'] == 1
assert pc['actions'][0]['severity'] == 'Info'
assert UUID(pc['actions'][0]['author'])
assert 'actions_components' not in pc, 'actions_components are internal use only'
assert 'actions_one' not in pc, 'they are internal use only'
assert 'author' not in pc
assert tuple(c['id'] for c in pc['components']) == (2, 3)
assert pc['hid'] == 'desktop-p1ma-p1mo-p1s'
assert pc['model'] == 'p1mo'
assert pc['manufacturer'] == 'p1ma'
assert pc['serialNumber'] == 'p1s'
assert pc['type'] == d.Desktop.t
pc = d.Desktop(model='p1mo',
manufacturer='p1ma',
serial_number='p1s',
chassis=ComputerChassis.Tower,
owner_id=user.user['id'])
pc.components = OrderedSet([
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)
# todo test is an abstract class. replace with another one
db.session.add(TestConnectivity(device=pc,
severity=Severity.Info,
agent=Person(name='Timmy'),
author=User(email='bar@bar.com')))
db.session.commit()
pc_api, _ = user.get(res=d.Device, item=pc.devicehub_id)
assert len(pc_api['actions']) == 1
assert pc_api['actions'][0]['type'] == 'TestConnectivity'
assert pc_api['actions'][0]['device'] == pc.id
assert pc_api['actions'][0]['severity'] == 'Info'
assert UUID(pc_api['actions'][0]['author'])
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 'author' not in pc_api
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['model'] == 'p1mo'
assert pc_api['manufacturer'] == 'p1ma'
assert pc_api['serialNumber'] == 'p1s'
assert pc_api['type'] == d.Desktop.t
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_get_devices(app: Devicehub, user: UserClient):
"""Checks GETting multiple devices."""
with app.app_context():
pc = d.Desktop(model='p1mo',
manufacturer='p1ma',
serial_number='p1s',
chassis=ComputerChassis.Tower,
owner_id=user.user['id'])
pc.components = OrderedSet([
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'])
])
pc1 = d.Desktop(model='p2mo',
manufacturer='p2ma',
serial_number='p2s',
chassis=ComputerChassis.Tower,
owner_id=user.user['id'])
pc2 = d.Laptop(model='p3mo',
manufacturer='p3ma',
serial_number='p3s',
chassis=ComputerChassis.Netbook,
owner_id=user.user['id'])
db.session.add_all((pc, pc1, pc2))
db.session.commit()
pc = d.Desktop(model='p1mo',
manufacturer='p1ma',
serial_number='p1s',
chassis=ComputerChassis.Tower,
owner_id=user.user['id'])
pc.components = OrderedSet([
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'])
])
pc1 = d.Desktop(model='p2mo',
manufacturer='p2ma',
serial_number='p2s',
chassis=ComputerChassis.Tower,
owner_id=user.user['id'])
pc2 = d.Laptop(model='p3mo',
manufacturer='p3ma',
serial_number='p3s',
chassis=ComputerChassis.Netbook,
owner_id=user.user['id'])
db.session.add_all((pc, pc1, pc2))
db.session.commit()
devices, _ = user.get(res=d.Device)
assert tuple(dev['id'] for dev in devices['items']) == (1, 2, 3, 4, 5)
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['type'] for dev in devices['items']) == (
d.Desktop.t, d.Desktop.t, d.Laptop.t, d.NetworkAdapter.t, d.GraphicCard.t
)
@ -536,7 +535,7 @@ def test_device_properties_format(app: Devicehub, user: UserClient):
user.post(file('asus-eee-1000h.snapshot.11'), res=m.Snapshot)
with app.app_context():
pc = d.Laptop.query.one() # type: d.Laptop
assert format(pc) == 'Laptop 1: model 1000h, S/N 94oaaq021116'
assert format(pc) == 'Laptop 3: model 1000h, S/N 94oaaq021116'
assert format(pc, 't') == 'Netbook 1000h'
assert format(pc, 's') == '(asustek computer inc.) S/N 94OAAQ021116'
assert pc.ram_size == 1024
@ -544,12 +543,12 @@ def test_device_properties_format(app: Devicehub, user: UserClient):
assert pc.graphic_card_model == 'mobile 945gse express integrated graphics controller'
assert pc.processor_model == 'intel atom cpu n270 @ 1.60ghz'
net = next(c for c in pc.components if isinstance(c, d.NetworkAdapter))
assert format(net) == 'NetworkAdapter 2: model ar8121/ar8113/ar8114 ' \
assert format(net) == 'NetworkAdapter 4: model ar8121/ar8113/ar8114 ' \
'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, 's') == 'qualcomm atheros 00:24:8C:7F:CF:2D – 100 Mbps'
hdd = next(c for c in pc.components if isinstance(c, d.DataStorage))
assert format(hdd) == 'HardDrive 7: model st9160310as, S/N 5sv4tqa6'
assert format(hdd) == 'HardDrive 9: model st9160310as, S/N 5sv4tqa6'
assert format(hdd, 't') == 'HardDrive st9160310as'
assert format(hdd, 's') == 'seagate 5SV4TQA6 – 152 GB'

10
tests/test_device_find.py
View File

@ -176,10 +176,10 @@ def test_device_query_filter_lots(user: UserClient):
@pytest.mark.mvp
def test_device_query(user: UserClient):
"""Checks result of inventory."""
snap, _ = user.post(conftest.file('basic.snapshot'), res=Snapshot)
snapshot, _ = user.post(conftest.file('basic.snapshot'), res=Snapshot)
i, _ = user.get(res=Device)
assert i['url'] == '/devices/'
assert i['items'][0]['url'] == '/devices/%s' % snap['device']['devicehubID']
assert i['items'][0]['url'] == '/devices/%s' % snapshot['device']['devicehubID']
pc = next(d for d in i['items'] if d['type'] == 'Desktop')
assert len(pc['actions']) == 4
assert len(pc['components']) == 3
@ -241,16 +241,16 @@ def test_device_search_regenerate_table(app: DeviceSearch, user: UserClient):
@pytest.mark.mvp
def test_device_query_search(user: UserClient):
# todo improve
user.post(file('basic.snapshot'), res=Snapshot)
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)
user.post(file('computer-monitor.snapshot'), res=Snapshot)
user.post(file('real-eee-1001pxd.snapshot.11'), res=Snapshot)
i, _ = user.get(res=Device, query=[('search', 'desktop')])
assert i['items'][0]['id'] == 1
assert i['items'][0]['id'] == snapshot['device']['id']
i, _ = user.get(res=Device, query=[('search', 'intel')])
assert len(i['items']) == 1
i, _ = user.get(res=Device, query=[('search', i['items'][0]['devicehubID'])])
assert len(i['items']) == 1
i, _ = user.get(res=Device, query=[('search', '1')])
i, _ = user.get(res=Device, query=[('search', snapshot['device']['id'])])
assert len(i['items']) == 1

55
tests/test_documents.py
View File

@ -10,14 +10,17 @@ from werkzeug.exceptions import Unauthorized
import teal.marshmallow
from ereuse_utils.test import ANY
from ereuse_devicehub import auth
from ereuse_devicehub.client import Client, UserClient
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.user.models import Session
from ereuse_devicehub.resources.action.models import Snapshot, Allocate, Live
from ereuse_devicehub.resources.documents import documents
from ereuse_devicehub.resources.device import models as d
from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.tag.model import Tag
from ereuse_devicehub.resources.hash_reports import ReportHash
from ereuse_devicehub.resources.enums import SessionType
from ereuse_devicehub.db import db
from tests import conftest
from tests.conftest import file
@ -262,6 +265,7 @@ def test_export_extended(app: Devicehub, user: UserClient):
pc.tags.add(tag)
db.session.add(pc)
db.session.commit()
csv_str, _ = user.get(res=documents.DocumentDef.t,
item='devices/',
accept='text/csv',
@ -464,7 +468,7 @@ def test_get_document_lots(user: UserClient, user2: UserClient):
assert export2_csv[1][3] == 'comments,lot3,testcomment-lot3,'
@pytest.mark.mvp
@pytest.mark.mvp
def test_verify_stamp(user: UserClient, client: Client):
"""Test verify stamp of one export device information in a csv file."""
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)
@ -472,12 +476,12 @@ def test_verify_stamp(user: UserClient, client: Client):
item='devices/',
accept='text/csv',
query=[('filter', {'type': ['Computer']})])
response, _ = client.post(res=documents.DocumentDef.t,
item='stamps/',
content_type='multipart/form-data',
accept='text/html',
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), 'example.csv')]},
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), 'example.csv')]},
status=200)
assert "alert alert-info" in response
assert not "alert alert-danger" in response
@ -501,10 +505,10 @@ def test_verify_stamp(user: UserClient, client: Client):
assert not "alert alert-danger" in response
@pytest.mark.mvp
@pytest.mark.mvp
def test_verify_stamp_log_info(user: UserClient, client: Client):
"""Test verify stamp of one export lots-info in a csv file."""
l, _ = user.post({'name': 'Lot1', 'description': 'comments,lot1,testcomment-lot1,'}, res=Lot)
l, _ = user.post({'name': 'Lot2', 'description': 'comments,lot2,testcomment-lot2,'}, res=Lot)
@ -516,8 +520,8 @@ def test_verify_stamp_log_info(user: UserClient, client: Client):
item='stamps/',
content_type='multipart/form-data',
accept='text/html',
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')),
'example.csv')]},
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')),
'example.csv')]},
status=200)
assert "alert alert-info" in response
@ -538,7 +542,7 @@ def test_verify_stamp_devices_stock(user: UserClient, client: Client):
content_type='multipart/form-data',
accept='text/html',
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')),
'example.csv')]},
'example.csv')]},
status=200)
assert "alert alert-info" in response
@ -573,8 +577,8 @@ def test_verify_stamp_csv_actions(user: UserClient, client: Client):
item='stamps/',
content_type='multipart/form-data',
accept='text/html',
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')),
'example.csv')]},
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')),
'example.csv')]},
status=200)
assert "alert alert-info" in response
@ -594,8 +598,8 @@ def test_verify_stamp_erasure_certificate(user: UserClient, client: Client):
item='stamps/',
content_type='multipart/form-data',
accept='text/html',
data={'docUpload': [(BytesIO(bytes(doc, 'utf-8')),
'example.csv')]},
data={'docUpload': [(BytesIO(bytes(doc, 'utf-8')),
'example.csv')]},
status=200)
assert "alert alert-danger" in response
@ -611,15 +615,15 @@ def test_verify_stamp_erasure_certificate(user: UserClient, client: Client):
item='stamps/',
content_type='multipart/form-data',
accept='text/html',
data={'docUpload': [(BytesIO(doc),
'example.csv')]},
data={'docUpload': [(BytesIO(doc),
'example.csv')]},
status=200)
assert "alert alert-info" in response
@pytest.mark.mvp
def test_get_document_internal_stats(user: UserClient, user2: UserClient):
"""Tests for get teh internal stats."""
"""Tests for get the internal stats."""
# csv_str, _ = user.get(res=documents.DocumentDef.t,
# item='internalstats/')
@ -644,3 +648,24 @@ def test_get_document_internal_stats(user: UserClient, user2: UserClient):
export_csv = list(obj_csv)
assert csv_str.strip() == '""'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_get_wbconf(user: UserClient):
"""Tests for get env file for usb wb."""
env, _ = user.get(res=documents.DocumentDef.t, item='wbconf/usodyrate', accept=ANY)
assert 'WB_ERASE = False' in env
env, _ = user.get(res=documents.DocumentDef.t, item='wbconf/usodywipe', accept=ANY)
assert 'WB_ERASE = False' in env
# assert 'WB_ERASE = True' in env
session = Session.query.filter_by(user_id=user.user['id'],
type=SessionType.Internal).first()
token = session.token
token = auth.Auth.encode(session.token)
assert token in env
user.user['token'] = token
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)

3
tests/test_snapshot.py
View File

@ -37,6 +37,7 @@ from tests import conftest
@pytest.mark.mvp
@pytest.mark.usefixtures('auth_app_context')
# cayop
def test_snapshot_model():
"""Tests creating a Snapshot with its relationships ensuring correct
DB mapping.
@ -318,7 +319,7 @@ def test_snapshot_tag_inner_tag(user: UserClient, tag_id: str, app: Devicehub):
action_types=(RateComputer.t, BenchmarkProcessor.t, VisualTest.t))
with app.app_context():
tag = Tag.query.one() # type: Tag
assert tag.device_id == 1, 'Tag should be linked to the first device'
assert tag.device_id == 3, 'Tag should be linked to the first device'
@pytest.mark.mvp

28
tests/test_workbench.py
View File

@ -32,20 +32,24 @@ def test_workbench_server_condensed(user: UserClient):
user.post({'id': t['id']}, res=Tag)
snapshot, _ = user.post(res=em.Snapshot, data=s)
pc_id = snapshot['device']['id']
cpu_id = snapshot['components'][3]['id']
ssd_id= snapshot['components'][4]['id']
hdd_id = snapshot['components'][5]['id']
actions = snapshot['actions']
assert {(action['type'], action['device']) for action in actions} == {
('BenchmarkProcessorSysbench', 5),
('StressTest', 1),
('EraseSectors', 6),
('EreusePrice', 1),
('BenchmarkRamSysbench', 1),
('BenchmarkProcessor', 5),
('Install', 6),
('EraseSectors', 7),
('BenchmarkDataStorage', 6),
('BenchmarkDataStorage', 7),
('TestDataStorage', 6),
('RateComputer', 1)
('BenchmarkProcessorSysbench', cpu_id),
('StressTest', pc_id),
('EraseSectors', ssd_id),
('EreusePrice', pc_id),
('BenchmarkRamSysbench', pc_id),
('BenchmarkProcessor', cpu_id),
('Install', ssd_id),
('EraseSectors', hdd_id),
('BenchmarkDataStorage', ssd_id),
('BenchmarkDataStorage', hdd_id),
('TestDataStorage', ssd_id),
('RateComputer', pc_id)
}
assert snapshot['closed']
assert snapshot['severity'] == 'Info'