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

Merge pull request #225 from eReuse/feature/list-snapshots-view-#3113

Browse source 
Feature/list snapshots view #3113
This commit is contained in:
cayop 2022-05-23 09:27:59 +02:00 committed by GitHub
parent db5bbeb796 59379b41a5
commit 7ac6d9989a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 583 additions and 141 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
ereuse_devicehub/api/views.py
View file

@ -11,7 +11,7 @@ from werkzeug.exceptions import Unauthorized
from ereuse_devicehub.auth import Auth
from ereuse_devicehub.db import db
from ereuse_devicehub.parser.models import SnapshotErrors
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.views.snapshot import (
@ -59,6 +59,17 @@ class InventoryView(LoginMixin, SnapshotMixin):
snapshot = self.build()
db.session.add(snapshot)
snap_log = SnapshotsLog(
description='Ok',
snapshot_uuid=snapshot.uuid,
severity=Severity.Info,
sid=snapshot.sid,
version=str(snapshot.version),
snapshot=snapshot,
)
snap_log.save()
db.session().final_flush()
db.session.commit()
self.response = jsonify(
@ -80,8 +91,13 @@ class InventoryView(LoginMixin, SnapshotMixin):
txt = "{}".format(err)
uuid = snapshot_json.get('uuid')
sid = snapshot_json.get('sid')
error = SnapshotErrors(
description=txt, snapshot_uuid=uuid, severity=Severity.Error, sid=sid
version = snapshot_json.get('version')
error = SnapshotsLog(
description=txt,
snapshot_uuid=uuid,
severity=Severity.Error,
sid=sid,
version=str(version),
)
error.save(commit=True)
# raise err

19
ereuse_devicehub/inventory/forms.py
View file

@ -27,7 +27,7 @@ from wtforms import (
from wtforms.fields import FormField
from ereuse_devicehub.db import db
from ereuse_devicehub.parser.models import SnapshotErrors
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
@ -240,6 +240,9 @@ class UploadSnapshotForm(SnapshotMixin, FlaskForm):
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):
self.snapshot_json = schema_lite.load(snapshot_json)
snapshot_json = ParseSnapshotLsHw(self.snapshot_json).snapshot_json
@ -248,13 +251,12 @@ class UploadSnapshotForm(SnapshotMixin, FlaskForm):
snapshot_json = schema.load(snapshot_json)
except ValidationError as err:
txt = "{}".format(err)
uuid = snapshot_json.get('uuid')
sid = snapshot_json.get('sid')
error = SnapshotErrors(
error = SnapshotsLog(
description=txt,
snapshot_uuid=uuid,
severity=Severity.Error,
sid=sid,
version=software_version,
)
error.save(commit=True)
self.result[filename] = 'Error'
@ -263,6 +265,15 @@ class UploadSnapshotForm(SnapshotMixin, FlaskForm):
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'

70
ereuse_devicehub/inventory/views.py
View file

@ -25,6 +25,7 @@ from ereuse_devicehub.inventory.forms import (
UploadSnapshotForm,
)
from ereuse_devicehub.labels.forms import PrintLabelsForm
from ereuse_devicehub.parser.models import SnapshotsLog
from ereuse_devicehub.resources.action.models import Trade
from ereuse_devicehub.resources.device.models import Computer, DataStorage, Device
from ereuse_devicehub.resources.documents.device_row import ActionRow, DeviceRow
@ -512,6 +513,70 @@ class ExportsView(View):
return flask.render_template('inventory/erasure.html', **params)
class SnapshotListView(GenericMixin):
template_name = 'inventory/snapshots_list.html'
def dispatch_request(self):
self.get_context()
self.context['page_title'] = "Snapshots Logs"
self.context['snapshots_log'] = self.get_snapshots_log()
return flask.render_template(self.template_name, **self.context)
def get_snapshots_log(self):
snapshots_log = SnapshotsLog.query.filter(
SnapshotsLog.owner == g.user
).order_by(SnapshotsLog.created.desc())
logs = {}
for snap in snapshots_log:
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(),
'status': snap.get_status(),
'severity': snap.severity,
'created': snap.created,
}
continue
if snap.created > logs[snap.snapshot_uuid]['created']:
logs[snap.snapshot_uuid]['created'] = snap.created
if snap.severity > logs[snap.snapshot_uuid]['severity']:
logs[snap.snapshot_uuid]['severity'] = snap.severity
logs[snap.snapshot_uuid]['status'] = snap.get_status()
result = sorted(logs.values(), key=lambda d: d['created'])
result.reverse()
return result
class SnapshotDetailView(GenericMixin):
template_name = 'inventory/snapshot_detail.html'
def dispatch_request(self, snapshot_uuid):
self.snapshot_uuid = snapshot_uuid
self.get_context()
self.context['page_title'] = "Snapshot Detail"
self.context['snapshots_log'] = self.get_snapshots_log()
self.context['snapshot_uuid'] = snapshot_uuid
self.context['snapshot_sid'] = ''
if self.context['snapshots_log'].count():
self.context['snapshot_sid'] = self.context['snapshots_log'][0].sid
return flask.render_template(self.template_name, **self.context)
def get_snapshots_log(self):
return (
SnapshotsLog.query.filter(SnapshotsLog.owner == g.user)
.filter(SnapshotsLog.snapshot_uuid == self.snapshot_uuid)
.order_by(SnapshotsLog.created.desc())
)
devices.add_url_rule('/action/add/', view_func=NewActionView.as_view('action_add'))
devices.add_url_rule('/action/trade/add/', view_func=NewTradeView.as_view('trade_add'))
devices.add_url_rule(
@ -558,3 +623,8 @@ devices.add_url_rule(
devices.add_url_rule(
'/export/<string:export_id>/', view_func=ExportsView.as_view('export')
)
devices.add_url_rule('/snapshots/', view_func=SnapshotListView.as_view('snapshotslist'))
devices.add_url_rule(
'/snapshots/<string:snapshot_uuid>/',
view_func=SnapshotDetailView.as_view('snapshot_detail'),
)

102
ereuse_devicehub/migrations/versions/926865284103_snapshot_log.py Normal file
View file

@ -0,0 +1,102 @@
"""snapshot_log
Revision ID: 926865284103
Revises: 6f6771813f2e
Create Date: 2022-05-17 17:57:46.651106
"""
import citext
import sqlalchemy as sa
from alembic import context, op
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision = '926865284103'
down_revision = '6f6771813f2e'
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.create_table(
'snapshots_log',
sa.Column(
'updated',
sa.TIMESTAMP(timezone=True),
server_default=sa.text('CURRENT_TIMESTAMP'),
nullable=False,
comment='The last time Devicehub recorded a change for \n this thing.\n ',
),
sa.Column(
'created',
sa.TIMESTAMP(timezone=True),
server_default=sa.text('CURRENT_TIMESTAMP'),
nullable=False,
comment='When Devicehub created this.',
),
sa.Column('id', sa.BigInteger(), nullable=False),
sa.Column('description', citext.CIText(), nullable=True),
sa.Column('version', citext.CIText(), nullable=True),
sa.Column('sid', citext.CIText(), nullable=True),
sa.Column('severity', sa.SmallInteger(), nullable=False),
sa.Column('snapshot_uuid', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('snapshot_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('owner_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.ForeignKeyConstraint(
['snapshot_id'],
[f'{get_inv()}.snapshot.id'],
),
sa.ForeignKeyConstraint(
['owner_id'],
['common.user.id'],
),
sa.PrimaryKeyConstraint('id'),
schema=f'{get_inv()}',
)
op.execute(f"CREATE SEQUENCE {get_inv()}.snapshots_log_seq START 1;")
op.drop_table('snapshot_errors', schema=f'{get_inv()}')
op.execute(f"DROP SEQUENCE {get_inv()}.snapshot_errors_seq;")
def downgrade():
op.drop_table('snapshots_log', schema=f'{get_inv()}')
op.execute(f"DROP SEQUENCE {get_inv()}.snapshots_log_seq;")
op.create_table(
'snapshot_errors',
sa.Column(
'updated',
sa.TIMESTAMP(timezone=True),
server_default=sa.text('CURRENT_TIMESTAMP'),
nullable=False,
comment='The last time Devicehub recorded a change for \n this thing.\n ',
),
sa.Column(
'created',
sa.TIMESTAMP(timezone=True),
server_default=sa.text('CURRENT_TIMESTAMP'),
nullable=False,
comment='When Devicehub created this.',
),
sa.Column('id', sa.BigInteger(), nullable=False),
sa.Column('description', citext.CIText(), nullable=False),
sa.Column('snapshot_uuid', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('severity', sa.SmallInteger(), nullable=False),
sa.Column('sid', citext.CIText(), nullable=True),
sa.Column('owner_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.ForeignKeyConstraint(
['owner_id'],
['common.user.id'],
),
sa.PrimaryKeyConstraint('id'),
schema=f'{get_inv()}',
)
op.execute(f"CREATE SEQUENCE {get_inv()}.snapshot_errors_seq START 1;")

18
ereuse_devicehub/parser/computer.py
View file

@ -14,7 +14,7 @@ from ereuse_utils.nested_lookup import (
)
from ereuse_devicehub.parser import base2, unit, utils
from ereuse_devicehub.parser.models import SnapshotErrors
from ereuse_devicehub.parser.models import SnapshotsLog
from ereuse_devicehub.parser.utils import Dumpeable
from ereuse_devicehub.resources.enums import Severity
@ -422,7 +422,7 @@ class Computer(Device):
self._ram = None
@classmethod
def run(cls, lshw, hwinfo_raw, uuid=None, sid=None):
def run(cls, lshw, hwinfo_raw, uuid=None, sid=None, version=None):
"""
Gets hardware information from the computer and its components,
like serial numbers or model names, and benchmarks them.
@ -447,18 +447,24 @@ class Computer(Device):
txt = "Error: Snapshot: {uuid}, sid: {sid}, type_error: {type}, error: {error}".format(
uuid=uuid, sid=sid, type=err.__class__, error=err
)
cls.errors(txt, uuid=uuid, sid=sid)
cls.errors(txt, uuid=uuid, sid=sid, version=version)
return computer, components
@classmethod
def errors(cls, txt=None, uuid=None, sid=None, severity=Severity.Error):
def errors(
cls, txt=None, uuid=None, sid=None, version=None, severity=Severity.Error
):
if not txt:
return
logger.error(txt)
error = SnapshotErrors(
description=txt, snapshot_uuid=uuid, severity=severity, sid=sid
error = SnapshotsLog(
description=txt,
snapshot_uuid=uuid,
severity=severity,
sid=sid,
version=version,
)
error.save()

25
ereuse_devicehub/parser/models.py
View file

@ -4,25 +4,29 @@ from sqlalchemy import BigInteger, Column, Sequence, SmallInteger
from sqlalchemy.dialects.postgresql import UUID
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.action.models import Snapshot
from ereuse_devicehub.resources.enums import Severity
from ereuse_devicehub.resources.models import Thing
from ereuse_devicehub.resources.user.models import User
class SnapshotErrors(Thing):
"""A Snapshot errors."""
class SnapshotsLog(Thing):
"""A Snapshot log."""
id = Column(BigInteger, Sequence('snapshot_errors_seq'), primary_key=True)
description = Column(CIText(), default='', nullable=False)
sid = Column(CIText(), nullable=True)
id = Column(BigInteger, Sequence('snapshots_log_seq'), primary_key=True)
severity = Column(SmallInteger, default=Severity.Info, nullable=False)
snapshot_uuid = Column(UUID(as_uuid=True), nullable=False)
version = Column(CIText(), default='', nullable=True)
description = Column(CIText(), default='', nullable=True)
sid = Column(CIText(), nullable=True)
snapshot_uuid = Column(UUID(as_uuid=True), nullable=True)
snapshot_id = Column(UUID(as_uuid=True), db.ForeignKey(Snapshot.id), nullable=True)
owner_id = db.Column(
UUID(as_uuid=True),
db.ForeignKey(User.id),
nullable=False,
default=lambda: g.user.id,
)
snapshot = db.relationship(Snapshot, primaryjoin=snapshot_id == Snapshot.id)
owner = db.relationship(User, primaryjoin=owner_id == User.id)
def save(self, commit=False):
@ -30,3 +34,12 @@ class SnapshotErrors(Thing):
if commit:
db.session.commit()
def get_status(self):
return Severity(self.severity)
def get_device(self):
if self.snapshot:
return self.snapshot.device.devicehub_id
return ''

33
ereuse_devicehub/parser/parser.py
View file

@ -8,7 +8,7 @@ from marshmallow.exceptions import ValidationError
from ereuse_devicehub.parser import base2
from ereuse_devicehub.parser.computer import Computer
from ereuse_devicehub.parser.models import SnapshotErrors
from ereuse_devicehub.parser.models import SnapshotsLog
from ereuse_devicehub.resources.action.schemas import Snapshot
from ereuse_devicehub.resources.enums import DataStorageInterface, Severity
@ -320,6 +320,7 @@ class ParseSnapshotLsHw:
self.default = default
self.uuid = snapshot.get("uuid")
self.sid = snapshot.get("sid")
self.version = str(snapshot.get("version"))
self.dmidecode_raw = snapshot["data"]["dmidecode"]
self.smart = snapshot["data"]["smart"]
self.hwinfo_raw = snapshot["data"]["hwinfo"]
@ -362,7 +363,7 @@ class ParseSnapshotLsHw:
def set_basic_datas(self):
try:
pc, self.components_obj = Computer.run(
self.lshw, self.hwinfo_raw, self.uuid, self.sid
self.lshw, self.hwinfo_raw, self.uuid, self.sid, self.version
)
pc = pc.dump()
minimum_hid = None in [pc['manufacturer'], pc['model'], pc['serialNumber']]
@ -417,11 +418,11 @@ class ParseSnapshotLsHw:
size = ram.get("Size")
if not len(size.split(" ")) == 2:
txt = (
"Error: Snapshot: {uuid}, tag: {sid} have this ram Size: {size}".format(
"Error: Snapshot: {uuid}, Sid: {sid} have this ram Size: {size}".format(
uuid=self.uuid, size=size, sid=self.sid
)
)
self.errors(txt)
self.errors(txt, severity=Severity.Warning)
return 128
size, units = size.split(" ")
return base2.Quantity(float(size), units).to('MiB').m
@ -429,10 +430,10 @@ class ParseSnapshotLsHw:
def get_ram_speed(self, ram):
speed = ram.get("Speed", "100")
if not len(speed.split(" ")) == 2:
txt = "Error: Snapshot: {uuid}, tag: {sid} have this ram Speed: {speed}".format(
txt = "Error: Snapshot: {uuid}, Sid: {sid} have this ram Speed: {speed}".format(
uuid=self.uuid, speed=speed, sid=self.sid
)
self.errors(txt)
self.errors(txt, severity=Severity.Warning)
return 100
speed, units = speed.split(" ")
return float(speed)
@ -464,10 +465,10 @@ class ParseSnapshotLsHw:
uuid.UUID(dmi_uuid)
except (ValueError, AttributeError) as err:
self.errors("{}".format(err))
txt = "Error: Snapshot: {uuid} tag: {sid} have this uuid: {device}".format(
txt = "Error: Snapshot: {uuid} sid: {sid} have this uuid: {device}".format(
uuid=self.uuid, device=dmi_uuid, sid=self.sid
)
self.errors(txt)
self.errors(txt, severity=Severity.Warning)
dmi_uuid = None
return dmi_uuid
@ -510,11 +511,11 @@ class ParseSnapshotLsHw:
try:
DataStorageInterface(interface.upper())
except ValueError as err:
txt = "tag: {}, interface {} is not in DataStorageInterface Enum".format(
interface, self.sid
txt = "Sid: {}, interface {} is not in DataStorageInterface Enum".format(
self.sid, interface
)
self.errors("{}".format(err))
self.errors(txt)
self.errors(txt, severity=Severity.Warning)
return "ATA"
def get_data_storage_size(self, x):
@ -546,13 +547,17 @@ class ParseSnapshotLsHw:
return action
def errors(self, txt=None, severity=Severity.Info):
def errors(self, txt=None, severity=Severity.Error):
if not txt:
return self._errors
logger.error(txt)
self._errors.append(txt)
error = SnapshotErrors(
description=txt, snapshot_uuid=self.uuid, severity=severity, sid=self.sid
error = SnapshotsLog(
description=txt,
snapshot_uuid=self.uuid,
severity=severity,
sid=self.sid,
version=self.version,
)
error.save()

4
ereuse_devicehub/resources/action/views/snapshot.py
View file

@ -11,7 +11,6 @@ from marshmallow import ValidationError
from sqlalchemy.util import OrderedSet
from ereuse_devicehub.db import db
from ereuse_devicehub.parser.models import SnapshotErrors
from ereuse_devicehub.resources.action.models import Snapshot
from ereuse_devicehub.resources.device.models import Computer
from ereuse_devicehub.resources.device.sync import Sync
@ -138,9 +137,10 @@ class SnapshotView(SnapshotMixin):
try:
self.snapshot_json = resource_def.schema.load(snapshot_json)
except ValidationError as err:
from ereuse_devicehub.parser.models import SnapshotsLog
txt = "{}".format(err)
uuid = snapshot_json.get('uuid')
error = SnapshotErrors(
error = SnapshotsLog(
description=txt, snapshot_uuid=uuid, severity=Severity.Error
)
error.save(commit=True)

1
ereuse_devicehub/resources/device/models.py
View file

@ -358,7 +358,6 @@ class Device(Thing):
from ereuse_devicehub.resources.device import states
with suppress(LookupError, ValueError):
# import pdb; pdb.set_trace()
return self.last_action_of(*states.Physical.actions())
@property

175
ereuse_devicehub/resources/lot/models.py
View file

@ -10,7 +10,7 @@ from sqlalchemy import TEXT
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy_utils import LtreeType
from sqlalchemy_utils.types.ltree import LQUERY
from teal.db import CASCADE_OWN, UUIDLtree, check_range, IntEnum
from teal.db import CASCADE_OWN, IntEnum, UUIDLtree, check_range
from teal.resource import url_for_resource
from ereuse_devicehub.db import create_view, db, exp, f
@ -21,24 +21,29 @@ from ereuse_devicehub.resources.user.models import User
class Lot(Thing):
id = db.Column(UUID(as_uuid=True), primary_key=True) # uuid is generated on init by default
id = db.Column(
UUID(as_uuid=True), primary_key=True
) # uuid is generated on init by default
name = db.Column(CIText(), nullable=False)
description = db.Column(CIText())
description.comment = """A comment about the lot."""
closed = db.Column(db.Boolean, default=False, nullable=False)
closed.comment = """A closed lot cannot be modified anymore."""
devices = db.relationship(Device,
devices = db.relationship(
Device,
backref=db.backref('lots', lazy=True, collection_class=set),
secondary=lambda: LotDevice.__table__,
lazy=True,
collection_class=set)
collection_class=set,
)
"""The **children** devices that the lot has.
Note that the lot can have more devices, if they are inside
descendant lots.
"""
parents = db.relationship(lambda: Lot,
parents = db.relationship(
lambda: Lot,
viewonly=True,
lazy=True,
collection_class=set,
@ -46,45 +51,58 @@ class Lot(Thing):
primaryjoin=lambda: Lot.id == LotParent.child_id,
secondaryjoin=lambda: LotParent.parent_id == Lot.id,
cascade='refresh-expire', # propagate changes outside ORM
backref=db.backref('children',
backref=db.backref(
'children',
viewonly=True,
lazy=True,
cascade='refresh-expire',
collection_class=set)
collection_class=set,
),
)
"""The parent lots."""
all_devices = db.relationship(Device,
all_devices = db.relationship(
Device,
viewonly=True,
lazy=True,
collection_class=set,
secondary=lambda: LotDeviceDescendants.__table__,
primaryjoin=lambda: Lot.id == LotDeviceDescendants.ancestor_lot_id,
secondaryjoin=lambda: LotDeviceDescendants.device_id == Device.id)
secondaryjoin=lambda: LotDeviceDescendants.device_id == Device.id,
)
"""All devices, including components, inside this lot and its
descendants.
"""
amount = db.Column(db.Integer, check_range('amount', min=0, max=100), default=0)
owner_id = db.Column(UUID(as_uuid=True),
owner_id = db.Column(
UUID(as_uuid=True),
db.ForeignKey(User.id),
nullable=False,
default=lambda: g.user.id)
default=lambda: g.user.id,
)
owner = db.relationship(User, primaryjoin=owner_id == User.id)
transfer_state = db.Column(IntEnum(TransferState), default=TransferState.Initial, nullable=False)
transfer_state = db.Column(
IntEnum(TransferState), default=TransferState.Initial, nullable=False
)
transfer_state.comment = TransferState.__doc__
receiver_address = db.Column(CIText(),
receiver_address = db.Column(
CIText(),
db.ForeignKey(User.email),
nullable=False,
default=lambda: g.user.email)
default=lambda: g.user.email,
)
receiver = db.relationship(User, primaryjoin=receiver_address == User.email)
def __init__(self, name: str, closed: bool = closed.default.arg,
description: str = None) -> None:
def __init__(
self, name: str, closed: bool = closed.default.arg, description: str = None
) -> None:
"""Initializes a lot
:param name:
:param closed:
"""
super().__init__(id=uuid.uuid4(), name=name, closed=closed, description=description)
super().__init__(
id=uuid.uuid4(), name=name, closed=closed, description=description
)
Path(self) # Lots have always one edge per default.
@property
@ -106,16 +124,22 @@ class Lot(Thing):
@property
def is_incoming(self):
return bool(self.trade and self.trade.user_to == current_user)
if hasattr(self, 'trade'):
return self.trade.user_to == g.user
return False
@property
def is_outgoing(self):
return bool(self.trade and self.trade.user_from == current_user)
if hasattr(self, 'trade'):
return self.trade.user_to == g.user
return False
@classmethod
def descendantsq(cls, id):
_id = UUIDLtree.convert(id)
return (cls.id == Path.lot_id) & Path.path.lquery(exp.cast('*.{}.*'.format(_id), LQUERY))
return (cls.id == Path.lot_id) & Path.path.lquery(
exp.cast('*.{}.*'.format(_id), LQUERY)
)
@classmethod
def roots(cls):
@ -176,13 +200,17 @@ class Lot(Thing):
if isinstance(child, Lot):
return Path.has_lot(self.id, child.id)
elif isinstance(child, Device):
device = db.session.query(LotDeviceDescendants) \
.filter(LotDeviceDescendants.device_id == child.id) \
.filter(LotDeviceDescendants.ancestor_lot_id == self.id) \
device = (
db.session.query(LotDeviceDescendants)
.filter(LotDeviceDescendants.device_id == child.id)
.filter(LotDeviceDescendants.ancestor_lot_id == self.id)
.one_or_none()
)
return device
else:
raise TypeError('Lot only contains devices and lots, not {}'.format(child.__class__))
raise TypeError(
'Lot only contains devices and lots, not {}'.format(child.__class__)
)
def __repr__(self) -> str:
return '<Lot {0.name} devices={0.devices!r}>'.format(self)
@ -192,35 +220,44 @@ class LotDevice(db.Model):
device_id = db.Column(db.BigInteger, db.ForeignKey(Device.id), primary_key=True)
lot_id = db.Column(UUID(as_uuid=True), db.ForeignKey(Lot.id), primary_key=True)
created = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
author_id = db.Column(UUID(as_uuid=True),
author_id = db.Column(
UUID(as_uuid=True),
db.ForeignKey(User.id),
nullable=False,
default=lambda: g.user.id)
default=lambda: g.user.id,
)
author = db.relationship(User, primaryjoin=author_id == User.id)
author_id.comment = """The user that put the device in the lot."""
class Path(db.Model):
id = db.Column(db.UUID(as_uuid=True),
id = db.Column(
db.UUID(as_uuid=True),
primary_key=True,
server_default=db.text('gen_random_uuid()'))
server_default=db.text('gen_random_uuid()'),
)
lot_id = db.Column(db.UUID(as_uuid=True), db.ForeignKey(Lot.id), nullable=False)
lot = db.relationship(Lot,
backref=db.backref('paths',
lazy=True,
collection_class=set,
cascade=CASCADE_OWN),
primaryjoin=Lot.id == lot_id)
lot = db.relationship(
Lot,
backref=db.backref(
'paths', lazy=True, collection_class=set, cascade=CASCADE_OWN
),
primaryjoin=Lot.id == lot_id,
)
path = db.Column(LtreeType, nullable=False)
created = db.Column(db.TIMESTAMP(timezone=True), server_default=db.text('CURRENT_TIMESTAMP'))
created = db.Column(
db.TIMESTAMP(timezone=True), server_default=db.text('CURRENT_TIMESTAMP')
)
created.comment = """When Devicehub created this."""
__table_args__ = (
# dag.delete_edge needs to disable internally/temporarily the unique constraint
db.UniqueConstraint(path, name='path_unique', deferrable=True, initially='immediate'),
db.UniqueConstraint(
path, name='path_unique', deferrable=True, initially='immediate'
),
db.Index('path_gist', path, postgresql_using='gist'),
db.Index('path_btree', path, postgresql_using='btree'),
db.Index('lot_id_index', lot_id, postgresql_using='hash')
db.Index('lot_id_index', lot_id, postgresql_using='hash'),
)
def __init__(self, lot: Lot) -> None:
@ -243,7 +280,9 @@ class Path(db.Model):
child_id = UUIDLtree.convert(child_id)
return bool(
db.session.execute(
"SELECT 1 from path where path ~ '*.{}.*.{}.*'".format(parent_id, child_id)
"SELECT 1 from path where path ~ '*.{}.*.{}.*'".format(
parent_id, child_id
)
).first()
)
@ -263,47 +302,73 @@ class LotDeviceDescendants(db.Model):
"""Ancestor lot table."""
_desc = Lot.__table__.alias()
"""Descendant lot table."""
lot_device = _desc \
.join(LotDevice, _desc.c.id == LotDevice.lot_id) \
.join(Path, _desc.c.id == Path.lot_id)
lot_device = _desc.join(LotDevice, _desc.c.id == LotDevice.lot_id).join(
Path, _desc.c.id == Path.lot_id
)
"""Join: Path -- Lot -- LotDevice"""
descendants = "path.path ~ (CAST('*.'|| replace(CAST({}.id as text), '-', '_') " \
descendants = (
"path.path ~ (CAST('*.'|| replace(CAST({}.id as text), '-', '_') "
"|| '.*' AS LQUERY))".format(_ancestor.name)
)
"""Query that gets the descendants of the ancestor lot."""
devices = db.select([
devices = (
db.select(
[
LotDevice.device_id,
_desc.c.id.label('parent_lot_id'),
_ancestor.c.id.label('ancestor_lot_id'),
None
]).select_from(_ancestor).select_from(lot_device).where(db.text(descendants))
None,
]
)
.select_from(_ancestor)
.select_from(lot_device)
.where(db.text(descendants))
)
# Components
_parent_device = Device.__table__.alias(name='parent_device')
"""The device that has the access to the lot."""
lot_device_component = lot_device \
.join(_parent_device, _parent_device.c.id == LotDevice.device_id) \
.join(Component, _parent_device.c.id == Component.parent_id)
lot_device_component = lot_device.join(
_parent_device, _parent_device.c.id == LotDevice.device_id
).join(Component, _parent_device.c.id == Component.parent_id)
"""Join: Path -- Lot -- LotDevice -- ParentDevice (Device) -- Component"""
components = db.select([
components = (
db.select(
[
Component.id.label('device_id'),
_desc.c.id.label('parent_lot_id'),
_ancestor.c.id.label('ancestor_lot_id'),
LotDevice.device_id.label('device_parent_id'),
]).select_from(_ancestor).select_from(lot_device_component).where(db.text(descendants))
]
)
.select_from(_ancestor)
.select_from(lot_device_component)
.where(db.text(descendants))
)
__table__ = create_view('lot_device_descendants', devices.union(components))
class LotParent(db.Model):
i = f.index(Path.path, db.func.text2ltree(f.replace(exp.cast(Path.lot_id, TEXT), '-', '_')))
i = f.index(
Path.path, db.func.text2ltree(f.replace(exp.cast(Path.lot_id, TEXT), '-', '_'))
)
__table__ = create_view(
'lot_parent',
db.select([
db.select(
[
Path.lot_id.label('child_id'),
exp.cast(f.replace(exp.cast(f.subltree(Path.path, i - 1, i), TEXT), '_', '-'),
UUID).label('parent_id')
]).select_from(Path).where(i > 0),
exp.cast(
f.replace(
exp.cast(f.subltree(Path.path, i - 1, i), TEXT), '_', '-'
),
UUID,
).label('parent_id'),
]
)
.select_from(Path)
.where(i > 0),
)

9
ereuse_devicehub/templates/ereuse_devicehub/base_site.html
View file

@ -111,6 +111,15 @@
</a>
</li><!-- End Dashboard Nav -->
<li class="nav-heading">Snapshots</li>
<li class="nav-item">
<a class="nav-link collapsed" href="{{ url_for('inventory.snapshotslist') }}">
<i class="bi-menu-button-wide"></i>
<span>Uploaded Snapshots</span>
</a>
</li>
<li class="nav-heading">Devices</li>
<li class="nav-item">

54
ereuse_devicehub/templates/inventory/snapshot_detail.html Normal file
View file

@ -0,0 +1,54 @@
{% extends "ereuse_devicehub/base_site.html" %}
{% block main %}
<div class="pagetitle">
<h1>Inventory</h1>
<nav>
<ol class="breadcrumb">
<li class="breadcrumb-item"><a href="{{ url_for('inventory.devicelist')}}">Inventory</a></li>
<li class="breadcrumb-item active">{{ page_title }}</li>
</ol>
</nav>
</div><!-- End Page Title -->
<section class="section profile">
<div class="row">
<div class="col-xl-12">
<div class="card">
<div class="card-body pt-3">
<h3>{{ snapshot_sid }} | {{ snapshot_uuid }}</h3>
<!-- Bordered Tabs -->
<div class="tab-content pt-2">
<div class="tab-pane fade show active">
<h5 class="card-title">Traceability log Details</h5>
<div class="list-group col-6">
{% for log in snapshots_log %}
<div class="list-group-item">
<div class="d-flex w-100 justify-content-between">
<h5 class="mb-1">{{ log.get_status() }}</h5>
<small class="text-muted">{{ log.created.strftime('%H:%M %d-%m-%Y') }}</small>
</div>
<p class="mb-1">
Device:
{{ log.get_device() }}<br />
Version: {{ log.version }}<br />
</p>
<p>
<small class="text-muted">
{{ log.description }}
</small>
</p>
</div>
{% endfor %}
</div>
</div>
</div>
</div>
</div>
</div>
</section>
{% endblock main %}

88
ereuse_devicehub/templates/inventory/snapshots_list.html Normal file
View file

@ -0,0 +1,88 @@
{% extends "ereuse_devicehub/base_site.html" %}
{% block main %}
<div class="pagetitle">
<h1>Inventory</h1>
<nav>
<ol class="breadcrumb">
<li class="breadcrumb-item"><a href="{{ url_for('inventory.devicelist')}}">Inventory</a></li>
<li class="breadcrumb-item active">Snapshots</li>
</ol>
</nav>
</div><!-- End Page Title -->
<section class="section profile">
<div class="row">
<div class="col-xl-12">
<div class="card">
<div class="card-body pt-3" style="min-height: 650px;">
<!-- Bordered Tabs -->
<div class="tab-content pt-5">
<div id="devices-list" class="tab-pane fade devices-list active show">
<div class="tab-content pt-2">
<table class="table">
<thead>
<tr>
<th scope="col">SID</th>
<th scope="col">Snapshot id</th>
<th scope="col">Version</th>
<th scope="col">Device</th>
<th scope="col">Status</th>
<th scope="col" data-type="date" data-format="DD-MM-YYYY">Time</th>
</tr>
</thead>
<tbody>
{% for snap in snapshots_log %}
<tr>
<td>
{% if snap.sid %}
<a href="{{ url_for('inventory.snapshot_detail', snapshot_uuid=snap.snapshot_uuid) }}">
{{ snap.sid }}
</a>
{% endif %}
</td>
<td>
<a href="{{ url_for('inventory.snapshot_detail', snapshot_uuid=snap.snapshot_uuid) }}">
{{ snap.snapshot_uuid }}
</a>
</td>
<td>
{{ snap.version }}
</td>
<td>
{% if snap.device %}
<a href="{{ url_for('inventory.device_details', id=snap.device) }}">
{{ snap.device }}
</a>
{% endif %}
</td>
<td>
{{ snap.status }}
</td>
<td>{{ snap.created.strftime('%H:%M %d-%m-%Y') }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
</div><!-- End Bordered Tabs -->
</div>
</div>
</div>
<div id="NotificationsContainer" style="position: absolute; bottom: 0; right: 0; margin: 10px; margin-top: 70px; width: calc(100% - 310px);"></div>
</div>
</div>
</section>
<!-- Custom Code -->
<script>
const table = new simpleDatatables.DataTable("table")
</script>
{% endblock main %}

1
ereuse_devicehub/views.py
View file

@ -50,6 +50,7 @@ class LogoutView(View):
class GenericMixin(View):
methods = ['GET']
decorators = [login_required]
def get_lots(self):

33
tests/test_snapshot.py
View file

@ -15,10 +15,10 @@ from requests.exceptions import HTTPError
from teal.db import DBError, UniqueViolation
from teal.marshmallow import ValidationError
from ereuse_devicehub.client import UserClient, Client
from ereuse_devicehub.client import Client, UserClient
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.parser.models import SnapshotErrors
from ereuse_devicehub.parser.models import SnapshotsLog
from ereuse_devicehub.resources.action.models import (
Action,
BenchmarkDataStorage,
@ -977,8 +977,9 @@ def test_snapshot_wb_lite(user: UserClient):
assert '00:28:f8:a6:d5:7e' in dev.hid
assert dev.actions[0].power_on_hours == 6032
errors = SnapshotErrors.query.filter().all()
assert errors == []
errors = SnapshotsLog.query.filter().all()
assert len(errors) == 1
assert errors[0].description == 'Ok'
@pytest.mark.mvp
@ -986,9 +987,7 @@ def test_snapshot_wb_lite(user: UserClient):
def test_snapshot_wb_lite_qemu(user: UserClient):
"""This test check the minimum validation of json that come from snapshot"""
snapshot = file_json(
"qemu-cc9927a9-55ad-4937-b36b-7185147d9fa9.json"
)
snapshot = file_json("qemu-cc9927a9-55ad-4937-b36b-7185147d9fa9.json")
body, res = user.post(snapshot, uri="/api/inventory/")
assert body['sid'] == "VL0L5"
@ -1172,8 +1171,8 @@ def test_snapshot_lite_error_in_components(user: UserClient):
dev = m.Device.query.filter_by(devicehub_id=bodyLite['dhid']).one()
assert 'Motherboard' not in [x.type for x in dev.components]
error = SnapshotErrors.query.all()[0]
assert 'StopIteration' in error.description
error = SnapshotsLog.query.all()
assert 'StopIteration' in error[0].description
@pytest.mark.mvp
@ -1225,12 +1224,12 @@ def test_snapshot_errors(user: UserClient):
},
}
assert SnapshotErrors.query.all() == []
assert SnapshotsLog.query.all() == []
body11, res = user.post(snapshot_11, res=Snapshot)
assert SnapshotErrors.query.all() == []
assert SnapshotsLog.query.all() == []
bodyLite, res = user.post(snapshot_lite, uri="/api/inventory/")
dev = m.Device.query.filter_by(devicehub_id=bodyLite['dhid']).one()
assert len(SnapshotErrors.query.all()) == 2
assert len(SnapshotsLog.query.all()) == 3
assert body11['device'].get('hid') == dev.hid
assert body11['device']['id'] == dev.id
@ -1267,7 +1266,9 @@ def test_snapshot_errors_no_serial_number(user: UserClient):
bodyLite, res = user.post(snapshot_lite, uri="/api/inventory/")
assert res.status_code == 201
assert len(SnapshotErrors.query.all()) == 0
logs = SnapshotsLog.query.all()
assert len(logs) == 1
assert logs[0].description == 'Ok'
dev = m.Device.query.filter_by(devicehub_id=bodyLite['dhid']).one()
assert not dev.model
assert not dev.manufacturer
@ -1286,9 +1287,11 @@ def test_snapshot_errors_no_serial_number(user: UserClient):
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_snapshot_check_tests_lite(user: UserClient):
"""This test check the minimum validation of json that come from snapshot"""
snapshot_lite = file_json('test_lite/2022-4-13-19-5_user@dhub.com_b27dbf43-b88a-4505-ae27-10de5a95919e.json')
snapshot_lite = file_json(
'test_lite/2022-4-13-19-5_user@dhub.com_b27dbf43-b88a-4505-ae27-10de5a95919e.json'
)
bodyLite, res = user.post(snapshot_lite, uri="/api/inventory/")
assert res.status_code == 201
SnapshotErrors.query.all()
SnapshotsLog.query.all()
dev = m.Device.query.filter_by(devicehub_id=bodyLite['dhid']).one()