Merge pull request #131 from eReuse/feature/#130-code-for-device

code for device
This commit is contained in:
cayop 2021-04-16 15:05:09 +02:00 committed by GitHub
commit 02fdcb1d42
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
23 changed files with 166 additions and 98 deletions

View File

@ -14,6 +14,7 @@ ml).
## [1.0.5-beta] ## [1.0.5-beta]
- [addend] #124 adding endpoint for extract the internal stats of use - [addend] #124 adding endpoint for extract the internal stats of use
- [addend] #122 system for verify all documents that it's produced from devicehub - [addend] #122 system for verify all documents that it's produced from devicehub
- [addend] #131 add one code for every device
## [1.0.4-beta] ## [1.0.4-beta]
- [addend] #95 adding endpoint for check the hash of one report - [addend] #95 adding endpoint for check the hash of one report

View File

@ -95,6 +95,7 @@ class Dummy:
s, _ = user1.post(res=m.Snapshot, data=snapshot) s, _ = user1.post(res=m.Snapshot, data=snapshot)
if s.get('uuid', None) == 'ec23c11b-80b6-42cd-ac5c-73ba7acddbc4': if s.get('uuid', None) == 'ec23c11b-80b6-42cd-ac5c-73ba7acddbc4':
sample_pc = s['device']['id'] sample_pc = s['device']['id']
sample_pc_devicehub_id = s['device']['devicehubID']
else: else:
pcs.add(s['device']['id']) pcs.add(s['device']['id'])
if s.get('uuid', None) == 'de4f495e-c58b-40e1-a33e-46ab5e84767e': # oreo if s.get('uuid', None) == 'de4f495e-c58b-40e1-a33e-46ab5e84767e': # oreo
@ -180,9 +181,9 @@ class Dummy:
# res=m.Action) # res=m.Action)
# todo Receive # todo Receive
user1.get(res=Device, item=sample_pc) # Test user1.get(res=Device, item=sample_pc_devicehub_id) # Test
anonymous = self.app.test_client() anonymous = self.app.test_client()
html, _ = anonymous.get(res=Device, item=sample_pc, accept=ANY) html, _ = anonymous.get(res=Device, item=sample_pc_devicehub_id, accept=ANY)
assert 'intel core2 duo cpu' in html assert 'intel core2 duo cpu' in html
# For netbook: to preapre -> torepair -> to dispose -> disposed # For netbook: to preapre -> torepair -> to dispose -> disposed

View File

@ -0,0 +1,49 @@
"""add code to device
Revision ID: 8cb91ad1cc40
Revises: 6a2a939d5668
Create Date: 2021-03-03 10:39:19.331027
"""
import citext
import sqlalchemy as sa
from alembic import op
from alembic import context
from ereuse_devicehub.resources.utils import hashcode
# revision identifiers, used by Alembic.
revision = '8cb91ad1cc40'
down_revision = '6a2a939d5668'
branch_labels = None
depends_on = None
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_data():
con = op.get_bind()
devices = con.execute(f"select id from {get_inv()}.device")
for d in devices:
id_dev = d.id
code = hashcode.encode(d.id)
sql = f"update {get_inv()}.device set devicehub_id='{code}' where id={id_dev};"
con.execute(sql)
def upgrade():
op.add_column('device', sa.Column('devicehub_id', citext.CIText(),
unique=True,
nullable=True), schema=f'{get_inv()}')
upgrade_data()
def downgrade():
op.drop_column('device', 'devicehub_id', schema=f'{get_inv()}')

View File

@ -10,7 +10,7 @@ from ereuse_devicehub.resources.device.views import DeviceView, DeviceMergeView,
class DeviceDef(Resource): class DeviceDef(Resource):
SCHEMA = schemas.Device SCHEMA = schemas.Device
VIEW = DeviceView VIEW = DeviceView
ID_CONVERTER = Converters.int ID_CONVERTER = Converters.string
AUTH = False # We manage this at each view AUTH = False # We manage this at each view
def __init__(self, app, def __init__(self, app,

View File

@ -8,6 +8,7 @@ from typing import Dict, List, Set
from boltons import urlutils from boltons import urlutils
from citext import CIText from citext import CIText
from flask_sqlalchemy import event
from ereuse_utils.naming import HID_CONVERSION_DOC, Naming from ereuse_utils.naming import HID_CONVERSION_DOC, Naming
from flask import g from flask import g
from more_itertools import unique_everseen from more_itertools import unique_everseen
@ -27,12 +28,19 @@ from teal.marshmallow import ValidationError
from teal.resource import url_for_resource from teal.resource import url_for_resource
from ereuse_devicehub.db import db from ereuse_devicehub.db import db
from ereuse_devicehub.resources.utils import hashcode
from ereuse_devicehub.resources.enums import BatteryTechnology, CameraFacing, ComputerChassis, \ from ereuse_devicehub.resources.enums import BatteryTechnology, CameraFacing, ComputerChassis, \
DataStorageInterface, DisplayTech, PrinterTechnology, RamFormat, RamInterface, Severity, TransferState DataStorageInterface, DisplayTech, PrinterTechnology, RamFormat, RamInterface, Severity, TransferState
from ereuse_devicehub.resources.models import STR_SM_SIZE, Thing, listener_reset_field_updated_in_actual_time from ereuse_devicehub.resources.models import STR_SM_SIZE, Thing, listener_reset_field_updated_in_actual_time
from ereuse_devicehub.resources.user.models import User from ereuse_devicehub.resources.user.models import User
def create_code(context):
_id = Device.query.order_by(Device.id.desc()).first() or 1
if not _id == 1:
_id = _id.id + 1
return hashcode.encode(_id)
class Device(Thing): class Device(Thing):
"""Base class for any type of physical object that can be identified. """Base class for any type of physical object that can be identified.
@ -115,6 +123,8 @@ class Device(Thing):
owner = db.relationship(User, primaryjoin=owner_id == User.id) owner = db.relationship(User, primaryjoin=owner_id == User.id)
allocated = db.Column(Boolean, default=False) allocated = db.Column(Boolean, default=False)
allocated.comment = "device is allocated or not." allocated.comment = "device is allocated or not."
devicehub_id = db.Column(db.CIText(), nullable=True, unique=True, default=create_code)
devicehub_id.comment = "device have a unique code."
_NON_PHYSICAL_PROPS = { _NON_PHYSICAL_PROPS = {
'id', 'id',
@ -137,7 +147,8 @@ class Device(Thing):
'version', 'version',
'sku', 'sku',
'image', 'image',
'allocated' 'allocated',
'devicehub_id'
} }
__table_args__ = ( __table_args__ = (
@ -225,7 +236,7 @@ class Device(Thing):
@property @property
def url(self) -> urlutils.URL: def url(self) -> urlutils.URL:
"""The URL where to GET this device.""" """The URL where to GET this device."""
return urlutils.URL(url_for_resource(Device, item_id=self.id)) return urlutils.URL(url_for_resource(Device, item_id=self.devicehub_id))
@property @property
def rate(self): def rate(self):
@ -363,7 +374,7 @@ class Device(Thing):
if act.type == 'Allocate': if act.type == 'Allocate':
allo = {'type': 'Allocate', allo = {'type': 'Allocate',
'systemId': self.id, 'devicehubID': self.devicehub_id,
'finalUserCode': act.final_user_code, 'finalUserCode': act.final_user_code,
'numEndUsers': act.end_users, 'numEndUsers': act.end_users,
'hid': self.hid, 'hid': self.hid,
@ -384,7 +395,7 @@ class Device(Thing):
if act.type == 'Deallocate': if act.type == 'Deallocate':
deallo = {'type': 'Deallocate', deallo = {'type': 'Deallocate',
'systemId': self.id, 'devicehubID': self.devicehub_id,
'finalUserCode': '', 'finalUserCode': '',
'numEndUsers': '', 'numEndUsers': '',
'hid': self.hid, 'hid': self.hid,
@ -1023,3 +1034,4 @@ class Manufacturer(db.Model):
listener_reset_field_updated_in_actual_time(Device) listener_reset_field_updated_in_actual_time(Device)

View File

@ -66,6 +66,8 @@ class Device(Thing):
sku = SanitizedStr(description=m.Device.sku.comment) sku = SanitizedStr(description=m.Device.sku.comment)
image = URL(description=m.Device.image.comment) image = URL(description=m.Device.image.comment)
allocated = Boolean(description=m.Device.allocated.comment) allocated = Boolean(description=m.Device.allocated.comment)
devicehub_id = SanitizedStr(data_key='devicehubID',
description=m.Device.devicehub_id.comment)
@pre_load @pre_load
def from_actions_to_actions_one(self, data: dict): def from_actions_to_actions_one(self, data: dict):

View File

@ -118,7 +118,7 @@ class DeviceView(View):
return Response(status=204) return Response(status=204)
raise ValueError('Cannot patch a non computer') raise ValueError('Cannot patch a non computer')
def one(self, id: int): def one(self, id: str):
"""Gets one device.""" """Gets one device."""
if not request.authorization: if not request.authorization:
return self.one_public(id) return self.one_public(id)
@ -126,12 +126,12 @@ class DeviceView(View):
return self.one_private(id) return self.one_private(id)
def one_public(self, id: int): def one_public(self, id: int):
device = Device.query.filter_by(id=id).one() device = Device.query.filter_by(devicehub_id=id).one()
return render_template('devices/layout.html', device=device, states=states) return render_template('devices/layout.html', device=device, states=states)
@auth.Auth.requires_auth @auth.Auth.requires_auth
def one_private(self, id: int): def one_private(self, id: str):
device = Device.query.filter_by(id=id, owner_id=g.user.id).first() device = Device.query.filter_by(devicehub_id=id, owner_id=g.user.id).first()
if not device: if not device:
return self.one_public(id) return self.one_public(id)
return self.schema.jsonify(device) return self.schema.jsonify(device)

View File

@ -42,10 +42,10 @@ class DeviceRow(OrderedDict):
software = "{software} {version}".format( software = "{software} {version}".format(
software=snapshot.software.name, version=snapshot.version) software=snapshot.software.name, version=snapshot.version)
# General information about device # General information about device
self['System ID'] = device.id self['DevicehubID'] = device.devicehub_id
self['DocumentID'] = self.document_id self['DocumentID'] = self.document_id
self['Public Link'] = '{url}{id}'.format(url=url_for('Device.main', _external=True), self['Public Link'] = '{url}{id}'.format(url=url_for('Device.main', _external=True),
id=device.id) id=device.devicehub_id)
self['Tag 1 Type'] = self['Tag 1 ID'] = self['Tag 1 Organization'] = '' self['Tag 1 Type'] = self['Tag 1 ID'] = self['Tag 1 Organization'] = ''
self['Tag 2 Type'] = self['Tag 2 ID'] = self['Tag 2 Organization'] = '' self['Tag 2 Type'] = self['Tag 2 ID'] = self['Tag 2 Organization'] = ''
self['Tag 3 Type'] = self['Tag 3 ID'] = self['Tag 3 Organization'] = '' self['Tag 3 Type'] = self['Tag 3 ID'] = self['Tag 3 Organization'] = ''
@ -369,7 +369,7 @@ class ActionRow(OrderedDict):
def __init__(self, allocate): def __init__(self, allocate):
super().__init__() super().__init__()
# General information about allocates, deallocate and lives # General information about allocates, deallocate and lives
self['SystemId'] = allocate['systemId'] self['DevicehubID'] = allocate['devicehubID']
self['Hid'] = allocate['hid'] self['Hid'] = allocate['hid']
self['Start'] = allocate['start'] self['Start'] = allocate['start']
self['FinalUserCode'] = allocate['finalUserCode'] self['FinalUserCode'] = allocate['finalUserCode']

View File

@ -102,13 +102,10 @@ class DocumentView(DeviceView):
url_pdf = boltons.urlutils.URL(flask.request.url) url_pdf = boltons.urlutils.URL(flask.request.url)
url_pdf.query_params['format'] = 'PDF' url_pdf.query_params['format'] = 'PDF'
url_web = boltons.urlutils.URL(flask.request.url)
url_web.query_params['format'] = 'HTML'
params = { params = {
'title': 'Erasure Certificate', 'title': 'Erasure Certificate',
'erasures': tuple(erasures()), 'erasures': tuple(erasures()),
'url_pdf': url_pdf.to_text(), 'url_pdf': url_pdf.to_text()
'url_web': url_web.to_text()
} }
return flask.render_template('documents/erasure.html', **params) return flask.render_template('documents/erasure.html', **params)

View File

@ -42,13 +42,13 @@
<dt>Computer where was erase:</dt> <dt>Computer where was erase:</dt>
<dd>Title: {{ erasure.parent.__format__('ts') }}</dd> <dd>Title: {{ erasure.parent.__format__('ts') }}</dd>
<dd>SystemId: {{ erasure.parent.id }}</dd> <dd>DevicehubID: {{ erasure.parent.devicehub_id }}</dd>
<dd>Hid: {{ erasure.parent.hid }}</dd> <dd>Hid: {{ erasure.parent.hid }}</dd>
<dd>Tags: {{ erasure.parent.tags }}</dd> <dd>Tags: {{ erasure.parent.tags }}</dd>
<dt>Computer where it resides:</dt> <dt>Computer where it resides:</dt>
<dd>Title: {{ erasure.device.parent.__format__('ts') }}</dd> <dd>Title: {{ erasure.device.parent.__format__('ts') }}</dd>
<dd>SystemId: {{ erasure.device.parent.id }}</dd> <dd>DevicehubID: {{ erasure.device.parent.devicehub_id }}</dd>
<dd>Hid: {{ erasure.device.parent.hid }}</dd> <dd>Hid: {{ erasure.device.parent.hid }}</dd>
<dd>Tags: {{ erasure.device.parent.tags }}</dd> <dd>Tags: {{ erasure.device.parent.tags }}</dd>
@ -87,6 +87,6 @@
<a href="{{ url_pdf }}">Click here to download the PDF.</a> <a href="{{ url_pdf }}">Click here to download the PDF.</a>
</div> </div>
<div class="print-only"> <div class="print-only">
<a href="{{ url_web }}">Verify on-line the integrity of this document</a> <a href="{{ url_for('Document.StampsView', _external=True) }}">Verify on-line the integrity of this document</a>
</div> </div>
{% endblock %} {% endblock %}

View File

@ -35,7 +35,6 @@ class TagDef(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)
_get_device_from_tag = app.auth.requires_auth(get_device_from_tag)
# DeviceTagView URLs # DeviceTagView URLs
device_view = TagDeviceView.as_view('tag-device-view', definition=self, auth=app.auth) device_view = TagDeviceView.as_view('tag-device-view', definition=self, auth=app.auth)

View File

@ -18,7 +18,7 @@ class TagView(View):
tag = Tag.query.filter_by(internal_id=internal_id).one() # type: Tag tag = Tag.query.filter_by(internal_id=internal_id).one() # type: Tag
if not tag.device: if not tag.device:
raise TagNotLinked(tag.id) raise TagNotLinked(tag.id)
return redirect(location=url_for_resource(Device, tag.device.id)) return redirect(location=url_for_resource(Device, tag.device.devicehub_id))
@auth.Auth.requires_auth @auth.Auth.requires_auth
def post(self): def post(self):
@ -82,7 +82,9 @@ class TagDeviceView(View):
tag = Tag.from_an_id(id).one() # type: Tag tag = Tag.from_an_id(id).one() # type: Tag
if not tag.device: if not tag.device:
raise TagNotLinked(tag.id) raise TagNotLinked(tag.id)
return redirect(location=url_for_resource(Device, tag.device.id)) if not request.authorization:
return redirect(location=url_for_resource(Device, tag.device.devicehub_id))
return app.resources[Device.t].schema.jsonify(tag.device.devicehub_id)
@auth.Auth.requires_auth @auth.Auth.requires_auth
def one_authorization(self, id): def one_authorization(self, id):
@ -91,11 +93,10 @@ class TagDeviceView(View):
raise TagNotLinked(tag.id) raise TagNotLinked(tag.id)
return app.resources[Device.t].schema.jsonify(tag.device) return app.resources[Device.t].schema.jsonify(tag.device)
# noinspection PyMethodOverriding
@auth.Auth.requires_auth @auth.Auth.requires_auth
def put(self, tag_id: str, device_id: str): def put(self, tag_id: str, device_id: int):
"""Links an existing tag with a device.""" """Links an existing tag with a device."""
# tag = Tag.from_an_id(tag_id).one() # type: Tag device_id = int(device_id)
tag = Tag.from_an_id(tag_id).filter_by(owner=g.user).one() # type: Tag tag = Tag.from_an_id(tag_id).filter_by(owner=g.user).one() # type: Tag
if tag.device_id: if tag.device_id:
if tag.device_id == device_id: if tag.device_id == device_id:
@ -137,7 +138,7 @@ def get_device_from_tag(id: str):
# todo this could be more efficient by Device.query... join with tag # todo this could be more efficient by Device.query... join with tag
device = Tag.query.filter_by(id=id).one().device device = Tag.query.filter_by(id=id).one().device
if not request.authorization: if not request.authorization:
return redirect(location=url_for_resource(Device, device.id)) return redirect(location=url_for_resource(Device, device.devicehub_id))
if device is None: if device is None:
raise TagNotLinked(id) raise TagNotLinked(id)
return app.resources[Device.t].schema.jsonify(device) return app.resources[Device.t].schema.jsonify(device)

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View File

@ -242,7 +242,7 @@ def test_generic_action(action_model_state: Tuple[models.Action, states.Trading]
action = {'type': action_model.t, 'devices': [snapshot['device']['id']]} action = {'type': action_model.t, 'devices': [snapshot['device']['id']]}
action, _ = user.post(action, res=models.Action) action, _ = user.post(action, res=models.Action)
assert action['devices'][0]['id'] == snapshot['device']['id'] assert action['devices'][0]['id'] == snapshot['device']['id']
device, _ = user.get(res=Device, item=snapshot['device']['id']) device, _ = user.get(res=Device, item=snapshot['device']['devicehubID'])
assert device['actions'][-1]['id'] == action['id'] assert device['actions'][-1]['id'] == action['id']
assert device['physical'] == state.name assert device['physical'] == state.name
# Check if the update of device is changed # Check if the update of device is changed
@ -625,6 +625,7 @@ def test_allocate(user: UserClient):
""" Tests allocate """ """ Tests allocate """
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot) snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
device_id = snapshot['device']['id'] device_id = snapshot['device']['id']
devicehub_id = snapshot['device']['devicehubID']
post_request = {"transaction": "ccc", post_request = {"transaction": "ccc",
"finalUserCode": "aabbcc", "finalUserCode": "aabbcc",
"name": "John", "name": "John",
@ -638,7 +639,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=device_id) device, _ = user.get(res=Device, item=devicehub_id)
assert device['allocated'] == True assert device['allocated'] == 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']
@ -691,6 +692,7 @@ def test_deallocate(user: UserClient):
""" Tests deallocate """ """ Tests deallocate """
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot) snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
device_id = snapshot['device']['id'] device_id = snapshot['device']['id']
devicehub_id = snapshot['device']['devicehubID']
post_deallocate = {"startTime": "2020-11-01T02:00:00+00:00", post_deallocate = {"startTime": "2020-11-01T02:00:00+00:00",
"transaction": "ccc", "transaction": "ccc",
"devices": [device_id] "devices": [device_id]
@ -705,7 +707,7 @@ 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=device_id) device, _ = user.get(res=Device, item=devicehub_id)
assert device['allocated'] == True assert device['allocated'] == 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']
@ -750,7 +752,6 @@ def test_deallocate_bad_dates(user: UserClient):
def test_trade(action_model_state: Tuple[Type[models.Action], states.Trading], user: UserClient): def test_trade(action_model_state: Tuple[Type[models.Action], states.Trading], user: UserClient):
"""Tests POSTing all Trade actions.""" """Tests POSTing all Trade actions."""
# todo missing None states.Trading for after cancelling renting, for example # todo missing None states.Trading for after cancelling renting, for example
# import pdb; pdb.set_trace()
# Remove this test # Remove this test
action_model, state = action_model_state action_model, state = action_model_state
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot) snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
@ -764,7 +765,7 @@ def test_trade(action_model_state: Tuple[Type[models.Action], states.Trading], u
action['invoiceNumber'] = 'ABC' action['invoiceNumber'] = 'ABC'
action, _ = user.post(action, res=models.Action) action, _ = user.post(action, res=models.Action)
assert action['devices'][0]['id'] == snapshot['device']['id'] assert action['devices'][0]['id'] == snapshot['device']['id']
device, _ = user.get(res=Device, item=snapshot['device']['id']) device, _ = user.get(res=Device, item=snapshot['device']['devicehubID'])
assert device['actions'][-1]['id'] == action['id'] assert device['actions'][-1]['id'] == action['id']
assert device['trading'] == state.name assert device['trading'] == state.name
@ -787,7 +788,7 @@ def test_price_custom():
assert p['price'] == 25.25 assert p['price'] == 25.25
assert p['currency'] == Currency.EUR.name == 'EUR' assert p['currency'] == Currency.EUR.name == 'EUR'
c, _ = client.get(res=Device, item=computer.id) c, _ = client.get(res=Device, item=computer.devicehub_id)
assert c['price']['id'] == p['id'] assert c['price']['id'] == p['id']
@ -805,7 +806,7 @@ def test_price_custom_client(user: UserClient):
assert 25 == price['price'] assert 25 == price['price']
assert Currency.EUR.name == price['currency'] assert Currency.EUR.name == price['currency']
device, _ = user.get(res=Device, item=price['device']['id']) device, _ = user.get(res=Device, item=price['device']['devicehubID'])
assert 25 == device['price']['price'] assert 25 == device['price']['price']

View File

@ -416,7 +416,9 @@ 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()
pc, _ = user.get(res=d.Device, item=1) devicehub_id = pc.devicehub_id
pc, _ = user.get(res=d.Device, item=devicehub_id)
assert len(pc['actions']) == 1 assert len(pc['actions']) == 1
assert pc['actions'][0]['type'] == 'TestConnectivity' assert pc['actions'][0]['type'] == 'TestConnectivity'
assert pc['actions'][0]['device'] == 1 assert pc['actions'][0]['device'] == 1
@ -473,14 +475,14 @@ def test_get_device_permissions(app: Devicehub, user: UserClient, user2: UserCli
"""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)
pc, res = user.get(res=d.Device, item=s['device']['id']) pc, res = user.get(res=d.Device, item=s['device']['devicehubID'])
assert res.status_code == 200 assert res.status_code == 200
assert len(pc['actions']) == 9 assert len(pc['actions']) == 9
html, _ = client.get(res=d.Device, item=s['device']['id'], accept=ANY) html, _ = client.get(res=d.Device, item=s['device']['devicehubID'], accept=ANY)
assert 'intel atom cpu n270 @ 1.60ghz' in html assert 'intel atom cpu n270 @ 1.60ghz' in html
assert '00:24:8C:7F:CF:2D 100 Mbps' in html assert '00:24:8C:7F:CF:2D 100 Mbps' in html
pc2, res2 = user2.get(res=d.Device, item=s['device']['id'], accept=ANY) pc2, res2 = user2.get(res=d.Device, item=s['device']['devicehubID'], accept=ANY)
assert res2.status_code == 200 assert res2.status_code == 200
assert pc2 == html assert pc2 == html
@ -555,7 +557,7 @@ def test_device_properties_format(app: Devicehub, user: UserClient):
@pytest.mark.mvp @pytest.mark.mvp
def test_device_public(user: UserClient, client: Client): def test_device_public(user: UserClient, client: Client):
s, _ = user.post(file('asus-eee-1000h.snapshot.11'), res=m.Snapshot) s, _ = user.post(file('asus-eee-1000h.snapshot.11'), res=m.Snapshot)
html, _ = client.get(res=d.Device, item=s['device']['id'], accept=ANY) html, _ = client.get(res=d.Device, item=s['device']['devicehubID'], accept=ANY)
assert 'intel atom cpu n270 @ 1.60ghz' in html assert 'intel atom cpu n270 @ 1.60ghz' in html
assert '00:24:8C:7F:CF:2D 100 Mbps' in html assert '00:24:8C:7F:CF:2D 100 Mbps' in html
@ -614,8 +616,8 @@ def test_cooking_mixer_api(user: UserClient):
def test_hid_with_mac(app: Devicehub, user: UserClient): def test_hid_with_mac(app: Devicehub, user: UserClient):
"""Checks hid with mac.""" """Checks hid with mac."""
snapshot = file('asus-eee-1000h.snapshot.11') snapshot = file('asus-eee-1000h.snapshot.11')
user.post(snapshot, res=m.Snapshot) snap, _ = user.post(snapshot, res=m.Snapshot)
pc, _ = user.get(res=d.Device, item=1) pc, _ = user.get(res=d.Device, item=snap['device']['devicehubID'])
assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d' assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d'
@ -624,8 +626,8 @@ def test_hid_without_mac(app: Devicehub, user: UserClient):
"""Checks hid without mac.""" """Checks hid without mac."""
snapshot = file('asus-eee-1000h.snapshot.11') snapshot = file('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']
user.post(snapshot, res=m.Snapshot) snap, _ = user.post(snapshot, res=m.Snapshot)
pc, _ = user.get(res=d.Device, item=1) 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'
@ -635,8 +637,8 @@ def test_hid_with_mac_none(app: Devicehub, user: UserClient):
snapshot = file('asus-eee-1000h.snapshot.11') snapshot = file('asus-eee-1000h.snapshot.11')
network = [c for c in snapshot['components'] if c['type'] == 'NetworkAdapter'][0] network = [c for c in snapshot['components'] if c['type'] == 'NetworkAdapter'][0]
network['serialNumber'] = None network['serialNumber'] = None
user.post(snapshot, res=m.Snapshot) snap, _ = user.post(snapshot, res=m.Snapshot)
pc, _ = user.get(res=d.Device, item=1) 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'
@ -664,8 +666,8 @@ def test_hid_with_2network_and_drop_no_mac_in_hid(app: Devicehub, user: UserClie
network2 = copy.copy(network) network2 = copy.copy(network)
snapshot['components'].append(network2) snapshot['components'].append(network2)
network['serialNumber'] = 'a0:24:8c:7f:cf:2d' network['serialNumber'] = 'a0:24:8c:7f:cf:2d'
user.post(snapshot, res=m.Snapshot) snap, _ = user.post(snapshot, res=m.Snapshot)
pc, _ = user.get(res=d.Device, item=1) pc, _ = user.get(res=d.Device, item=snap['device']['devicehubID'])
assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d' assert pc['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'
@ -687,8 +689,8 @@ def test_hid_with_2network_and_drop_mac_in_hid(app: Devicehub, user: UserClient)
network2 = copy.copy(network) network2 = copy.copy(network)
snapshot['components'].append(network2) snapshot['components'].append(network2)
network['serialNumber'] = 'a0:24:8c:7f:cf:2d' network['serialNumber'] = 'a0:24:8c:7f:cf:2d'
user.post(snapshot, res=m.Snapshot) snap, _ = user.post(snapshot, res=m.Snapshot)
pc, _ = user.get(res=d.Device, item=1) pc, _ = user.get(res=d.Device, item=snap['device']['devicehubID'])
assert pc['hid'] == 'laptop-asustek_computer_inc-1000h-94oaaq021116-00:24:8c:7f:cf:2d' assert pc['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

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."""
user.post(conftest.file('basic.snapshot'), res=Snapshot) snap, _ = 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/1' assert i['items'][0]['url'] == '/devices/%s' % snap['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

View File

@ -174,7 +174,7 @@ def test_live_export_csv2(user: UserClient, client: Client, app: Devicehub):
assert "4692" in csv_user assert "4692" in csv_user
assert "8692" in csv_user assert "8692" in csv_user
assert "SystemId" in csv_user assert "DevicehubID" in csv_user
@pytest.mark.mvp @pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__) @pytest.mark.usefixtures(conftest.app_context.__name__)
@ -200,7 +200,7 @@ def test_live_example2(user: UserClient, client: Client, app: Devicehub):
assert str(action_live[0].snapshot_uuid) == acer['uuid'] assert str(action_live[0].snapshot_uuid) == acer['uuid']
@pytest.mark.mvp @pytest.mark.mvp
def test_export_basic_snapshot(user: UserClient): def test_export_basic_snapshot(user: UserClient):
"""Test export device information in a csv file.""" """Test export device information in a csv file."""
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot) snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)

View File

@ -365,13 +365,14 @@ def test_lot_post_add_remove_device_view(app: Devicehub, user: UserClient):
db.session.add(device) db.session.add(device)
db.session.commit() db.session.commit()
device_id = device.id device_id = device.id
devicehub_id = device.devicehub_id
parent, _ = user.post(({'name': 'lot'}), res=Lot) parent, _ = user.post(({'name': 'lot'}), res=Lot)
lot, _ = user.post({}, lot, _ = user.post({},
res=Lot, res=Lot,
item='{}/devices'.format(parent['id']), item='{}/devices'.format(parent['id']),
query=[('id', device_id)]) query=[('id', device_id)])
assert lot['devices'][0]['id'] == device_id, 'Lot contains device' assert lot['devices'][0]['id'] == device_id, 'Lot contains device'
device, _ = user.get(res=Device, item=device_id) device, _ = user.get(res=Device, item=devicehub_id)
assert len(device['lots']) == 1 assert len(device['lots']) == 1
assert device['lots'][0]['id'] == lot['id'], 'Device is inside lot' assert device['lots'][0]['id'] == lot['id'], 'Device is inside lot'

View File

@ -39,7 +39,7 @@ def test_rate_with_multiple_visual_tests(user: UserClient):
""" """
s = file('real-hp.snapshot.11') s = file('real-hp.snapshot.11')
snapshot, _ = user.post(s, res=Snapshot) snapshot, _ = user.post(s, res=Snapshot)
device, _ = user.get(res=Device, item=snapshot['device']['id']) device, _ = user.get(res=Device, item=snapshot['device']['devicehubID'])
visual_test = next(e for e in reversed(device['actions']) if e['type'] == VisualTest.t) visual_test = next(e for e in reversed(device['actions']) if e['type'] == VisualTest.t)
assert visual_test['appearanceRange'] == 'B' assert visual_test['appearanceRange'] == 'B'
@ -53,7 +53,7 @@ def test_rate_with_multiple_visual_tests(user: UserClient):
'appearanceRange': 'A', 'appearanceRange': 'A',
'functionalityRange': 'A' 'functionalityRange': 'A'
}, res=Action) }, res=Action)
device, _ = user.get(res=Device, item=snapshot['device']['id']) device, _ = user.get(res=Device, item=snapshot['device']['devicehubID'])
visual_test = next(e for e in reversed(device['actions']) if e['type'] == VisualTest.t) visual_test = next(e for e in reversed(device['actions']) if e['type'] == VisualTest.t)
assert visual_test['appearanceRange'] == 'A' assert visual_test['appearanceRange'] == 'A'

View File

@ -63,7 +63,7 @@ def test_snapshot_model():
assert m.Desktop.query.one_or_none() is None assert m.Desktop.query.one_or_none() is None
assert m.Device.query.one_or_none() is None assert m.Device.query.one_or_none() is None
# Check properties # Check properties
assert device.url == urlutils.URL('http://localhost/devices/1') assert device.url == urlutils.URL('http://localhost/devices/%s' % device.devicehub_id)
@pytest.mark.mvp @pytest.mark.mvp
@ -92,7 +92,7 @@ def test_snapshot_post(user: UserClient):
assert snapshot['author']['id'] == user.user['id'] assert snapshot['author']['id'] == user.user['id']
assert 'actions' not in snapshot['device'] assert 'actions' not in snapshot['device']
assert 'author' not in snapshot['device'] assert 'author' not in snapshot['device']
device, _ = user.get(res=m.Device, item=snapshot['device']['id']) device, _ = user.get(res=m.Device, item=snapshot['device']['devicehubID'])
key = itemgetter('serialNumber') key = itemgetter('serialNumber')
snapshot['components'].sort(key=key) snapshot['components'].sort(key=key)
device['components'].sort(key=key) device['components'].sort(key=key)
@ -115,7 +115,8 @@ def test_same_device_tow_users(user: UserClient, user2: UserClient):
i, _ = user.get(res=m.Device) i, _ = user.get(res=m.Device)
pc = next(d for d in i['items'] if d['type'] == 'Desktop') pc = next(d for d in i['items'] if d['type'] == 'Desktop')
pc_id = pc['id'] pc_id = pc['id']
assert i['items'][0]['url'] == f'/devices/{pc_id}' devicehub_id = pc['devicehubID']
assert i['items'][0]['url'] == f'/devices/{devicehub_id}'
basic_snapshot = file('basic.snapshot') basic_snapshot = file('basic.snapshot')
basic_snapshot['uuid'] = f"{uuid.uuid4()}" basic_snapshot['uuid'] = f"{uuid.uuid4()}"
@ -140,8 +141,8 @@ def test_snapshot_update_timefield_updated(user: UserClient):
computer2 = file('2-second-device-with-components-of-first.snapshot') computer2 = file('2-second-device-with-components-of-first.snapshot')
snapshot_and_check(user, computer2, action_types=('Remove', 'RateComputer'), snapshot_and_check(user, computer2, action_types=('Remove', 'RateComputer'),
perform_second_snapshot=False) perform_second_snapshot=False)
pc1_id = snapshot['device']['id'] pc1_devicehub_id = snapshot['device']['devicehubID']
pc1, _ = user.get(res=m.Device, item=pc1_id) pc1, _ = user.get(res=m.Device, item=pc1_devicehub_id)
assert pc1['updated'] != snapshot['device']['updated'] assert pc1['updated'] != snapshot['device']['updated']
@ -170,7 +171,8 @@ def test_snapshot_component_add_remove(user: UserClient):
RateComputer.t), RateComputer.t),
perform_second_snapshot=False) perform_second_snapshot=False)
pc1_id = snapshot1['device']['id'] pc1_id = snapshot1['device']['id']
pc1, _ = user.get(res=m.Device, item=pc1_id) pc1_devicehub_id = snapshot1['device']['devicehubID']
pc1, _ = user.get(res=m.Device, item=pc1_devicehub_id)
update1_pc1 = pc1['updated'] update1_pc1 = pc1['updated']
# Parent contains components # Parent contains components
assert tuple(c['serialNumber'] for c in pc1['components']) == ('p1c1s', 'p1c2s', 'p1c3s') assert tuple(c['serialNumber'] for c in pc1['components']) == ('p1c1s', 'p1c2s', 'p1c3s')
@ -180,7 +182,7 @@ def test_snapshot_component_add_remove(user: UserClient):
assert len(pc1['actions']) == 3 assert len(pc1['actions']) == 3
assert pc1['actions'][1]['type'] == Snapshot.t assert pc1['actions'][1]['type'] == Snapshot.t
# p1c1s has Snapshot # p1c1s has Snapshot
p1c1s, _ = user.get(res=m.Device, item=pc1['components'][0]['id']) p1c1s, _ = user.get(res=m.Device, item=pc1['components'][0]['devicehubID'])
assert tuple(e['type'] for e in p1c1s['actions']) == ('Snapshot', 'RateComputer') assert tuple(e['type'] for e in p1c1s['actions']) == ('Snapshot', 'RateComputer')
# We register a new device # We register a new device
@ -192,8 +194,9 @@ def test_snapshot_component_add_remove(user: UserClient):
snapshot2 = snapshot_and_check(user, s2, action_types=('Remove', 'RateComputer'), snapshot2 = snapshot_and_check(user, s2, action_types=('Remove', 'RateComputer'),
perform_second_snapshot=False) perform_second_snapshot=False)
pc2_id = snapshot2['device']['id'] pc2_id = snapshot2['device']['id']
pc1, _ = user.get(res=m.Device, item=pc1_id) pc2_devicehub_id = snapshot2['device']['devicehubID']
pc2, _ = user.get(res=m.Device, item=pc2_id) pc1, _ = user.get(res=m.Device, item=pc1_devicehub_id)
pc2, _ = user.get(res=m.Device, item=pc2_devicehub_id)
# Check if the update_timestamp is updated # Check if the update_timestamp is updated
update1_pc2 = pc2['updated'] update1_pc2 = pc2['updated']
update2_pc1 = pc1['updated'] update2_pc1 = pc1['updated']
@ -207,7 +210,7 @@ def test_snapshot_component_add_remove(user: UserClient):
assert all(c['parent'] == pc2_id for c in pc2['components']) assert all(c['parent'] == pc2_id for c in pc2['components'])
assert tuple(e['type'] for e in pc2['actions']) == ('Snapshot', 'RateComputer') assert tuple(e['type'] for e in pc2['actions']) == ('Snapshot', 'RateComputer')
# p1c2s has two Snapshots, a Remove and an Add # p1c2s has two Snapshots, a Remove and an Add
p1c2s, _ = user.get(res=m.Device, item=pc2['components'][0]['id']) p1c2s, _ = user.get(res=m.Device, item=pc2['components'][0]['devicehubID'])
assert tuple(e['type'] for e in p1c2s['actions']) == ( assert tuple(e['type'] for e in p1c2s['actions']) == (
'BenchmarkProcessor', 'Snapshot', 'RateComputer', 'Snapshot', 'Remove', 'RateComputer' 'BenchmarkProcessor', 'Snapshot', 'RateComputer', 'Snapshot', 'Remove', 'RateComputer'
) )
@ -218,8 +221,8 @@ def test_snapshot_component_add_remove(user: UserClient):
# PC 0: p1c2s, p1c3s. PC 1: p2c1s # PC 0: p1c2s, p1c3s. PC 1: p2c1s
s3 = file('3-first-device-but-removing-motherboard-and-adding-processor-from-2.snapshot') s3 = file('3-first-device-but-removing-motherboard-and-adding-processor-from-2.snapshot')
snapshot_and_check(user, s3, ('Remove', 'RateComputer'), perform_second_snapshot=False) snapshot_and_check(user, s3, ('Remove', 'RateComputer'), perform_second_snapshot=False)
pc1, _ = user.get(res=m.Device, item=pc1_id) pc1, _ = user.get(res=m.Device, item=pc1_devicehub_id)
pc2, _ = user.get(res=m.Device, item=pc2_id) pc2, _ = user.get(res=m.Device, item=pc2_devicehub_id)
# Check if the update_timestamp is updated # Check if the update_timestamp is updated
update2_pc2 = pc2['updated'] update2_pc2 = pc2['updated']
update3_pc1 = pc1['updated'] update3_pc1 = pc1['updated']
@ -247,7 +250,7 @@ def test_snapshot_component_add_remove(user: UserClient):
'Remove' # the processor we added in 2. 'Remove' # the processor we added in 2.
) )
# p1c2s has Snapshot, Remove and Add # p1c2s has Snapshot, Remove and Add
p1c2s, _ = user.get(res=m.Device, item=pc1['components'][0]['id']) p1c2s, _ = user.get(res=m.Device, item=pc1['components'][0]['devicehubID'])
assert tuple(get_actions_info(p1c2s['actions'])) == ( assert tuple(get_actions_info(p1c2s['actions'])) == (
('BenchmarkProcessor', []), # first BenchmarkProcessor ('BenchmarkProcessor', []), # first BenchmarkProcessor
('Snapshot', ['p1c1s', 'p1c2s', 'p1c3s']), # First Snapshot to PC1 ('Snapshot', ['p1c1s', 'p1c2s', 'p1c3s']), # First Snapshot to PC1
@ -264,8 +267,8 @@ def test_snapshot_component_add_remove(user: UserClient):
# adding a graphic card and adding a new component # adding a graphic card and adding a new component
s4 = file('4-first-device-but-removing-processor.snapshot-and-adding-graphic-card') s4 = file('4-first-device-but-removing-processor.snapshot-and-adding-graphic-card')
snapshot4 = snapshot_and_check(user, s4, ('RateComputer',), perform_second_snapshot=False) snapshot4 = snapshot_and_check(user, s4, ('RateComputer',), perform_second_snapshot=False)
pc1, _ = user.get(res=m.Device, item=pc1_id) pc1, _ = user.get(res=m.Device, item=pc1_devicehub_id)
pc2, _ = user.get(res=m.Device, item=pc2_id) pc2, _ = user.get(res=m.Device, item=pc2_devicehub_id)
# Check if the update_timestamp is updated # Check if the update_timestamp is updated
update3_pc2 = pc2['updated'] update3_pc2 = pc2['updated']
update4_pc1 = pc1['updated'] update4_pc1 = pc1['updated']
@ -429,7 +432,7 @@ def test_erase_privacy_standards_endtime_sort(user: UserClient):
# The actual test # The actual test
storage = next(e for e in snapshot['components'] if e['type'] == SolidStateDrive.t) storage = next(e for e in snapshot['components'] if e['type'] == SolidStateDrive.t)
storage, _ = user.get(res=m.Device, item=storage['id']) # Let's get storage actions too storage, _ = user.get(res=m.Device, item=storage['devicehubID']) # Let's get storage actions too
# order: endTime ascending # order: endTime ascending
# erasure1/2 have an user defined time and others actions endTime = created # erasure1/2 have an user defined time and others actions endTime = created
erasure1, erasure2, benchmark_hdd1, _snapshot1, _, _, benchmark_hdd2, _snapshot2 = storage['actions'][:8] erasure1, erasure2, benchmark_hdd1, _snapshot1, _, _, benchmark_hdd2, _snapshot2 = storage['actions'][:8]
@ -455,17 +458,17 @@ def test_erase_privacy_standards_endtime_sort(user: UserClient):
assert 'num' not in step2 assert 'num' not in step2
assert ['HMG_IS5'] == erasure['standards'] assert ['HMG_IS5'] == erasure['standards']
assert storage['privacy']['type'] == 'EraseSectors' assert storage['privacy']['type'] == 'EraseSectors'
pc, _ = user.get(res=m.Device, item=snapshot['device']['id']) pc, _ = user.get(res=m.Device, item=snapshot['device']['devicehubID'])
assert pc['privacy'] == [storage['privacy']] assert pc['privacy'] == [storage['privacy']]
# Let's try a second erasure with an error # Let's try a second erasure with an error
s['uuid'] = uuid4() s['uuid'] = uuid4()
s['components'][0]['actions'][0]['severity'] = 'Error' s['components'][0]['actions'][0]['severity'] = 'Error'
snapshot, _ = user.post(s, res=Snapshot) snapshot, _ = user.post(s, res=Snapshot)
storage, _ = user.get(res=m.Device, item=storage['id']) storage, _ = user.get(res=m.Device, item=storage['devicehubID'])
assert storage['hid'] == 'solidstatedrive-c1mr-c1ml-c1s' assert storage['hid'] == 'solidstatedrive-c1mr-c1ml-c1s'
assert storage['privacy']['type'] == 'EraseSectors' assert storage['privacy']['type'] == 'EraseSectors'
pc, _ = user.get(res=m.Device, item=snapshot['device']['id']) pc, _ = user.get(res=m.Device, item=snapshot['device']['devicehubID'])
assert pc['privacy'] == [storage['privacy']] assert pc['privacy'] == [storage['privacy']]
@ -548,7 +551,7 @@ def snapshot_and_check(user: UserClient,
@pytest.mark.mvp @pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__) @pytest.mark.usefixtures(conftest.app_context.__name__)
def test_erase_changing_hdd_between_pcs(user: UserClient): def test_erase_changing_hdd_between_pcs(user: UserClient):
"""Tests when we erase one device and next change the disk in other device we """Tests when we erase one device and next change the disk in other device we
want see in the second device the disks erase.""" want see in the second device the disks erase."""
s1 = file('erase-sectors-2-hdd.snapshot') s1 = file('erase-sectors-2-hdd.snapshot')
s2 = file('erase-sectors-2-hdd.snapshot2') s2 = file('erase-sectors-2-hdd.snapshot2')
@ -615,7 +618,7 @@ def test_save_snapshot_in_file(app: Devicehub, user: UserClient):
@pytest.mark.mvp @pytest.mark.mvp
def test_action_no_snapshot_without_save_file(app: Devicehub, user: UserClient): def test_action_no_snapshot_without_save_file(app: Devicehub, user: UserClient):
""" This test check if the function save_snapshot_in_file not work when we """ This test check if the function save_snapshot_in_file not work when we
send one other action different to snapshot send one other action different to snapshot
""" """
s = file('laptop-hp_255_g3_notebook-hewlett-packard-cnd52270fw.snapshot') s = file('laptop-hp_255_g3_notebook-hewlett-packard-cnd52270fw.snapshot')
@ -736,7 +739,7 @@ def test_snapshot_not_failed_null_chassis(app: Devicehub, user: UserClient):
tmp_snapshots = app.config['TMP_SNAPSHOTS'] tmp_snapshots = app.config['TMP_SNAPSHOTS']
path_dir_base = os.path.join(tmp_snapshots, user.user['email'], 'errors') path_dir_base = os.path.join(tmp_snapshots, user.user['email'], 'errors')
snapshot_error = file('desktop-9644w8n-lenovo-0169622.snapshot') snapshot_error = file('desktop-9644w8n-lenovo-0169622.snapshot')
snapshot_error['device']['chassis'] = None snapshot_error['device']['chassis'] = None
uuid = snapshot_error['uuid'] uuid = snapshot_error['uuid']
snapshot, res = user.post(res=Snapshot, data=snapshot_error) snapshot, res = user.post(res=Snapshot, data=snapshot_error)
@ -780,7 +783,7 @@ def test_snapshot_failed_end_time_bug(app: Devicehub, user: UserClient):
""" """
snapshot_file = file('asus-end_time_bug88.snapshot') snapshot_file = file('asus-end_time_bug88.snapshot')
snapshot, _ = user.post(res=Snapshot, data=snapshot_file) snapshot, _ = user.post(res=Snapshot, data=snapshot_file)
device, _ = user.get(res=m.Device, item=snapshot['device']['id']) device, _ = user.get(res=m.Device, item=snapshot['device']['devicehubID'])
end_times = [x['endTime'] for x in device['actions']] end_times = [x['endTime'] for x in device['actions']]
assert '1970-01-02T00:00:00+00:00' in end_times assert '1970-01-02T00:00:00+00:00' in end_times
@ -797,7 +800,7 @@ def test_snapshot_not_failed_end_time_bug(app: Devicehub, user: UserClient):
snapshot_file = file('asus-end_time_bug88.snapshot') snapshot_file = file('asus-end_time_bug88.snapshot')
snapshot_file['endTime'] = '2001-01-01 00:00:00+00:00' snapshot_file['endTime'] = '2001-01-01 00:00:00+00:00'
snapshot, _ = user.post(res=Snapshot, data=snapshot_file) snapshot, _ = user.post(res=Snapshot, data=snapshot_file)
device, _ = user.get(res=m.Device, item=snapshot['device']['id']) device, _ = user.get(res=m.Device, item=snapshot['device']['devicehubID'])
end_times = [x['endTime'] for x in device['actions']] end_times = [x['endTime'] for x in device['actions']]
assert not '1970-01-02T00:00:00+00:00' in end_times assert not '1970-01-02T00:00:00+00:00' in end_times
@ -817,7 +820,7 @@ def test_snapshot_bug_smallint_hdd(app: Devicehub, user: UserClient):
snapshot, _ = user.post(res=Snapshot, data=snapshot_file) snapshot, _ = user.post(res=Snapshot, data=snapshot_file)
act = [act for act in snapshot['actions'] if act['type'] == 'TestDataStorage'][0] act = [act for act in snapshot['actions'] if act['type'] == 'TestDataStorage'][0]
assert act['currentPendingSectorCount'] == 473302660 assert act['currentPendingSectorCount'] == 473302660
assert act['offlineUncorrectable'] == 182042944 assert act['offlineUncorrectable'] == 182042944
tmp_snapshots = app.config['TMP_SNAPSHOTS'] tmp_snapshots = app.config['TMP_SNAPSHOTS']
@ -831,9 +834,7 @@ def test_snapshot_mobil(app: Devicehub, user: UserClient):
""" """
snapshot_file = file('mobil') snapshot_file = file('mobil')
snapshot, _ = user.post(res=Snapshot, data=snapshot_file) snapshot, _ = user.post(res=Snapshot, data=snapshot_file)
device, _ = user.get(res=m.Device, item=snapshot['device']['id']) device, _ = user.get(res=m.Device, item=snapshot['device']['devicehubID'])
tmp_snapshots = app.config['TMP_SNAPSHOTS'] tmp_snapshots = app.config['TMP_SNAPSHOTS']
shutil.rmtree(tmp_snapshots) shutil.rmtree(tmp_snapshots)

View File

@ -281,8 +281,9 @@ def test_tag_manual_link_search(app: Devicehub, user: UserClient):
db.session.add(desktop) db.session.add(desktop)
db.session.commit() db.session.commit()
desktop_id = desktop.id desktop_id = desktop.id
devicehub_id = desktop.devicehub_id
user.put({}, res=Tag, item='foo-bar/device/{}'.format(desktop_id), status=204) user.put({}, res=Tag, item='foo-bar/device/{}'.format(desktop_id), status=204)
device, _ = user.get(res=Device, item=1) device, _ = user.get(res=Device, item=devicehub_id)
assert device['tags'][0]['id'] == 'foo-bar' assert device['tags'][0]['id'] == 'foo-bar'
# Device already linked # Device already linked
@ -321,7 +322,7 @@ def test_tag_secondary_workbench_link_find(user: UserClient):
s = file('basic.snapshot') s = file('basic.snapshot')
s['device']['tags'] = [{'id': 'foo', 'secondary': 'bar', 'type': 'Tag'}] s['device']['tags'] = [{'id': 'foo', 'secondary': 'bar', 'type': 'Tag'}]
snapshot, _ = user.post(s, res=Snapshot) snapshot, _ = user.post(s, res=Snapshot)
device, _ = user.get(res=Device, item=snapshot['device']['id']) device, _ = user.get(res=Device, item=snapshot['device']['devicehubID'])
assert device['tags'][0]['id'] == 'foo' assert device['tags'][0]['id'] == 'foo'
assert device['tags'][0]['secondary'] == 'bar' assert device['tags'][0]['secondary'] == 'bar'

View File

@ -49,7 +49,7 @@ def test_workbench_server_condensed(user: UserClient):
} }
assert snapshot['closed'] assert snapshot['closed']
assert snapshot['severity'] == 'Info' assert snapshot['severity'] == 'Info'
device, _ = user.get(res=Device, item=snapshot['device']['id']) device, _ = user.get(res=Device, item=snapshot['device']['devicehubID'])
assert device['dataStorageSize'] == 1100 assert device['dataStorageSize'] == 1100
assert device['chassis'] == 'Tower' assert device['chassis'] == 'Tower'
assert device['hid'] == 'desktop-d1mr-d1ml-d1s-na1-s' assert device['hid'] == 'desktop-d1mr-d1ml-d1s-na1-s'
@ -129,7 +129,7 @@ def test_workbench_server_phases(user: UserClient):
assert snapshot['closed'] assert snapshot['closed']
assert snapshot['severity'] == 'Info' assert snapshot['severity'] == 'Info'
pc, _ = user.get(res=Device, item=snapshot['id']) pc, _ = user.get(res=Device, item=snapshot['devicehubID'])
assert len(pc['actions']) == 10 # todo shall I add child actions? assert len(pc['actions']) == 10 # todo shall I add child actions?
@ -173,7 +173,7 @@ def test_snapshot_real_eee_1001pxd_with_rate(user: UserClient):
""" """
s = file('real-eee-1001pxd.snapshot.11') s = file('real-eee-1001pxd.snapshot.11')
snapshot, _ = user.post(res=em.Snapshot, data=s) snapshot, _ = user.post(res=em.Snapshot, data=s)
pc, _ = user.get(res=Device, item=snapshot['device']['id']) pc, _ = user.get(res=Device, item=snapshot['device']['devicehubID'])
assert pc['type'] == 'Laptop' assert pc['type'] == 'Laptop'
assert pc['chassis'] == 'Netbook' assert pc['chassis'] == 'Netbook'
assert pc['model'] == '1001pxd' assert pc['model'] == '1001pxd'
@ -213,7 +213,7 @@ def test_snapshot_real_eee_1001pxd_with_rate(user: UserClient):
assert cpu['speed'] == 1.667 assert cpu['speed'] == 1.667
assert 'hid' not in cpu assert 'hid' not in cpu
assert pc['processorModel'] == cpu['model'] == 'intel atom cpu n455 @ 1.66ghz' assert pc['processorModel'] == cpu['model'] == 'intel atom cpu n455 @ 1.66ghz'
cpu, _ = user.get(res=Device, item=cpu['id']) cpu, _ = user.get(res=Device, item=cpu['devicehubID'])
actions = cpu['actions'] actions = cpu['actions']
sysbench = next(e for e in actions if e['type'] == em.BenchmarkProcessorSysbench.t) sysbench = next(e for e in actions if e['type'] == em.BenchmarkProcessorSysbench.t)
assert sysbench['elapsed'] == 164 assert sysbench['elapsed'] == 164
@ -233,7 +233,7 @@ def test_snapshot_real_eee_1001pxd_with_rate(user: UserClient):
assert gpu['model'] == 'atom processor d4xx/d5xx/n4xx/n5xx integrated graphics controller' assert gpu['model'] == 'atom processor d4xx/d5xx/n4xx/n5xx integrated graphics controller'
assert gpu['manufacturer'] == 'intel corporation' assert gpu['manufacturer'] == 'intel corporation'
assert gpu['memory'] == 256 assert gpu['memory'] == 256
gpu, _ = user.get(res=Device, item=gpu['id']) gpu, _ = user.get(res=Device, item=gpu['devicehubID'])
action_types = tuple(e['type'] for e in gpu['actions']) action_types = tuple(e['type'] for e in gpu['actions'])
assert em.BenchmarkRamSysbench.t in action_types assert em.BenchmarkRamSysbench.t in action_types
assert em.StressTest.t in action_types assert em.StressTest.t in action_types
@ -252,7 +252,7 @@ def test_snapshot_real_eee_1001pxd_with_rate(user: UserClient):
assert hdd['hid'] == 'harddrive-hitachi-hts54322-e2024242cv86hj' assert hdd['hid'] == 'harddrive-hitachi-hts54322-e2024242cv86hj'
assert hdd['interface'] == 'ATA' assert hdd['interface'] == 'ATA'
assert hdd['size'] == 238475 assert hdd['size'] == 238475
hdd, _ = user.get(res=Device, item=hdd['id']) hdd, _ = user.get(res=Device, item=hdd['devicehubID'])
action_types = tuple(e['type'] for e in hdd['actions']) action_types = tuple(e['type'] for e in hdd['actions'])
assert em.BenchmarkRamSysbench.t in action_types assert em.BenchmarkRamSysbench.t in action_types
assert em.StressTest.t in action_types assert em.StressTest.t in action_types