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

Merge pull request #302 from eReuse/changes/hdi 

Changes/hid
This commit is contained in:
cayop 2022-06-24 12:03:01 +02:00 committed by GitHub
parent cd3439b3ac 1f85abc1e2
commit ac883d514c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 4946 additions and 231 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@ -8,6 +8,7 @@ ml).
## master
## testing
- [changed] #302 add system uuid for check the identity of one device.
## [2.2.0] - 2022-06-24
- [changed] #304 change anchor of link devices lots.

37
ereuse_devicehub/inventory/forms.py
View File

@ -29,7 +29,6 @@ from wtforms.fields import FormField
from ereuse_devicehub.db import db
from ereuse_devicehub.inventory.models import DeliveryNote, ReceiverNote, Transfer
from ereuse_devicehub.parser.models import SnapshotsLog
from ereuse_devicehub.parser.parser import ParseSnapshotLsHw
from ereuse_devicehub.parser.schemas import Snapshot_lite
from ereuse_devicehub.resources.action.models import Snapshot, Trade
@ -260,45 +259,35 @@ class UploadSnapshotForm(SnapshotMixin, FlaskForm):
self.tmp_snapshots = app.config['TMP_SNAPSHOTS']
for filename, snapshot_json in self.snapshots:
path_snapshot = save_json(snapshot_json, self.tmp_snapshots, g.user.email)
snapshot_json.pop('debug', None)
version = snapshot_json.get('schema_api')
uuid = snapshot_json.get('uuid')
sid = snapshot_json.get('sid')
software_version = snapshot_json.get('version')
if self.is_wb_lite_snapshot(version):
debug = snapshot_json.pop('debug', None)
self.version = snapshot_json.get('schema_api')
self.uuid = snapshot_json.get('uuid')
self.sid = snapshot_json.get('sid')
if self.is_wb_lite_snapshot(self.version):
self.snapshot_json = schema_lite.load(snapshot_json)
snapshot_json = ParseSnapshotLsHw(self.snapshot_json).snapshot_json
else:
self.version = snapshot_json.get('version')
system_uuid = self.get_uuid(debug)
if system_uuid:
snapshot_json['device']['system_uuid'] = system_uuid
try:
snapshot_json = schema.load(snapshot_json)
except ValidationError as err:
txt = "{}".format(err)
error = SnapshotsLog(
description=txt,
snapshot_uuid=uuid,
severity=Severity.Error,
sid=sid,
version=software_version,
)
error.save(commit=True)
self.errors(txt=txt)
self.result[filename] = 'Error'
continue
response = self.build(snapshot_json)
db.session.add(response)
devices.append(response.device)
snap_log = SnapshotsLog(
description='Ok',
snapshot_uuid=uuid,
severity=Severity.Info,
sid=sid,
version=software_version,
snapshot=response,
)
snap_log.save()
if hasattr(response, 'type'):
self.result[filename] = 'Ok'
self.errors(txt="Ok", severity=Severity.Info, snapshot=response)
else:
self.result[filename] = 'Error'

6
ereuse_devicehub/inventory/views.py
View File

@ -702,12 +702,18 @@ class SnapshotListView(GenericMixin):
).order_by(SnapshotsLog.created.desc())
logs = {}
for snap in snapshots_log:
try:
system_uuid = snap.snapshot.device.system_uuid or ''
except AttributeError:
system_uuid = ''
if snap.snapshot_uuid not in logs:
logs[snap.snapshot_uuid] = {
'sid': snap.sid,
'snapshot_uuid': snap.snapshot_uuid,
'version': snap.version,
'device': snap.get_device(),
'system_uuid': system_uuid,
'status': snap.get_status(),
'severity': snap.severity,
'created': snap.created,

33
ereuse_devicehub/migrations/versions/73348969a583_system_uuid_instead_of_uuid.py Normal file
View File

@ -0,0 +1,33 @@
"""system_uuid instead of uuid
Revision ID: 73348969a583
Revises: dac62da1621a
Create Date: 2022-06-15 12:27:23.170313
"""
from alembic import context, op
# revision identifiers, used by Alembic.
revision = '73348969a583'
down_revision = 'dac62da1621a'
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():
op.alter_column(
'computer', 'uuid', new_column_name="system_uuid", schema=f'{get_inv()}'
)
def downgrade():
op.alter_column(
'computer', 'system_uuid', new_column_name="uuid", schema=f'{get_inv()}'
)

67
ereuse_devicehub/migrations/versions/8d4fe4b497b3_add_system_uuid_to_old_registers.py Normal file
View File

@ -0,0 +1,67 @@
"""add system uuid to old registers
Revision ID: 8d4fe4b497b3
Revises: 73348969a583
Create Date: 2022-06-15 15:52:39.205192
"""
import os
from uuid import UUID
from alembic import context, op
# revision identifiers, used by Alembic.
revision = '8d4fe4b497b3'
down_revision = '73348969a583'
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 update_db(con, system_uuid, snapshot_uuid):
sql_snapshot = f'select id from {get_inv()}.snapshot where uuid=\'{snapshot_uuid}\''
sql_device_id = f'select device_id from {get_inv()}.action_with_one_device where id in ({sql_snapshot})'
sql = f'select id, system_uuid from {get_inv()}.computer where id in ({sql_device_id})'
for device_id, db_system_uuid in con.execute(sql):
if db_system_uuid:
return
sql = f'update {get_inv()}.computer set system_uuid=\'{system_uuid}\' where id=\'{device_id}\''
con.execute(sql)
def update_to_little_endian(uuid):
uuid = UUID(uuid)
return UUID(bytes_le=uuid.bytes)
def upgrade():
uuids = []
system_uuids_file = 'system_uuids.csv'
if os.path.exists(system_uuids_file):
with open(system_uuids_file) as f:
for x in f.read().split('\n'):
z = x.split(';')
if len(z) != 2:
continue
x, y = z
uuids.append([x.strip(), y.strip()])
con = op.get_bind()
for u in uuids[1:]:
if u[0] == '':
continue
u[0] = update_to_little_endian(u[0])
update_db(con, u[0], u[1])
def downgrade():
pass

4
ereuse_devicehub/parser/parser.py
View File

@ -52,7 +52,7 @@ class ParseSnapshot:
self.device['type'] = self.get_type()
self.device['sku'] = self.get_sku()
self.device['version'] = self.get_version()
self.device['uuid'] = self.get_uuid()
self.device['system_uuid'] = self.get_uuid()
def set_components(self):
self.get_cpu()
@ -379,7 +379,7 @@ class ParseSnapshotLsHw:
raise ValidationError(txt)
self.device = pc
self.device['uuid'] = self.get_uuid()
self.device['system_uuid'] = self.get_uuid()
def set_components(self):
memory = None

74
ereuse_devicehub/resources/action/views/snapshot.py
View File

@ -4,10 +4,10 @@ import json
import os
import shutil
from datetime import datetime
from uuid import UUID
from flask import current_app as app
from flask import g
from marshmallow import ValidationError
from sqlalchemy.util import OrderedSet
from ereuse_devicehub.db import db
@ -117,6 +117,49 @@ class SnapshotMixin:
return snapshot
def get_old_smbios_version(self, debug):
capabilities = debug.get('lshw', {}).get('capabilities', {})
for x in capabilities.values():
if "SMBIOS version" in x:
e = x.split("SMBIOS version ")[1].split(".")
if int(e[0]) < 3 and int(e[1]) < 6:
self.errors(txt=x)
return True
return False
def get_uuid(self, debug):
if not debug or not isinstance(debug, dict):
self.errors(txt="There is not uuid")
return
if self.get_old_smbios_version(debug):
return
hw_uuid = debug.get('lshw', {}).get('configuration', {}).get('uuid')
if not hw_uuid:
self.errors(txt="There is not uuid")
return
uuid = UUID(hw_uuid)
return UUID(bytes_le=uuid.bytes)
def errors(self, txt=None, severity=Severity.Error, snapshot=None, commit=False):
if not txt:
return
from ereuse_devicehub.parser.models import SnapshotsLog
error = SnapshotsLog(
description=txt,
snapshot_uuid=self.uuid,
severity=severity,
sid=self.sid,
version=self.version,
snapshot=snapshot,
)
error.save(commit=commit)
class SnapshotView(SnapshotMixin):
"""Performs a Snapshot.
@ -129,38 +172,29 @@ class SnapshotView(SnapshotMixin):
# snapshot, and we want to wait to flush snapshot at the end
def __init__(self, snapshot_json: dict, resource_def, schema):
from ereuse_devicehub.parser.models import SnapshotsLog
self.schema = schema
self.resource_def = resource_def
self.tmp_snapshots = app.config['TMP_SNAPSHOTS']
self.path_snapshot = save_json(snapshot_json, self.tmp_snapshots, g.user.email)
snapshot_json.pop('debug', None)
self.version = snapshot_json.get('version')
self.uuid = snapshot_json.get('uuid')
self.sid = None
system_uuid = self.get_uuid(snapshot_json.pop('debug', None))
if system_uuid:
snapshot_json['device']['system_uuid'] = system_uuid
try:
self.snapshot_json = resource_def.schema.load(snapshot_json)
snapshot = self.build()
except Exception as err:
txt = "{}".format(err)
uuid = snapshot_json.get('uuid')
version = snapshot_json.get('version')
error = SnapshotsLog(
description=txt,
snapshot_uuid=uuid,
severity=Severity.Error,
version=str(version),
)
error.save(commit=True)
self.errors(txt=txt, commit=True)
raise err
db.session.add(snapshot)
snap_log = SnapshotsLog(
description='Ok',
snapshot_uuid=snapshot.uuid,
severity=Severity.Info,
version=str(snapshot.version),
snapshot=snapshot,
)
snap_log.save()
self.errors(txt="Ok", severity=Severity.Info, snapshot=snapshot, commit=False)
db.session().final_flush()
self.response = self.schema.jsonify(snapshot) # transform it back
self.response.status_code = 201

3
ereuse_devicehub/resources/device/models.py
View File

@ -191,6 +191,7 @@ class Device(Thing):
'image',
'allocated',
'devicehub_id',
'system_uuid',
'active',
}
@ -818,7 +819,7 @@ class Computer(Device):
transfer_state.comment = TransferState.__doc__
receiver_id = db.Column(UUID(as_uuid=True), db.ForeignKey(User.id), nullable=True)
receiver = db.relationship(User, primaryjoin=receiver_id == User.id)
uuid = db.Column(UUID(as_uuid=True), nullable=True)
system_uuid = db.Column(UUID(as_uuid=True), nullable=True)
def __init__(self, *args, **kwargs) -> None:
if args:

370
ereuse_devicehub/resources/device/schemas.py
View File

@ -1,17 +1,30 @@
import datetime
from marshmallow import post_load, pre_load, fields as f
from marshmallow.fields import Boolean, Date, DateTime, Float, Integer, List, Str, String, UUID, Dict
from marshmallow import fields as f
from marshmallow import post_load, pre_load
from marshmallow.fields import (
UUID,
Boolean,
Date,
DateTime,
Dict,
Float,
Integer,
List,
Str,
String,
)
from marshmallow.validate import Length, OneOf, Range
from sqlalchemy.util import OrderedSet
from stdnum import imei, meid
from teal.enums import Layouts
from teal.marshmallow import EnumField, SanitizedStr, URL, ValidationError
from teal.marshmallow import URL, EnumField, SanitizedStr, ValidationError
from teal.resource import Schema
from ereuse_devicehub.marshmallow import NestedOn
from ereuse_devicehub.resources import enums
from ereuse_devicehub.resources.device import models as m, states
from ereuse_devicehub.resources.device import models as m
from ereuse_devicehub.resources.device import states
from ereuse_devicehub.resources.models import STR_BIG_SIZE, STR_SIZE
from ereuse_devicehub.resources.schemas import Thing, UnitCodes
@ -20,58 +33,90 @@ class Device(Thing):
__doc__ = m.Device.__doc__
id = Integer(description=m.Device.id.comment, dump_only=True)
hid = SanitizedStr(lower=True, description=m.Device.hid.comment)
tags = NestedOn('Tag',
many=True,
collection_class=OrderedSet,
description='A set of tags that identify the device.')
model = SanitizedStr(lower=True,
validate=Length(max=STR_BIG_SIZE),
description=m.Device.model.comment)
manufacturer = SanitizedStr(lower=True,
validate=Length(max=STR_SIZE),
description=m.Device.manufacturer.comment)
serial_number = SanitizedStr(lower=True,
validate=Length(max=STR_BIG_SIZE),
data_key='serialNumber')
brand = SanitizedStr(validate=Length(max=STR_BIG_SIZE), description=m.Device.brand.comment)
generation = Integer(validate=Range(1, 100), description=m.Device.generation.comment)
tags = NestedOn(
'Tag',
many=True,
collection_class=OrderedSet,
description='A set of tags that identify the device.',
)
model = SanitizedStr(
lower=True,
validate=Length(max=STR_BIG_SIZE),
description=m.Device.model.comment,
)
manufacturer = SanitizedStr(
lower=True,
validate=Length(max=STR_SIZE),
description=m.Device.manufacturer.comment,
)
serial_number = SanitizedStr(
lower=True, validate=Length(max=STR_BIG_SIZE), data_key='serialNumber'
)
brand = SanitizedStr(
validate=Length(max=STR_BIG_SIZE), description=m.Device.brand.comment
)
generation = Integer(
validate=Range(1, 100), description=m.Device.generation.comment
)
version = SanitizedStr(description=m.Device.version)
weight = Float(validate=Range(0.1, 5), unit=UnitCodes.kgm, description=m.Device.weight.comment)
width = Float(validate=Range(0.1, 5), unit=UnitCodes.m, description=m.Device.width.comment)
height = Float(validate=Range(0.1, 5), unit=UnitCodes.m, description=m.Device.height.comment)
depth = Float(validate=Range(0.1, 5), unit=UnitCodes.m, description=m.Device.depth.comment)
weight = Float(
validate=Range(0.1, 5), unit=UnitCodes.kgm, description=m.Device.weight.comment
)
width = Float(
validate=Range(0.1, 5), unit=UnitCodes.m, description=m.Device.width.comment
)
height = Float(
validate=Range(0.1, 5), unit=UnitCodes.m, description=m.Device.height.comment
)
depth = Float(
validate=Range(0.1, 5), unit=UnitCodes.m, description=m.Device.depth.comment
)
# TODO TimeOut 2. Comment actions and lots if there are time out.
actions = NestedOn('Action', many=True, dump_only=True, description=m.Device.actions.__doc__)
actions = NestedOn(
'Action', many=True, dump_only=True, description=m.Device.actions.__doc__
)
# TODO TimeOut 2. Comment actions_one and lots if there are time out.
actions_one = NestedOn('Action', many=True, load_only=True, collection_class=OrderedSet)
problems = NestedOn('Action', many=True, dump_only=True, description=m.Device.problems.__doc__)
actions_one = NestedOn(
'Action', many=True, load_only=True, collection_class=OrderedSet
)
problems = NestedOn(
'Action', many=True, dump_only=True, description=m.Device.problems.__doc__
)
url = URL(dump_only=True, description=m.Device.url.__doc__)
# TODO TimeOut 2. Comment actions and lots if there are time out.
lots = NestedOn('Lot',
many=True,
dump_only=True,
description='The lots where this device is directly under.')
lots = NestedOn(
'Lot',
many=True,
dump_only=True,
description='The lots where this device is directly under.',
)
rate = NestedOn('Rate', dump_only=True, description=m.Device.rate.__doc__)
price = NestedOn('Price', dump_only=True, description=m.Device.price.__doc__)
tradings = Dict(dump_only=True, description='')
physical = EnumField(states.Physical, dump_only=True, description=m.Device.physical.__doc__)
traking = EnumField(states.Traking, dump_only=True, description=m.Device.physical.__doc__)
usage = EnumField(states.Usage, dump_only=True, description=m.Device.physical.__doc__)
physical = EnumField(
states.Physical, dump_only=True, description=m.Device.physical.__doc__
)
traking = EnumField(
states.Traking, dump_only=True, description=m.Device.physical.__doc__
)
usage = EnumField(
states.Usage, dump_only=True, description=m.Device.physical.__doc__
)
revoke = UUID(dump_only=True)
physical_possessor = NestedOn('Agent', dump_only=True, data_key='physicalPossessor')
production_date = DateTime('iso',
description=m.Device.updated.comment,
data_key='productionDate')
working = NestedOn('Action',
many=True,
dump_only=True,
description=m.Device.working.__doc__)
production_date = DateTime(
'iso', description=m.Device.updated.comment, data_key='productionDate'
)
working = NestedOn(
'Action', many=True, dump_only=True, description=m.Device.working.__doc__
)
variant = SanitizedStr(description=m.Device.variant.comment)
sku = SanitizedStr(description=m.Device.sku.comment)
image = URL(description=m.Device.image.comment)
allocated = Boolean(description=m.Device.allocated.comment)
devicehub_id = SanitizedStr(data_key='devicehubID',
description=m.Device.devicehub_id.comment)
devicehub_id = SanitizedStr(
data_key='devicehubID', description=m.Device.devicehub_id.comment
)
@pre_load
def from_actions_to_actions_one(self, data: dict):
@ -91,52 +136,73 @@ class Device(Thing):
@post_load
def validate_snapshot_actions(self, data):
"""Validates that only snapshot-related actions can be uploaded."""
from ereuse_devicehub.resources.action.models import EraseBasic, Test, Rate, Install, \
Benchmark
from ereuse_devicehub.resources.action.models import (
Benchmark,
EraseBasic,
Install,
Rate,
Test,
)
for action in data['actions_one']:
if not isinstance(action, (Install, EraseBasic, Rate, Test, Benchmark)):
raise ValidationError('You cannot upload {}'.format(action),
field_names=['actions'])
raise ValidationError(
'You cannot upload {}'.format(action), field_names=['actions']
)
class Computer(Device):
__doc__ = m.Computer.__doc__
# TODO TimeOut 1. Comment components if there are time out.
components = NestedOn('Component',
many=True,
dump_only=True,
collection_class=OrderedSet,
description='The components that are inside this computer.')
chassis = EnumField(enums.ComputerChassis,
description=m.Computer.chassis.comment)
ram_size = Integer(dump_only=True,
data_key='ramSize',
description=m.Computer.ram_size.__doc__)
data_storage_size = Integer(dump_only=True,
data_key='dataStorageSize',
description=m.Computer.data_storage_size.__doc__)
processor_model = Str(dump_only=True,
data_key='processorModel',
description=m.Computer.processor_model.__doc__)
graphic_card_model = Str(dump_only=True,
data_key='graphicCardModel',
description=m.Computer.graphic_card_model.__doc__)
network_speeds = List(Integer(dump_only=True),
dump_only=True,
data_key='networkSpeeds',
description=m.Computer.network_speeds.__doc__)
privacy = NestedOn('Action',
many=True,
dump_only=True,
collection_class=set,
description=m.Computer.privacy.__doc__)
amount = Integer(validate=f.validate.Range(min=0, max=100),
description=m.Computer.amount.__doc__)
components = NestedOn(
'Component',
many=True,
dump_only=True,
collection_class=OrderedSet,
description='The components that are inside this computer.',
)
chassis = EnumField(enums.ComputerChassis, description=m.Computer.chassis.comment)
ram_size = Integer(
dump_only=True, data_key='ramSize', description=m.Computer.ram_size.__doc__
)
data_storage_size = Integer(
dump_only=True,
data_key='dataStorageSize',
description=m.Computer.data_storage_size.__doc__,
)
processor_model = Str(
dump_only=True,
data_key='processorModel',
description=m.Computer.processor_model.__doc__,
)
graphic_card_model = Str(
dump_only=True,
data_key='graphicCardModel',
description=m.Computer.graphic_card_model.__doc__,
)
network_speeds = List(
Integer(dump_only=True),
dump_only=True,
data_key='networkSpeeds',
description=m.Computer.network_speeds.__doc__,
)
privacy = NestedOn(
'Action',
many=True,
dump_only=True,
collection_class=set,
description=m.Computer.privacy.__doc__,
)
amount = Integer(
validate=f.validate.Range(min=0, max=100), description=m.Computer.amount.__doc__
)
# author_id = NestedOn(s_user.User, only_query='author_id')
owner_id = UUID(data_key='ownerID')
transfer_state = EnumField(enums.TransferState, description=m.Computer.transfer_state.comment)
transfer_state = EnumField(
enums.TransferState, description=m.Computer.transfer_state.comment
)
receiver_id = UUID(data_key='receiverID')
uuid = UUID(required=False)
system_uuid = UUID(required=False)
class Desktop(Computer):
@ -155,29 +221,36 @@ class Server(Computer):
class DisplayMixin:
__doc__ = m.DisplayMixin.__doc__
size = Float(description=m.DisplayMixin.size.comment, validate=Range(2, 150))
technology = EnumField(enums.DisplayTech,
description=m.DisplayMixin.technology.comment)
resolution_width = Integer(data_key='resolutionWidth',
validate=Range(10, 20000),
description=m.DisplayMixin.resolution_width.comment,
)
resolution_height = Integer(data_key='resolutionHeight',
validate=Range(10, 20000),
description=m.DisplayMixin.resolution_height.comment,
)
technology = EnumField(
enums.DisplayTech, description=m.DisplayMixin.technology.comment
)
resolution_width = Integer(
data_key='resolutionWidth',
validate=Range(10, 20000),
description=m.DisplayMixin.resolution_width.comment,
)
resolution_height = Integer(
data_key='resolutionHeight',
validate=Range(10, 20000),
description=m.DisplayMixin.resolution_height.comment,
)
refresh_rate = Integer(data_key='refreshRate', validate=Range(10, 1000))
contrast_ratio = Integer(data_key='contrastRatio', validate=Range(100, 100000))
touchable = Boolean(description=m.DisplayMixin.touchable.comment)
aspect_ratio = String(dump_only=True, description=m.DisplayMixin.aspect_ratio.__doc__)
aspect_ratio = String(
dump_only=True, description=m.DisplayMixin.aspect_ratio.__doc__
)
widescreen = Boolean(dump_only=True, description=m.DisplayMixin.widescreen.__doc__)
class NetworkMixin:
__doc__ = m.NetworkMixin.__doc__
speed = Integer(validate=Range(min=10, max=10000),
unit=UnitCodes.mbps,
description=m.NetworkAdapter.speed.comment)
speed = Integer(
validate=Range(min=10, max=10000),
unit=UnitCodes.mbps,
description=m.NetworkAdapter.speed.comment,
)
wireless = Boolean(required=True)
@ -198,16 +271,22 @@ class Mobile(Device):
imei = Integer(description=m.Mobile.imei.comment)
meid = Str(description=m.Mobile.meid.comment)
ram_size = Integer(validate=Range(min=128, max=36000),
data_key='ramSize',
unit=UnitCodes.mbyte,
description=m.Mobile.ram_size.comment)
data_storage_size = Integer(validate=Range(0, 10 ** 8),
data_key='dataStorageSize',
description=m.Mobile.data_storage_size)
display_size = Float(validate=Range(min=0.1, max=30.0),
data_key='displaySize',
description=m.Mobile.display_size.comment)
ram_size = Integer(
validate=Range(min=128, max=36000),
data_key='ramSize',
unit=UnitCodes.mbyte,
description=m.Mobile.ram_size.comment,
)
data_storage_size = Integer(
validate=Range(0, 10**8),
data_key='dataStorageSize',
description=m.Mobile.data_storage_size,
)
display_size = Float(
validate=Range(min=0.1, max=30.0),
data_key='displaySize',
description=m.Mobile.display_size.comment,
)
@pre_load
def convert_check_imei(self, data):
@ -243,17 +322,21 @@ class Component(Device):
class GraphicCard(Component):
__doc__ = m.GraphicCard.__doc__
memory = Integer(validate=Range(0, 10000),
unit=UnitCodes.mbyte,
description=m.GraphicCard.memory.comment)
memory = Integer(
validate=Range(0, 10000),
unit=UnitCodes.mbyte,
description=m.GraphicCard.memory.comment,
)
class DataStorage(Component):
__doc__ = m.DataStorage.__doc__
size = Integer(validate=Range(0, 10 ** 8),
unit=UnitCodes.mbyte,
description=m.DataStorage.size.comment)
size = Integer(
validate=Range(0, 10**8),
unit=UnitCodes.mbyte,
description=m.DataStorage.size.comment,
)
interface = EnumField(enums.DataStorageInterface)
privacy = NestedOn('Action', dump_only=True)
@ -269,16 +352,21 @@ class SolidStateDrive(DataStorage):
class Motherboard(Component):
__doc__ = m.Motherboard.__doc__
slots = Integer(validate=Range(0, 20),
description=m.Motherboard.slots.comment)
slots = Integer(validate=Range(0, 20), description=m.Motherboard.slots.comment)
usb = Integer(validate=Range(0, 20), description=m.Motherboard.usb.comment)
firewire = Integer(validate=Range(0, 20), description=m.Motherboard.firewire.comment)
firewire = Integer(
validate=Range(0, 20), description=m.Motherboard.firewire.comment
)
serial = Integer(validate=Range(0, 20), description=m.Motherboard.serial.comment)
pcmcia = Integer(validate=Range(0, 20), description=m.Motherboard.pcmcia.comment)
bios_date = Date(validate=Range(datetime.date(year=1980, month=1, day=1),
datetime.date(year=2030, month=1, day=1)),
data_key='biosDate',
description=m.Motherboard.bios_date)
bios_date = Date(
validate=Range(
datetime.date(year=1980, month=1, day=1),
datetime.date(year=2030, month=1, day=1),
),
data_key='biosDate',
description=m.Motherboard.bios_date,
)
ram_slots = Integer(validate=Range(1), data_key='ramSlots')
ram_max_size = Integer(validate=Range(1), data_key='ramMaxSize')
@ -290,22 +378,32 @@ class NetworkAdapter(NetworkMixin, Component):
class Processor(Component):
__doc__ = m.Processor.__doc__
speed = Float(validate=Range(min=0.1, max=15),
unit=UnitCodes.ghz,
description=m.Processor.speed.comment)
cores = Integer(validate=Range(min=1, max=10), description=m.Processor.cores.comment)
threads = Integer(validate=Range(min=1, max=20), description=m.Processor.threads.comment)
address = Integer(validate=OneOf({8, 16, 32, 64, 128, 256}),
description=m.Processor.address.comment)
speed = Float(
validate=Range(min=0.1, max=15),
unit=UnitCodes.ghz,
description=m.Processor.speed.comment,
)
cores = Integer(
validate=Range(min=1, max=10), description=m.Processor.cores.comment
)
threads = Integer(
validate=Range(min=1, max=20), description=m.Processor.threads.comment
)
address = Integer(
validate=OneOf({8, 16, 32, 64, 128, 256}),
description=m.Processor.address.comment,
)
abi = SanitizedStr(lower=True, description=m.Processor.abi.comment)
class RamModule(Component):
__doc__ = m.RamModule.__doc__
size = Integer(validate=Range(min=128, max=17000),
unit=UnitCodes.mbyte,
description=m.RamModule.size.comment)
size = Integer(
validate=Range(min=128, max=17000),
unit=UnitCodes.mbyte,
description=m.RamModule.size.comment,
)
speed = Integer(validate=Range(min=100, max=10000), unit=UnitCodes.mhz)
interface = EnumField(enums.RamInterface)
format = EnumField(enums.RamFormat)
@ -323,7 +421,9 @@ class Battery(Component):
__doc__ = m.Battery.__doc__
wireless = Boolean(description=m.Battery.wireless.comment)
technology = EnumField(enums.BatteryTechnology, description=m.Battery.technology.comment)
technology = EnumField(
enums.BatteryTechnology, description=m.Battery.technology.comment
)
size = Integer(required=True, description=m.Battery.size.comment)
@ -393,12 +493,18 @@ class WirelessAccessPoint(Networking):
class Printer(Device):
__doc__ = m.Printer.__doc__
wireless = Boolean(required=True, missing=False, description=m.Printer.wireless.comment)
scanning = Boolean(required=True, missing=False, description=m.Printer.scanning.comment)
technology = EnumField(enums.PrinterTechnology,
required=True,
description=m.Printer.technology.comment)
monochrome = Boolean(required=True, missing=True, description=m.Printer.monochrome.comment)
wireless = Boolean(
required=True, missing=False, description=m.Printer.wireless.comment
)
scanning = Boolean(
required=True, missing=False, description=m.Printer.scanning.comment
)
technology = EnumField(
enums.PrinterTechnology, required=True, description=m.Printer.technology.comment
)
monochrome = Boolean(
required=True, missing=True, description=m.Printer.monochrome.comment
)
class LabelPrinter(Printer):

136
ereuse_devicehub/resources/device/sync.py
View File

@ -13,10 +13,14 @@ from teal.marshmallow import ValidationError
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.action.models import Remove
from ereuse_devicehub.resources.device.models import Component, Computer, Device, DataStorage
from ereuse_devicehub.resources.device.models import (
Component,
Computer,
DataStorage,
Device,
)
from ereuse_devicehub.resources.tag.model import Tag
DEVICES_ALLOW_DUPLICITY = [
'RamModule',
'Display',
@ -26,12 +30,13 @@ DEVICES_ALLOW_DUPLICITY = [
'GraphicCard',
]
class Sync:
"""Synchronizes the device and components with the database."""
def run(self,
device: Device,
components: Iterable[Component] or None) -> (Device, OrderedSet):
def run(
self, device: Device, components: Iterable[Component] or None
) -> (Device, OrderedSet):
"""Synchronizes the device and components with the database.
Identifies if the device and components exist in the database
@ -72,9 +77,9 @@ class Sync:
blacklist = set() # type: Set[int]
not_new_components = set()
for component in components:
db_component, is_new = self.execute_register_component(component,
blacklist,
parent=db_device)
db_component, is_new = self.execute_register_component(
component, blacklist, parent=db_device
)
db_components.add(db_component)
if not is_new:
not_new_components.add(db_component)
@ -83,10 +88,9 @@ class Sync:
db_device.components = db_components
return db_device, actions
def execute_register_component(self,
component: Component,
blacklist: Set[int],
parent: Computer):
def execute_register_component(
self, component: Component, blacklist: Set[int], parent: Computer
):
"""Synchronizes one component to the DB.
This method is a specialization of :meth:`.execute_register`
@ -118,9 +122,12 @@ class Sync:
# if not, then continue with the traditional behaviour
try:
if component.hid:
db_component = Device.query.filter_by(hid=component.hid, owner_id=g.user.id).one()
assert isinstance(db_component, Device), \
'{} must be a component'.format(db_component)
db_component = Device.query.filter_by(
hid=component.hid, owner_id=g.user.id
).one()
assert isinstance(
db_component, Device
), '{} must be a component'.format(db_component)
else:
# Is there a component similar to ours?
db_component = component.similar_one(parent, blacklist)
@ -166,15 +173,35 @@ class Sync:
:return: The synced device from the db with the tags linked.
"""
assert inspect(device).transient, 'Device cannot be already synced from DB'
assert all(inspect(tag).transient for tag in device.tags), 'Tags cannot be synced from DB'
assert all(
inspect(tag).transient for tag in device.tags
), 'Tags cannot be synced from DB'
db_device = None
if device.hid:
if isinstance(device, Computer):
# first search by uuid
if device.system_uuid:
with suppress(ResourceNotFound):
db_device = Computer.query.filter_by(
system_uuid=device.system_uuid, owner_id=g.user.id, active=True
).one()
# if no there are any Computer by uuid search by hid
if not db_device and device.hid:
with suppress(ResourceNotFound):
db_device = Device.query.filter_by(
hid=device.hid, owner_id=g.user.id, active=True
).one()
elif device.hid:
with suppress(ResourceNotFound):
db_device = Device.query.filter_by(hid=device.hid, owner_id=g.user.id, active=True).one()
db_device = Device.query.filter_by(
hid=device.hid, owner_id=g.user.id, active=True
).one()
if db_device and db_device.allocated:
raise ResourceNotFound('device is actually allocated {}'.format(device))
try:
tags = {Tag.from_an_id(tag.id).one() for tag in device.tags} # type: Set[Tag]
tags = {
Tag.from_an_id(tag.id).one() for tag in device.tags
} # type: Set[Tag]
except ResourceNotFound:
raise ResourceNotFound('tag you are linking to device {}'.format(device))
linked_tags = {tag for tag in tags if tag.device_id} # type: Set[Tag]
@ -182,16 +209,22 @@ class Sync:
sample_tag = next(iter(linked_tags))
for tag in linked_tags:
if tag.device_id != sample_tag.device_id:
raise MismatchBetweenTags(tag, sample_tag) # Tags linked to different devices
raise MismatchBetweenTags(
tag, sample_tag
) # Tags linked to different devices
if db_device: # Device from hid
if sample_tag.device_id != db_device.id: # Device from hid != device from tags
if (
sample_tag.device_id != db_device.id
): # Device from hid != device from tags
raise MismatchBetweenTagsAndHid(db_device.id, db_device.hid)
else: # There was no device from hid
if sample_tag.device.physical_properties != device.physical_properties:
# Incoming physical props of device != props from tag's device
# which means that the devices are not the same
raise MismatchBetweenProperties(sample_tag.device.physical_properties,
device.physical_properties)
raise MismatchBetweenProperties(
sample_tag.device.physical_properties,
device.physical_properties,
)
db_device = sample_tag.device
if db_device: # Device from hid or tags
self.merge(device, db_device)
@ -199,17 +232,21 @@ class Sync:
device.tags.clear() # We don't want to add the transient dummy tags
db.session.add(device)
db_device = device
db_device.tags |= tags # Union of tags the device had plus the (potentially) new ones
db_device.tags |= (
tags # Union of tags the device had plus the (potentially) new ones
)
try:
db.session.flush()
except IntegrityError as e:
# Manage 'one tag per organization' unique constraint
if 'One tag per organization' in e.args[0]:
# todo test for this
id = int(e.args[0][135:e.args[0].index(',', 135)])
raise ValidationError('The device is already linked to tag {} '
'from the same organization.'.format(id),
field_names=['device.tags'])
id = int(e.args[0][135 : e.args[0].index(',', 135)])
raise ValidationError(
'The device is already linked to tag {} '
'from the same organization.'.format(id),
field_names=['device.tags'],
)
else:
raise
assert db_device is not None
@ -222,15 +259,21 @@ class Sync:
This method mutates db_device.
"""
if db_device.owner_id != g.user.id:
return
return
for field_name, value in device.physical_properties.items():
if value is not None:
setattr(db_device, field_name, value)
# if device.system_uuid and db_device.system_uuid and device.system_uuid != db_device.system_uuid:
# TODO @cayop send error to sentry.io
# there are 2 computers duplicate get db_device for hid
if hasattr(device, 'system_uuid') and device.system_uuid:
db_device.system_uuid = device.system_uuid
@staticmethod
def add_remove(device: Computer,
components: Set[Component]) -> OrderedSet:
def add_remove(device: Computer, components: Set[Component]) -> OrderedSet:
"""Generates the Add and Remove actions (but doesn't add them to
session).
@ -251,9 +294,13 @@ class Sync:
if adding:
# For the components we are adding, let's remove them from their old parents
def g_parent(component: Component) -> Device:
return component.parent or Device(id=0) # Computer with id 0 is our Identity
return component.parent or Device(
id=0
) # Computer with id 0 is our Identity
for parent, _components in groupby(sorted(adding, key=g_parent), key=g_parent):
for parent, _components in groupby(
sorted(adding, key=g_parent), key=g_parent
):
set_components = OrderedSet(_components)
check_owners = (x.owner_id == g.user.id for x in set_components)
# Is not Computer Identity and all components have the correct owner
@ -263,21 +310,18 @@ class Sync:
class MismatchBetweenTags(ValidationError):
def __init__(self,
tag: Tag,
other_tag: Tag,
field_names={'device.tags'}):
message = '{!r} and {!r} are linked to different devices.'.format(tag, other_tag)
def __init__(self, tag: Tag, other_tag: Tag, field_names={'device.tags'}):
message = '{!r} and {!r} are linked to different devices.'.format(
tag, other_tag
)
super().__init__(message, field_names)
class MismatchBetweenTagsAndHid(ValidationError):
def __init__(self,
device_id: int,
hid: str,
field_names={'device.hid'}):
message = 'Tags are linked to device {} but hid refers to device {}.'.format(device_id,
hid)
def __init__(self, device_id: int, hid: str, field_names={'device.hid'}):
message = 'Tags are linked to device {} but hid refers to device {}.'.format(
device_id, hid
)
super().__init__(message, field_names)
@ -285,6 +329,8 @@ class MismatchBetweenProperties(ValidationError):
def __init__(self, props1, props2, field_names={'device'}):
message = 'The device from the tag and the passed-in differ the following way:'
message += '\n'.join(
difflib.ndiff(yaml.dump(props1).splitlines(), yaml.dump(props2).splitlines())
difflib.ndiff(
yaml.dump(props1).splitlines(), yaml.dump(props2).splitlines()
)
)
super().__init__(message, field_names)

4
ereuse_devicehub/templates/inventory/snapshots_list.html
View File

@ -29,6 +29,7 @@
<th scope="col">Snapshot id</th>
<th scope="col">Version</th>
<th scope="col">DHID</th>
<th scope="col">System UUID</th>
<th scope="col">Status</th>
<th scope="col" data-type="date" data-format="DD-MM-YYYY">Time</th>
</tr>
@ -60,6 +61,9 @@
</a>
{% endif %}
</td>
<td>
{{ snap.system_uuid }}
</td>
<td>
{{ snap.status }}
</td>

36
examples/extract_uuid.py Normal file
View File

@ -0,0 +1,36 @@
import json
import sys
def get_old_smbios_version(snapshot):
capabilities = snapshot.get('debug', {}).get('lshw', {}).get('capabilities', {})
for x in capabilities.values():
if "SMBIOS version" in x:
e = x.split("SMBIOS version ")[1].split(".")
if int(e[0]) < 3 and int(e[1]) < 6:
return True
return False
def get_uuid(snapshot):
return (
snapshot.get('debug', {}).get('lshw', {}).get('configuration', {}).get('uuid')
)
def main():
_file = sys.argv[1]
with open(_file) as file_snapshot:
snapshot = json.loads(file_snapshot.read())
if get_old_smbios_version(snapshot):
return
system_uuid = get_uuid(snapshot)
if system_uuid:
print("{};{}".format(system_uuid, snapshot['uuid']))
if __name__ == '__main__':
main()

1
examples/extract_uuids.sh Normal file
View File

@ -0,0 +1 @@
for i in `ls ../snapshots/*/*.json`; do python examples/extract_uuid.py $i; done > system_uuids.csv

1
tests/files/system_uuid1.json Normal file
View File

File diff suppressed because one or more lines are too long

3416
tests/files/system_uuid2.json Normal file
View File

File diff suppressed because one or more lines are too long

172
tests/files/system_uuid3.json Normal file
View File

@ -0,0 +1,172 @@
{
"components": [
{
"size": 10.030411318500475,
"technology": "LCD",
"resolutionWidth": 1024,
"model": "AUO LCD Monitor",
"actions": [],
"type": "Display",
"refreshRate": 60,
"productionDate": "2009-01-04T00:00:00",
"manufacturer": "AUO \"AUO\"",
"serialNumber": null,
"resolutionHeight": 600
},
{
"generation": null,
"actions": [
{
"rate": 164.4981,
"type": "BenchmarkProcessorSysbench",
"elapsed": 165
},
{
"rate": 6650.48,
"type": "BenchmarkProcessor",
"elapsed": 0
}
],
"speed": 1.0,
"cores": 1,
"model": "Intel Atom CPU N450 @ 1.66GHz",
"address": 64,
"type": "Processor",
"threads": 2,
"manufacturer": "Intel Corp.",
"serialNumber": null,
"brand": "Atom"
},
{
"memory": null,
"model": "Atom Processor D4xx/D5xx/N4xx/N5xx Integrated Graphics Controller",
"actions": [],
"type": "GraphicCard",
"manufacturer": "Intel Corporation",
"serialNumber": null
},
{
"type": "SoundCard",
"actions": [],
"manufacturer": "Intel Corporation",
"serialNumber": null,
"model": "NM10/ICH7 Family High Definition Audio Controller"
},
{
"type": "SoundCard",
"actions": [],
"manufacturer": "XPA970VW0",
"serialNumber": null,
"model": "1.3M WebCam"
},
{
"size": 1024.0,
"actions": [],
"format": "SODIMM",
"model": "48594D503131325336344350362D53362020",
"interface": "DDR2",
"type": "RamModule",
"manufacturer": "Hynix Semiconductor",
"serialNumber": "4F43487B",
"speed": 667.0
},
{
"size": 160041.88569599998,
"variant": "1A01",
"actions": [
{
"type": "EraseBasic",
"steps": [
{
"type": "StepRandom",
"endTime": "2019-10-23T08:35:31.400587+00:00",
"severity": "Info",
"startTime": "2019-10-23T07:49:54.410830+00:00"
}
],
"endTime": "2019-10-23T08:35:31.400988+00:00",
"severity": "Info",
"startTime": "2019-10-23T07:49:54.410193+00:00"
},
{
"elapsed": 22,
"writeSpeed": 17.3,
"readSpeed": 41.6,
"type": "BenchmarkDataStorage"
},
{
"status": "Completed without error",
"reallocatedSectorCount": 0,
"currentPendingSectorCount": 0,
"assessment": true,
"severity": "Info",
"offlineUncorrectable": 0,
"lifetime": 4692,
"type": "TestDataStorage",
"length": "Short",
"elapsed": 118,
"powerCycleCount": 5293
}
],
"model": "WDC WD1600BEVT-2",
"interface": "ATA",
"type": "HardDrive",
"manufacturer": "Western Digital",
"serialNumber": "WD-WX11A80W7430"
},
{
"variant": "c1",
"actions": [],
"speed": 100.0,
"model": "AR8152 v1.1 Fast Ethernet",
"wireless": false,
"type": "NetworkAdapter",
"serialNumber": "88:ae:1d:a6:f3:d0",
"manufacturer": "Qualcomm Atheros"
},
{
"ramMaxSize": 4,
"slots": 1,
"model": "AOHAPPY",
"pcmcia": 0,
"type": "Motherboard",
"version": "V3.05(DDR2)",
"ramSlots": 2,
"serialNumber": "Base Board Serial Number",
"manufacturer": "Acer",
"serial": 1,
"actions": [],
"biosDate": "2010-08-12T00:00:00",
"firewire": 0,
"usb": 5
}
],
"software": "Workbench",
"device": {
"sku": null,
"chassis": "Netbook",
"actions": [
{
"type": "StressTest",
"elapsed": 60,
"severity": "Info"
},
{
"rate": 19.2726,
"type": "BenchmarkRamSysbench",
"elapsed": 19
}
],
"model": "AOHAPPY",
"type": "Laptop",
"version": "V3.05",
"manufacturer": "Acer",
"serialNumber": "LUSEA0D010038879A01601"
},
"uuid": "0973fda0-589a-11eb-ae93-0242ac130002",
"type": "Snapshot",
"version": "11.0b9",
"endTime": "2019-10-23T07:43:13.625104+00:00",
"elapsed": 3138,
"closed": true
}

189
tests/files/system_uuid4.json Normal file
View File

@ -0,0 +1,189 @@
{
"components": [
{
"size": 10.030411318500475,
"technology": "LCD",
"resolutionWidth": 1024,
"model": "AUO LCD Monitor",
"actions": [],
"type": "Display",
"refreshRate": 60,
"productionDate": "2009-01-04T00:00:00",
"manufacturer": "AUO \"AUO\"",
"serialNumber": null,
"resolutionHeight": 600
},
{
"generation": null,
"actions": [
{
"rate": 164.4981,
"type": "BenchmarkProcessorSysbench",
"elapsed": 165
},
{
"rate": 6650.48,
"type": "BenchmarkProcessor",
"elapsed": 0
}
],
"speed": 1.0,
"cores": 1,
"model": "Intel Atom CPU N450 @ 1.66GHz",
"address": 64,
"type": "Processor",
"threads": 2,
"manufacturer": "Intel Corp.",
"serialNumber": null,
"brand": "Atom"
},
{
"memory": null,
"model": "Atom Processor D4xx/D5xx/N4xx/N5xx Integrated Graphics Controller",
"actions": [],
"type": "GraphicCard",
"manufacturer": "Intel Corporation",
"serialNumber": null
},
{
"type": "SoundCard",
"actions": [],
"manufacturer": "Intel Corporation",
"serialNumber": null,
"model": "NM10/ICH7 Family High Definition Audio Controller"
},
{
"type": "SoundCard",
"actions": [],
"manufacturer": "XPA970VW0",
"serialNumber": null,
"model": "1.3M WebCam"
},
{
"size": 1024.0,
"actions": [],
"format": "SODIMM",
"model": "48594D503131325336344350362D53362020",
"interface": "DDR2",
"type": "RamModule",
"manufacturer": "Hynix Semiconductor",
"serialNumber": "4F43487B",
"speed": 667.0
},
{
"size": 160041.88569599998,
"variant": "1A01",
"actions": [
{
"type": "EraseBasic",
"steps": [
{
"type": "StepRandom",
"endTime": "2019-10-23T08:35:31.400587+00:00",
"severity": "Info",
"startTime": "2019-10-23T07:49:54.410830+00:00"
}
],
"endTime": "2019-10-23T08:35:31.400988+00:00",
"severity": "Info",
"startTime": "2019-10-23T07:49:54.410193+00:00"
},
{
"elapsed": 22,
"writeSpeed": 17.3,
"readSpeed": 41.6,
"type": "BenchmarkDataStorage"
},
{
"status": "Completed without error",
"reallocatedSectorCount": 0,
"currentPendingSectorCount": 0,
"assessment": true,
"severity": "Info",
"offlineUncorrectable": 0,
"lifetime": 4692,
"type": "TestDataStorage",
"length": "Short",
"elapsed": 118,
"powerCycleCount": 5293
}
],
"model": "WDC WD1600BEVT-2",
"interface": "ATA",
"type": "HardDrive",
"manufacturer": "Western Digital",
"serialNumber": "WD-WX11A80W7430"
},
{
"variant": "c1",
"actions": [],
"speed": 100.0,
"model": "AR8152 v1.1 Fast Ethernet",
"wireless": false,
"type": "NetworkAdapter",
"serialNumber": "88:ae:1d:a6:f3:d0",
"manufacturer": "Qualcomm Atheros"
},
{
"ramMaxSize": 4,
"slots": 1,
"model": "AOHAPPY",
"pcmcia": 0,
"type": "Motherboard",
"version": "V3.05(DDR2)",
"ramSlots": 2,
"serialNumber": "Base Board Serial Number",
"manufacturer": "Acer",
"serial": 1,
"actions": [],
"biosDate": "2010-08-12T00:00:00",
"firewire": 0,
"usb": 5
}
],
"software": "Workbench",
"device": {
"sku": null,
"chassis": "Netbook",
"actions": [
{
"type": "StressTest",
"elapsed": 60,
"severity": "Info"
},
{
"rate": 19.2726,
"type": "BenchmarkRamSysbench",
"elapsed": 19
}
],
"model": "AOHAPPY",
"type": "Laptop",
"version": "V3.05",
"manufacturer": "Acer",
"serialNumber": "LUSEA0D010038879A01601"
},
"debug": {
"lshw": {
"capabilities": {
"dmi-2.5": "DMI version 2.5",
"smbios-2.5": "SMBIOS version 2.5",
"smp": "Symmetric Multi-Processing",
"vsyscall32": "32-bit processes"
},
"configuration": {
"boot": "normal",
"chassis": "notebook",
"family": "Intel_Mobile",
"sku": "NetTopSku",
"uuid": "364ee69c-9c82-9cb1-2111-88ae1da6f3d0"
}
}
},
"uuid": "0973fda0-589a-11eb-ae93-0242ac130002",
"type": "Snapshot",
"version": "11.0b9",
"endTime": "2019-10-23T07:43:13.625104+00:00",
"elapsed": 3138,
"closed": true
}

1
tests/test_device.py
View File

@ -130,7 +130,6 @@ def test_physical_properties():
'model': 'foo',
'receiver_id': None,
'serial_number': 'foo-bar',
'uuid': None,
'transfer_state': TransferState.Initial
}

16
tests/test_snapshot.py
View File

@ -190,10 +190,12 @@ def test_snapshot_power_on_hours(user: UserClient):
)
errors = SnapshotsLog.query.filter().all()
snap_log = errors[0]
assert str(snap_log.snapshot.uuid) == snap['uuid']
assert len(errors) == 1
assert errors[0].description == 'Ok'
snap_log = errors[1]
assert len(errors) == 2
assert str(errors[0].snapshot_uuid) == snap['uuid']
assert str(errors[1].snapshot.uuid) == snap['uuid']
assert errors[0].description == 'There is not uuid'
assert errors[1].description == 'Ok'
@pytest.mark.mvp
@ -784,11 +786,13 @@ def test_backup_snapshot_with_errors(app: Devicehub, user: UserClient):
response = user.post(res=Snapshot, data=json_encode(snapshot_no_hid))
errors = SnapshotsLog.query.filter().all()
snap_log = errors[0]
snap_log = errors[1]
assert snap_log.description == "'BenchmarkProcessorr'"
assert errors[0].description == 'There is not uuid'
assert snap_log.version == "11.0b9"
assert str(snap_log.snapshot_uuid) == '9a3e7485-fdd0-47ce-bcc7-65c55226b598'
assert len(errors) == 1
assert str(errors[0].snapshot_uuid) == '9a3e7485-fdd0-47ce-bcc7-65c55226b598'
assert len(errors) == 2
files = [x for x in os.listdir(path_dir_base) if uuid in x]
if files:

610
tests/test_system_uuid.py Normal file
View File

@ -0,0 +1,610 @@
import json
from io import BytesIO
import pytest
from flask_wtf.csrf import generate_csrf
from ereuse_devicehub.client import UserClient, UserClientFlask
from ereuse_devicehub.resources.action.models import Snapshot
from ereuse_devicehub.resources.device.models import Computer
from tests import conftest
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_form(user3: UserClientFlask):
uri = '/inventory/upload-snapshot/'
file_name = 'system_uuid1.json'
user3.get(uri)
snapshot = conftest.file_json(file_name)
b_snapshot = bytes(json.dumps(snapshot), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
db_snapthot = Snapshot.query.one()
device = db_snapthot.device
assert device.hid == 'laptop-toshiba-satellite_l655-2b335208q-00:26:6c:ae:ee:78'
assert str(device.system_uuid) == 'f0dc6a7f-c23f-e011-b5d0-00266caeee78'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_api(user: UserClient):
file_name = 'system_uuid1.json'
snapshot_11 = conftest.file_json(file_name)
user.post(snapshot_11, res=Snapshot)
db_snapthot = Snapshot.query.one()
device = db_snapthot.device
assert device.hid == 'laptop-toshiba-satellite_l655-2b335208q-00:26:6c:ae:ee:78'
assert str(device.system_uuid) == 'f0dc6a7f-c23f-e011-b5d0-00266caeee78'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wbLite_form(user3: UserClientFlask):
uri = '/inventory/upload-snapshot/'
file_name = 'system_uuid2.json'
user3.get(uri)
snapshot = conftest.file_json(file_name)
b_snapshot = bytes(json.dumps(snapshot), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
db_snapthot = Snapshot.query.one()
device = db_snapthot.device
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wbLite_api(user: UserClient):
snapshot_lite = conftest.file_json('system_uuid2.json')
user.post(snapshot_lite, uri="/api/inventory/")
db_snapthot = Snapshot.query.one()
device = db_snapthot.device
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_to_wb11_with_uuid_api(user: UserClient):
# insert computer with wb11 with hid and without uuid, (old version)
snapshot_11 = conftest.file_json('system_uuid3.json')
user.post(snapshot_11, res=Snapshot)
db_snapthot = Snapshot.query.one()
device = db_snapthot.device
assert Computer.query.count() == 1
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert device.system_uuid is None
# insert the same computer with wb11 with hid and with uuid, (new version)
snapshot_lite = conftest.file_json('system_uuid2.json')
snapshot_11['debug'] = {'lshw': snapshot_lite['data']['lshw']}
snapshot_11['uuid'] = '0973fda0-589a-11eb-ae93-0242ac130003'
user.post(snapshot_11, res=Snapshot)
assert (
snapshot_11['debug']['lshw']['configuration']['uuid']
== '364ee69c-9c82-9cb1-2111-88ae1da6f3d0'
)
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_with_uuid_to_wb11_api(user: UserClient):
# insert computer with wb11 with hid and without uuid, (old version)
snapshot_11 = conftest.file_json('system_uuid3.json')
snapshot_lite = conftest.file_json('system_uuid2.json')
snapshot_11['debug'] = {'lshw': snapshot_lite['data']['lshw']}
snapshot_11['uuid'] = '0973fda0-589a-11eb-ae93-0242ac130003'
user.post(snapshot_11, res=Snapshot)
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
# insert the same computer with wb11 with hid and with uuid, (new version)
snapshot_11 = conftest.file_json('system_uuid3.json')
assert 'debug' not in snapshot_11
user.post(snapshot_11, res=Snapshot)
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_with_uuid_to_wb11_without_hid_api(user: UserClient):
# insert computer with wb11 with hid and without uuid, (old version)
snapshot_11 = conftest.file_json('system_uuid3.json')
snapshot_lite = conftest.file_json('system_uuid2.json')
snapshot_11['debug'] = {'lshw': snapshot_lite['data']['lshw']}
snapshot_11['uuid'] = '0973fda0-589a-11eb-ae93-0242ac130003'
user.post(snapshot_11, res=Snapshot)
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
# insert the same computer with wb11 with hid and with uuid, (new version)
snapshot_11 = conftest.file_json('system_uuid3.json')
components = [x for x in snapshot_11['components'] if x['type'] != 'NetworkAdapter']
snapshot_11['components'] = components
snapshot_11['debug'] = {'lshw': snapshot_lite['data']['lshw']}
user.post(snapshot_11, res=Snapshot)
assert Computer.query.count() == 1
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_to_wb11_with_uuid_form(user3: UserClientFlask):
# insert computer with wb11 with hid and without uuid, (old version)
uri = '/inventory/upload-snapshot/'
user3.get(uri)
file_name = 'system_uuid3.json'
snapshot = conftest.file_json(file_name)
b_snapshot = bytes(json.dumps(snapshot), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
db_snapthot = Snapshot.query.one()
device = db_snapthot.device
assert Computer.query.count() == 1
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert device.system_uuid is None
# insert the same computer with wb11 with hid and with uuid, (new version)
snapshot_lite = conftest.file_json('system_uuid2.json')
snapshot['debug'] = {'lshw': snapshot_lite['data']['lshw']}
snapshot['uuid'] = '0973fda0-589a-11eb-ae93-0242ac130003'
b_snapshot = bytes(json.dumps(snapshot), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_with_uuid_to_wb11_form(user3: UserClientFlask):
# insert computer with wb11 with hid and without uuid, (old version)
uri = '/inventory/upload-snapshot/'
user3.get(uri)
file_name = 'system_uuid3.json'
snapshot_lite = conftest.file_json('system_uuid2.json')
snapshot = conftest.file_json(file_name)
snapshot['debug'] = {'lshw': snapshot_lite['data']['lshw']}
snapshot['uuid'] = '0973fda0-589a-11eb-ae93-0242ac130003'
b_snapshot = bytes(json.dumps(snapshot), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
# insert the same computer with wb11 with hid and with uuid, (new version)
snapshot = conftest.file_json('system_uuid3.json')
assert 'debug' not in snapshot
b_snapshot = bytes(json.dumps(snapshot), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_with_uuid_to_wb11_without_hid_form(user3: UserClientFlask):
uri = '/inventory/upload-snapshot/'
user3.get(uri)
# insert computer with wb11 with hid and without uuid, (old version)
snapshot_lite = conftest.file_json('system_uuid2.json')
file_name = 'system_uuid3.json'
snapshot_11 = conftest.file_json(file_name)
snapshot_11['debug'] = {'lshw': snapshot_lite['data']['lshw']}
snapshot_11['uuid'] = '0973fda0-589a-11eb-ae93-0242ac130003'
b_snapshot = bytes(json.dumps(snapshot_11), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
# insert the same computer with wb11 with hid and with uuid, (new version)
snapshot_11 = conftest.file_json('system_uuid3.json')
components = [x for x in snapshot_11['components'] if x['type'] != 'NetworkAdapter']
snapshot_11['components'] = components
snapshot_11['debug'] = {'lshw': snapshot_lite['data']['lshw']}
b_snapshot = bytes(json.dumps(snapshot_11), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_to_wblite_api(user: UserClient):
# insert computer with wb11 with hid and without uuid, (old version)
snapshot_11 = conftest.file_json('system_uuid3.json')
user.post(snapshot_11, res=Snapshot)
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert device.system_uuid is None
snapshot_lite = conftest.file_json('system_uuid2.json')
user.post(snapshot_lite, uri="/api/inventory/")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wblite_to_wb11_api(user: UserClient):
snapshot_lite = conftest.file_json('system_uuid2.json')
user.post(snapshot_lite, uri="/api/inventory/")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
snapshot_11 = conftest.file_json('system_uuid3.json')
user.post(snapshot_11, res=Snapshot)
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_to_wblite_form(user3: UserClientFlask):
uri = '/inventory/upload-snapshot/'
user3.get(uri)
file_name = 'system_uuid3.json'
snapshot_11 = conftest.file_json(file_name)
b_snapshot = bytes(json.dumps(snapshot_11), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert device.system_uuid is None
file_name = 'system_uuid2.json'
snapshot_lite = conftest.file_json(file_name)
b_snapshot = bytes(json.dumps(snapshot_lite), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wblite_to_wb11_form(user3: UserClientFlask):
uri = '/inventory/upload-snapshot/'
user3.get(uri)
file_name = 'system_uuid2.json'
snapshot_lite = conftest.file_json(file_name)
b_snapshot = bytes(json.dumps(snapshot_lite), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
file_name = 'system_uuid3.json'
snapshot_11 = conftest.file_json(file_name)
b_snapshot = bytes(json.dumps(snapshot_11), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wblite_to_wblite_api(user: UserClient):
snapshot_lite = conftest.file_json('system_uuid2.json')
user.post(snapshot_lite, uri="/api/inventory/")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
snapshot_lite = conftest.file_json('system_uuid2.json')
snapshot_lite['uuid'] = '0973fda0-589a-11eb-ae93-0242ac130003'
user.post(snapshot_lite, uri="/api/inventory/")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wblite_to_wblite_form(user3: UserClientFlask):
uri = '/inventory/upload-snapshot/'
user3.get(uri)
file_name = 'system_uuid2.json'
snapshot_lite = conftest.file_json(file_name)
b_snapshot = bytes(json.dumps(snapshot_lite), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
file_name = 'system_uuid2.json'
snapshot_lite = conftest.file_json(file_name)
snapshot_lite['uuid'] = '0973fda0-589a-11eb-ae93-0242ac130003'
b_snapshot = bytes(json.dumps(snapshot_lite), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_to_wb11_duplicity_api(user: UserClient):
# insert computer with wb11 with hid and without uuid, (old version)
snapshot_11 = conftest.file_json('system_uuid3.json')
user.post(snapshot_11, res=Snapshot)
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert device.system_uuid is None
snapshot_11 = conftest.file_json('system_uuid3.json')
snapshot_11['uuid'] = '0973fda0-589a-11eb-ae93-0242ac130003'
components = [x for x in snapshot_11['components'] if x['type'] != 'NetworkAdapter']
snapshot_11['components'] = components
user.post(snapshot_11, res=Snapshot)
assert Computer.query.count() == 2
for c in Computer.query.all():
assert 'laptop-acer-aohappy-lusea0d010038879a01601' in c.hid
assert c.system_uuid is None
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_to_wb11_duplicity_form(user3: UserClientFlask):
uri = '/inventory/upload-snapshot/'
user3.get(uri)
file_name = 'system_uuid3.json'
snapshot_11 = conftest.file_json(file_name)
b_snapshot = bytes(json.dumps(snapshot_11), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert device.system_uuid is None
snapshot_11 = conftest.file_json('system_uuid3.json')
snapshot_11['uuid'] = '0973fda0-589a-11eb-ae93-0242ac130003'
components = [x for x in snapshot_11['components'] if x['type'] != 'NetworkAdapter']
snapshot_11['components'] = components
b_snapshot = bytes(json.dumps(snapshot_11), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 2
for c in Computer.query.all():
assert 'laptop-acer-aohappy-lusea0d010038879a01601' in c.hid
assert c.system_uuid is None
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_smbios_2_5_api(user: UserClient):
# insert computer with wb11 with hid and without uuid, (old version)
snapshot_11 = conftest.file_json('system_uuid4.json')
user.post(snapshot_11, res=Snapshot)
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert device.system_uuid is None
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wb11_smbios_2_5_form(user3: UserClientFlask):
uri = '/inventory/upload-snapshot/'
user3.get(uri)
file_name = 'system_uuid4.json'
snapshot_11 = conftest.file_json(file_name)
b_snapshot = bytes(json.dumps(snapshot_11), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert device.system_uuid is None
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wblite_smbios_2_5_api(user: UserClient):
# insert computer with wb11 with hid and without uuid, (old version)
snapshot_lite = conftest.file_json('system_uuid2.json')
snapshot_lite['data']['lshw']['capabilities']['smbios-3.0'] = 'SMBIOS version 2.5'
user.post(snapshot_lite, uri="/api/inventory/")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_wblite_smbios_2_5_form(user3: UserClientFlask):
uri = '/inventory/upload-snapshot/'
user3.get(uri)
file_name = 'system_uuid2.json'
snapshot_lite = conftest.file_json(file_name)
snapshot_lite['data']['lshw']['capabilities']['smbios-3.0'] = 'SMBIOS version 2.5'
b_snapshot = bytes(json.dumps(snapshot_lite), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data, content_type="multipart/form-data")
assert Computer.query.count() == 1
device = Computer.query.one()
assert device.hid == 'laptop-acer-aohappy-lusea0d010038879a01601-88:ae:1d:a6:f3:d0'
assert str(device.system_uuid) == '9ce64e36-829c-b19c-2111-88ae1da6f3d0'