Merge pull request #129 from eReuse/feature/#128-endpoint-config-wb

basic endpoint for wbconf
This commit is contained in:
cayop 2021-05-13 16:50:58 +02:00 committed by GitHub
commit 257532f10d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 334 additions and 118 deletions

View File

@ -11,6 +11,9 @@ ml).
## testing ## testing
[1.0.7-beta] [1.0.7-beta]
## [1.0.7-beta]
- [addend] #140 adding endpoint for download the settings for usb workbench
## [1.0.6-beta] ## [1.0.6-beta]
- [bugfix] #143 biginteger instead of integer in TestDataStorage - [bugfix] #143 biginteger instead of integer in TestDataStorage

View File

@ -3,12 +3,17 @@ from teal.auth import TokenAuth
from teal.db import ResourceNotFound from teal.db import ResourceNotFound
from werkzeug.exceptions import Unauthorized 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): class Auth(TokenAuth):
def authenticate(self, token: str, *args, **kw) -> User: def authenticate(self, token: str, *args, **kw) -> User:
try: 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): except (ResourceNotFound, DataError):
raise Unauthorized('Provide a suitable token.') raise Unauthorized('Provide a suitable token.')

View File

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

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.lot.models import Lot
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 ereuse_devicehub.resources.user.models import Session
from ereuse_devicehub.resources.enums import SessionType
class Dummy: class Dummy:
@ -194,6 +196,10 @@ class Dummy:
user.individuals.add(Person(name=name)) user.individuals.add(Person(name=name))
db.session.add(user) 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() db.session.commit()
client = UserClient(self.app, user.email, password, client = UserClient(self.app, user.email, password,

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 = '6a2a939d5668'
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')

View File

@ -1,7 +1,8 @@
import csv import csv
import datetime
import enum import enum
import uuid import uuid
import datetime
import pathlib
from collections import OrderedDict from collections import OrderedDict
from io import StringIO from io import StringIO
from typing import Callable, Iterable, Tuple from typing import Callable, Iterable, Tuple
@ -18,7 +19,9 @@ from flask.json import jsonify
from teal.cache import cache from teal.cache import cache
from teal.resource import Resource, View from teal.resource import Resource, View
from ereuse_devicehub import auth
from ereuse_devicehub.db import db 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.action import models as evs
from ereuse_devicehub.resources.device import models as devs from ereuse_devicehub.resources.device import models as devs
from ereuse_devicehub.resources.deliverynote.models import Deliverynote from ereuse_devicehub.resources.deliverynote.models import Deliverynote
@ -314,6 +317,30 @@ class InternalStatsView(DeviceView):
return output 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): class DocumentDef(Resource):
__type__ = 'Document' __type__ = 'Document'
SCHEMA = None SCHEMA = None
@ -380,3 +407,9 @@ class DocumentDef(Resource):
auth=app.auth) auth=app.auth)
actions_view = app.auth.requires_auth(actions_view) actions_view = app.auth.requires_auth(actions_view)
self.add_url_rule('/actions/', defaults=d, view_func=actions_view, methods=get) 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)

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

View File

@ -393,3 +393,24 @@ class TransferState(IntEnum):
def __str__(self): def __str__(self):
return self.name 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

View File

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

View File

@ -2,13 +2,15 @@ from uuid import uuid4
from citext import CIText from citext import CIText
from flask import current_app as app from flask import current_app as app
from sqlalchemy import Column from sqlalchemy import Column, BigInteger, Sequence
from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy_utils import EmailType, PasswordType from sqlalchemy_utils import EmailType, PasswordType
from teal.db import IntEnum
from ereuse_devicehub.db import db from ereuse_devicehub.db import db
from ereuse_devicehub.resources.inventory.model import Inventory from ereuse_devicehub.resources.inventory.model import Inventory
from ereuse_devicehub.resources.models import STR_SIZE, Thing from ereuse_devicehub.resources.models import STR_SIZE, Thing
from ereuse_devicehub.resources.enums import SessionType
class User(Thing): class User(Thing):
@ -57,3 +59,17 @@ class UserInventory(db.Model):
__table_args__ = {'schema': 'common'} __table_args__ = {'schema': 'common'}
user_id = db.Column(db.UUID(as_uuid=True), db.ForeignKey(User.id), primary_key=True) 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) inventory_id = db.Column(db.Unicode(), db.ForeignKey(Inventory.id), primary_key=True)
class Session(Thing):
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)

View File

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

View File

@ -1,11 +1,12 @@
from uuid import UUID from uuid import UUID, uuid4
from flask import g, request from flask import g, request
from flask.json import jsonify
from teal.resource import View from teal.resource import View
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.user.exceptions import WrongCredentials from ereuse_devicehub.resources.user.exceptions import WrongCredentials
from ereuse_devicehub.resources.user.models import User from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.user.schemas import User as UserS
class UserView(View): class UserView(View):
@ -24,3 +25,11 @@ def login():
return schema_with_token.jsonify(user) return schema_with_token.jsonify(user)
else: else:
raise WrongCredentials() 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')

View File

@ -17,6 +17,8 @@ from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.agent.models import Person from ereuse_devicehub.resources.agent.models import Person
from ereuse_devicehub.resources.tag import Tag from ereuse_devicehub.resources.tag import Tag
from ereuse_devicehub.resources.user.models import User 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) STARTT = datetime(year=2000, month=1, day=1, hour=1)
"""A dummy starting time to use in tests.""" """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: def create_user(email='foo@foo.com', password='foo') -> User:
user = User(email=email, password=password) user = User(email=email, password=password)
user.individuals.add(Person(name='Timmy')) 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(user)
db.session.add(session_internal)
db.session.add(session_external)
db.session.commit() db.session.commit()
return user return user

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View File

@ -382,7 +382,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=1).one() db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1, post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa", "devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi", "finalUserCode": "abcdefjhi",
@ -415,7 +415,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=1).one() db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1, post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa", "devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi", "finalUserCode": "abcdefjhi",
@ -446,7 +446,7 @@ def test_live_without_hdd_2(user: UserClient, client: Client, app: Devicehub):
acer['components'] = components acer['components'] = components
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=1).one() db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1, post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa", "devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi", "finalUserCode": "abcdefjhi",
@ -477,7 +477,7 @@ def test_live_without_hdd_3(user: UserClient, client: Client, app: Devicehub):
acer['components'] = components acer['components'] = components
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=1).one() db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1, post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa", "devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi", "finalUserCode": "abcdefjhi",
@ -510,7 +510,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=1).one() db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1, post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa", "devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi", "finalUserCode": "abcdefjhi",

View File

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

View File

@ -65,13 +65,13 @@ def test_device_model():
gcard = d.GraphicCard.query.one() gcard = d.GraphicCard.query.one()
db.session.delete(pc) db.session.delete(pc)
db.session.flush() db.session.flush()
assert pc.id == 1 assert pc.id == 3
assert d.Desktop.query.first() is None assert d.Desktop.query.first() is None
db.session.commit() db.session.commit()
assert d.Desktop.query.first() is None 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 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' assert d.GraphicCard.query.first() is None, 'We should have deleted it it was inside the pc'
@ -396,9 +396,9 @@ def test_sync_execute_register_mismatch_between_tags_and_hid():
@pytest.mark.mvp @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.""" """Checks GETting a d.Desktop with its components."""
with app.app_context():
pc = d.Desktop(model='p1mo', pc = d.Desktop(model='p1mo',
manufacturer='p1ma', manufacturer='p1ma',
serial_number='p1s', serial_number='p1s',
@ -416,29 +416,27 @@ def test_get_device(app: Devicehub, user: UserClient):
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()
devicehub_id = pc.devicehub_id pc_api, _ = user.get(res=d.Device, item=pc.devicehub_id)
assert len(pc_api['actions']) == 1
pc, _ = user.get(res=d.Device, item=devicehub_id) assert pc_api['actions'][0]['type'] == 'TestConnectivity'
assert len(pc['actions']) == 1 assert pc_api['actions'][0]['device'] == pc.id
assert pc['actions'][0]['type'] == 'TestConnectivity' assert pc_api['actions'][0]['severity'] == 'Info'
assert pc['actions'][0]['device'] == 1 assert UUID(pc_api['actions'][0]['author'])
assert pc['actions'][0]['severity'] == 'Info' assert 'actions_components' not in pc_api, 'actions_components are internal use only'
assert UUID(pc['actions'][0]['author']) assert 'actions_one' not in pc_api, 'they are internal use only'
assert 'actions_components' not in pc, 'actions_components are internal use only' assert 'author' not in pc_api
assert 'actions_one' not in pc, 'they are internal use only' assert tuple(c['id'] for c in pc_api['components']) == tuple(c.id for c in pc.components)
assert 'author' not in pc assert pc_api['hid'] == 'desktop-p1ma-p1mo-p1s'
assert tuple(c['id'] for c in pc['components']) == (2, 3) assert pc_api['model'] == 'p1mo'
assert pc['hid'] == 'desktop-p1ma-p1mo-p1s' assert pc_api['manufacturer'] == 'p1ma'
assert pc['model'] == 'p1mo' assert pc_api['serialNumber'] == 'p1s'
assert pc['manufacturer'] == 'p1ma' assert pc_api['type'] == d.Desktop.t
assert pc['serialNumber'] == 'p1s'
assert pc['type'] == d.Desktop.t
@pytest.mark.mvp @pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_get_devices(app: Devicehub, user: UserClient): def test_get_devices(app: Devicehub, user: UserClient):
"""Checks GETting multiple devices.""" """Checks GETting multiple devices."""
with app.app_context():
pc = d.Desktop(model='p1mo', pc = d.Desktop(model='p1mo',
manufacturer='p1ma', manufacturer='p1ma',
serial_number='p1s', serial_number='p1s',
@ -463,7 +461,8 @@ def test_get_devices(app: Devicehub, user: UserClient):
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)
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']) == ( 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
) )
@ -536,7 +535,7 @@ def test_device_properties_format(app: Devicehub, user: UserClient):
user.post(file('asus-eee-1000h.snapshot.11'), res=m.Snapshot) user.post(file('asus-eee-1000h.snapshot.11'), res=m.Snapshot)
with app.app_context(): with app.app_context():
pc = d.Laptop.query.one() # type: d.Laptop 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, 't') == 'Netbook 1000h'
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
@ -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.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 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' '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 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, 't') == 'HardDrive st9160310as'
assert format(hdd, 's') == 'seagate 5SV4TQA6 152 GB' assert format(hdd, 's') == 'seagate 5SV4TQA6 152 GB'

View File

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

View File

@ -10,14 +10,17 @@ from werkzeug.exceptions import Unauthorized
import teal.marshmallow import teal.marshmallow
from ereuse_utils.test import ANY from ereuse_utils.test import ANY
from ereuse_devicehub import auth
from ereuse_devicehub.client import Client, UserClient from ereuse_devicehub.client import Client, UserClient
from ereuse_devicehub.devicehub import Devicehub 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.action.models import Snapshot, Allocate, Live
from ereuse_devicehub.resources.documents import documents from ereuse_devicehub.resources.documents import documents
from ereuse_devicehub.resources.device import models as d from ereuse_devicehub.resources.device import models as d
from ereuse_devicehub.resources.lot.models import Lot from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.tag.model import Tag from ereuse_devicehub.resources.tag.model import Tag
from ereuse_devicehub.resources.hash_reports import ReportHash from ereuse_devicehub.resources.hash_reports import ReportHash
from ereuse_devicehub.resources.enums import SessionType
from ereuse_devicehub.db import db from ereuse_devicehub.db import db
from tests import conftest from tests import conftest
from tests.conftest import file from tests.conftest import file
@ -262,6 +265,7 @@ def test_export_extended(app: Devicehub, user: UserClient):
pc.tags.add(tag) pc.tags.add(tag)
db.session.add(pc) db.session.add(pc)
db.session.commit() db.session.commit()
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(res=documents.DocumentDef.t,
item='devices/', item='devices/',
accept='text/csv', accept='text/csv',
@ -619,7 +623,7 @@ def test_verify_stamp_erasure_certificate(user: UserClient, client: Client):
@pytest.mark.mvp @pytest.mark.mvp
def test_get_document_internal_stats(user: UserClient, user2: UserClient): 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, # csv_str, _ = user.get(res=documents.DocumentDef.t,
# item='internalstats/') # item='internalstats/')
@ -644,3 +648,24 @@ def test_get_document_internal_stats(user: UserClient, user2: UserClient):
export_csv = list(obj_csv) export_csv = list(obj_csv)
assert csv_str.strip() == '""' 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)

View File

@ -37,6 +37,7 @@ from tests import conftest
@pytest.mark.mvp @pytest.mark.mvp
@pytest.mark.usefixtures('auth_app_context') @pytest.mark.usefixtures('auth_app_context')
# cayop
def test_snapshot_model(): def test_snapshot_model():
"""Tests creating a Snapshot with its relationships ensuring correct """Tests creating a Snapshot with its relationships ensuring correct
DB mapping. 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)) action_types=(RateComputer.t, BenchmarkProcessor.t, VisualTest.t))
with app.app_context(): with app.app_context():
tag = Tag.query.one() # type: Tag 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 @pytest.mark.mvp

View File

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