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

validate fields for the new snapshot

This commit is contained in:
Cayo Puigdefabregas 2022-03-18 12:07:22 +01:00
parent 6bad17013d
commit 18883b2b8d
3 changed files with 485 additions and 263 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

275
ereuse_devicehub/resources/action/schemas.py
View File

@ -1,14 +1,28 @@
import copy
from datetime import datetime, timedelta
from dateutil.tz import tzutc
from flask import current_app as app, g
from marshmallow import Schema as MarshmallowSchema, ValidationError, fields as f, validates_schema, pre_load, post_load
from marshmallow.fields import Boolean, DateTime, Decimal, Float, Integer, Nested, String, \
TimeDelta, UUID
from flask import current_app as app
from flask import g
from marshmallow import Schema as MarshmallowSchema
from marshmallow import ValidationError
from marshmallow import fields as f
from marshmallow import post_load, pre_load, validates_schema
from marshmallow.fields import (
UUID,
Boolean,
DateTime,
Decimal,
Float,
Integer,
Nested,
String,
TimeDelta,
)
from marshmallow.validate import Length, OneOf, Range
from sqlalchemy.util import OrderedSet
from teal.enums import Country, Currency, Subdivision
from teal.marshmallow import EnumField, IP, SanitizedStr, URL, Version
from teal.marshmallow import IP, URL, EnumField, SanitizedStr, Version
from teal.resource import Schema
from ereuse_devicehub.marshmallow import NestedOn
@ -16,24 +30,32 @@ from ereuse_devicehub.resources import enums
from ereuse_devicehub.resources.action import models as m
from ereuse_devicehub.resources.agent import schemas as s_agent
from ereuse_devicehub.resources.device import schemas as s_device
from ereuse_devicehub.resources.tradedocument import schemas as s_document
from ereuse_devicehub.resources.documents import schemas as s_generic_document
from ereuse_devicehub.resources.enums import AppearanceRange, BiosAccessRange, FunctionalityRange, \
PhysicalErasureMethod, R_POSITIVE, RatingRange, \
Severity, SnapshotSoftware, TestDataStorageLength
from ereuse_devicehub.resources.enums import (
R_POSITIVE,
AppearanceRange,
BiosAccessRange,
FunctionalityRange,
PhysicalErasureMethod,
RatingRange,
Severity,
SnapshotSoftware,
TestDataStorageLength,
)
from ereuse_devicehub.resources.models import STR_BIG_SIZE, STR_SIZE
from ereuse_devicehub.resources.schemas import Thing
from ereuse_devicehub.resources.tradedocument import schemas as s_document
from ereuse_devicehub.resources.tradedocument.models import TradeDocument
from ereuse_devicehub.resources.user import schemas as s_user
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.tradedocument.models import TradeDocument
class Action(Thing):
__doc__ = m.Action.__doc__
id = UUID(dump_only=True)
name = SanitizedStr(default='',
validate=Length(max=STR_BIG_SIZE),
description=m.Action.name.comment)
name = SanitizedStr(
default='', validate=Length(max=STR_BIG_SIZE), description=m.Action.name.comment
)
closed = Boolean(missing=True, description=m.Action.closed.comment)
severity = EnumField(Severity, description=m.Action.severity.comment)
description = SanitizedStr(default='', description=m.Action.description.comment)
@ -43,7 +65,9 @@ class Action(Thing):
agent = NestedOn(s_agent.Agent, description=m.Action.agent_id.comment)
author = NestedOn(s_user.User, dump_only=True, exclude=('token',))
components = NestedOn(s_device.Component, dump_only=True, many=True)
parent = NestedOn(s_device.Computer, dump_only=True, description=m.Action.parent_id.comment)
parent = NestedOn(
s_device.Computer, dump_only=True, description=m.Action.parent_id.comment
)
url = URL(dump_only=True, description=m.Action.url.__doc__)
@validates_schema
@ -67,24 +91,27 @@ class ActionWithOneDevice(Action):
class ActionWithMultipleDocuments(Action):
__doc__ = m.ActionWithMultipleTradeDocuments.__doc__
documents = NestedOn(s_document.TradeDocument,
documents = NestedOn(
s_document.TradeDocument,
many=True,
required=True, # todo test ensuring len(devices) >= 1
only_query='id',
collection_class=OrderedSet)
collection_class=OrderedSet,
)
class ActionWithMultipleDevices(Action):
__doc__ = m.ActionWithMultipleDevices.__doc__
devices = NestedOn(s_device.Device,
devices = NestedOn(
s_device.Device,
many=True,
required=True, # todo test ensuring len(devices) >= 1
only_query='id',
collection_class=OrderedSet)
collection_class=OrderedSet,
)
class ActionWithMultipleDevicesCheckingOwner(ActionWithMultipleDevices):
@post_load
def check_owner_of_device(self, data):
for dev in data['devices']:
@ -102,20 +129,29 @@ class Remove(ActionWithOneDevice):
class Allocate(ActionWithMultipleDevicesCheckingOwner):
__doc__ = m.Allocate.__doc__
start_time = DateTime(data_key='startTime', required=True,
description=m.Action.start_time.comment)
end_time = DateTime(data_key='endTime', required=False,
description=m.Action.end_time.comment)
final_user_code = SanitizedStr(data_key="finalUserCode",
start_time = DateTime(
data_key='startTime', required=True, description=m.Action.start_time.comment
)
end_time = DateTime(
data_key='endTime', required=False, description=m.Action.end_time.comment
)
final_user_code = SanitizedStr(
data_key="finalUserCode",
validate=Length(min=1, max=STR_BIG_SIZE),
required=False,
description='This is a internal code for mainteing the secrets of the \
personal datas of the new holder')
transaction = SanitizedStr(validate=Length(min=1, max=STR_BIG_SIZE),
personal datas of the new holder',
)
transaction = SanitizedStr(
validate=Length(min=1, max=STR_BIG_SIZE),
required=False,
description='The code used from the owner for \
relation with external tool.')
end_users = Integer(data_key='endUsers', validate=[Range(min=1, error="Value must be greater than 0")])
relation with external tool.',
)
end_users = Integer(
data_key='endUsers',
validate=[Range(min=1, error="Value must be greater than 0")],
)
@validates_schema
def validate_allocate(self, data: dict):
@ -136,12 +172,15 @@ class Allocate(ActionWithMultipleDevicesCheckingOwner):
class Deallocate(ActionWithMultipleDevicesCheckingOwner):
__doc__ = m.Deallocate.__doc__
start_time = DateTime(data_key='startTime', required=True,
description=m.Action.start_time.comment)
transaction = SanitizedStr(validate=Length(min=1, max=STR_BIG_SIZE),
start_time = DateTime(
data_key='startTime', required=True, description=m.Action.start_time.comment
)
transaction = SanitizedStr(
validate=Length(min=1, max=STR_BIG_SIZE),
required=False,
description='The code used from the owner for \
relation with external tool.')
relation with external tool.',
)
@validates_schema
def validate_deallocate(self, data: dict):
@ -232,7 +271,9 @@ class MeasureBattery(Test):
__doc__ = m.MeasureBattery.__doc__
size = Integer(required=True, description=m.MeasureBattery.size.comment)
voltage = Integer(required=True, description=m.MeasureBattery.voltage.comment)
cycle_count = Integer(data_key='cycleCount', description=m.MeasureBattery.cycle_count.comment)
cycle_count = Integer(
data_key='cycleCount', description=m.MeasureBattery.cycle_count.comment
)
health = EnumField(enums.BatteryHealth, description=m.MeasureBattery.health.comment)
@ -289,28 +330,32 @@ class TestBios(Test):
class VisualTest(Test):
__doc__ = m.VisualTest.__doc__
appearance_range = EnumField(AppearanceRange, data_key='appearanceRange')
functionality_range = EnumField(FunctionalityRange,
data_key='functionalityRange')
functionality_range = EnumField(FunctionalityRange, data_key='functionalityRange')
labelling = Boolean()
class Rate(ActionWithOneDevice):
__doc__ = m.Rate.__doc__
rating = Integer(validate=Range(*R_POSITIVE),
rating = Integer(
validate=Range(*R_POSITIVE), dump_only=True, description=m.Rate._rating.comment
)
version = Version(dump_only=True, description=m.Rate.version.comment)
appearance = Integer(
validate=Range(enums.R_NEGATIVE),
dump_only=True,
description=m.Rate._rating.comment)
version = Version(dump_only=True,
description=m.Rate.version.comment)
appearance = Integer(validate=Range(enums.R_NEGATIVE),
description=m.Rate._appearance.comment,
)
functionality = Integer(
validate=Range(enums.R_NEGATIVE),
dump_only=True,
description=m.Rate._appearance.comment)
functionality = Integer(validate=Range(enums.R_NEGATIVE),
dump_only=True,
description=m.Rate._functionality.comment)
rating_range = EnumField(RatingRange,
description=m.Rate._functionality.comment,
)
rating_range = EnumField(
RatingRange,
dump_only=True,
data_key='ratingRange',
description=m.Rate.rating_range.__doc__)
description=m.Rate.rating_range.__doc__,
)
class RateComputer(Rate):
@ -320,19 +365,25 @@ class RateComputer(Rate):
data_storage = Float(dump_only=True, data_key='dataStorage')
graphic_card = Float(dump_only=True, data_key='graphicCard')
data_storage_range = EnumField(RatingRange, dump_only=True, data_key='dataStorageRange')
data_storage_range = EnumField(
RatingRange, dump_only=True, data_key='dataStorageRange'
)
ram_range = EnumField(RatingRange, dump_only=True, data_key='ramRange')
processor_range = EnumField(RatingRange, dump_only=True, data_key='processorRange')
graphic_card_range = EnumField(RatingRange, dump_only=True, data_key='graphicCardRange')
graphic_card_range = EnumField(
RatingRange, dump_only=True, data_key='graphicCardRange'
)
class Price(ActionWithOneDevice):
__doc__ = m.Price.__doc__
currency = EnumField(Currency, required=True, description=m.Price.currency.comment)
price = Decimal(places=m.Price.SCALE,
price = Decimal(
places=m.Price.SCALE,
rounding=m.Price.ROUND,
required=True,
description=m.Price.price.comment)
description=m.Price.price.comment,
)
version = Version(dump_only=True, description=m.Price.version.comment)
rating = NestedOn(Rate, dump_only=True, description=m.Price.rating_id.comment)
@ -356,13 +407,33 @@ class EreusePrice(Price):
class Install(ActionWithOneDevice):
__doc__ = m.Install.__doc__
name = SanitizedStr(validate=Length(min=4, max=STR_BIG_SIZE),
name = SanitizedStr(
validate=Length(min=4, max=STR_BIG_SIZE),
required=True,
description='The name of the OS installed.')
description='The name of the OS installed.',
)
elapsed = TimeDelta(precision=TimeDelta.SECONDS, required=True)
address = Integer(validate=OneOf({8, 16, 32, 64, 128, 256}))
class Snapshot2(MarshmallowSchema):
uuid = UUID()
version = Version(required=True, description='The version of the software.')
type = String()
endTime = DateTime('iso', dump_only=True, description=m.Thing.updated.comment)
@validates_schema
def validate_workbench_version(self, data: dict):
if data['version'] < app.config['MIN_WORKBENCH']:
raise ValidationError(
'Min. supported Workbench version is '
'{} but yours is {}.'.format(
app.config['MIN_WORKBENCH'], data['version']
),
field_names=['version'],
)
class Snapshot(ActionWithOneDevice):
__doc__ = m.Snapshot.__doc__
"""
@ -372,18 +443,22 @@ class Snapshot(ActionWithOneDevice):
See docs for more info.
"""
uuid = UUID()
software = EnumField(SnapshotSoftware,
software = EnumField(
SnapshotSoftware,
required=True,
description='The software that generated this Snapshot.')
description='The software that generated this Snapshot.',
)
version = Version(required=True, description='The version of the software.')
actions = NestedOn(Action, many=True, dump_only=True)
elapsed = TimeDelta(precision=TimeDelta.SECONDS)
components = NestedOn(s_device.Component,
components = NestedOn(
s_device.Component,
many=True,
description='A list of components that are inside of the device'
'at the moment of this Snapshot.'
'Order is preserved, so the component num 0 when'
'submitting is the component num 0 when returning it back.')
'submitting is the component num 0 when returning it back.',
)
@validates_schema
def validate_workbench_version(self, data: dict):
@ -391,16 +466,21 @@ class Snapshot(ActionWithOneDevice):
if data['version'] < app.config['MIN_WORKBENCH']:
raise ValidationError(
'Min. supported Workbench version is '
'{} but yours is {}.'.format(app.config['MIN_WORKBENCH'], data['version']),
field_names=['version']
'{} but yours is {}.'.format(
app.config['MIN_WORKBENCH'], data['version']
),
field_names=['version'],
)
@validates_schema
def validate_components_only_workbench(self, data: dict):
if (data['software'] != SnapshotSoftware.Workbench) and (data['software'] != SnapshotSoftware.WorkbenchAndroid):
if (data['software'] != SnapshotSoftware.Workbench) and (
data['software'] != SnapshotSoftware.WorkbenchAndroid
):
if data.get('components', None) is not None:
raise ValidationError('Only Workbench can add component info',
field_names=['components'])
raise ValidationError(
'Only Workbench can add component info', field_names=['components']
)
@validates_schema
def validate_only_workbench_fields(self, data: dict):
@ -408,22 +488,32 @@ class Snapshot(ActionWithOneDevice):
# todo test
if data['software'] == SnapshotSoftware.Workbench:
if not data.get('uuid', None):
raise ValidationError('Snapshots from Workbench and WorkbenchAndroid must have uuid',
field_names=['uuid'])
raise ValidationError(
'Snapshots from Workbench and WorkbenchAndroid must have uuid',
field_names=['uuid'],
)
if data.get('elapsed', None) is None:
raise ValidationError('Snapshots from Workbench must have elapsed',
field_names=['elapsed'])
raise ValidationError(
'Snapshots from Workbench must have elapsed',
field_names=['elapsed'],
)
elif data['software'] == SnapshotSoftware.WorkbenchAndroid:
if not data.get('uuid', None):
raise ValidationError('Snapshots from Workbench and WorkbenchAndroid must have uuid',
field_names=['uuid'])
raise ValidationError(
'Snapshots from Workbench and WorkbenchAndroid must have uuid',
field_names=['uuid'],
)
else:
if data.get('uuid', None):
raise ValidationError('Only Snapshots from Workbench or WorkbenchAndroid can have uuid',
field_names=['uuid'])
raise ValidationError(
'Only Snapshots from Workbench or WorkbenchAndroid can have uuid',
field_names=['uuid'],
)
if data.get('elapsed', None):
raise ValidationError('Only Snapshots from Workbench can have elapsed',
field_names=['elapsed'])
raise ValidationError(
'Only Snapshots from Workbench can have elapsed',
field_names=['elapsed'],
)
class ToRepair(ActionWithMultipleDevicesCheckingOwner):
@ -440,16 +530,20 @@ class Ready(ActionWithMultipleDevicesCheckingOwner):
class ActionStatus(Action):
rol_user = NestedOn(s_user.User, dump_only=True, exclude=('token',))
devices = NestedOn(s_device.Device,
devices = NestedOn(
s_device.Device,
many=True,
required=False, # todo test ensuring len(devices) >= 1
only_query='id',
collection_class=OrderedSet)
documents = NestedOn(s_document.TradeDocument,
collection_class=OrderedSet,
)
documents = NestedOn(
s_document.TradeDocument,
many=True,
required=False, # todo test ensuring len(devices) >= 1
only_query='id',
collection_class=OrderedSet)
collection_class=OrderedSet,
)
@pre_load
def put_devices(self, data: dict):
@ -508,20 +602,28 @@ class Live(ActionWithOneDevice):
See docs for more info.
"""
uuid = UUID()
software = EnumField(SnapshotSoftware,
software = EnumField(
SnapshotSoftware,
required=True,
description='The software that generated this Snapshot.')
description='The software that generated this Snapshot.',
)
version = Version(required=True, description='The version of the software.')
final_user_code = SanitizedStr(data_key="finalUserCode", dump_only=True)
licence_version = Version(required=True, description='The version of the software.')
components = NestedOn(s_device.Component,
components = NestedOn(
s_device.Component,
many=True,
description='A list of components that are inside of the device'
'at the moment of this Snapshot.'
'Order is preserved, so the component num 0 when'
'submitting is the component num 0 when returning it back.')
usage_time_allocate = TimeDelta(data_key='usageTimeAllocate', required=False,
precision=TimeDelta.HOURS, dump_only=True)
'submitting is the component num 0 when returning it back.',
)
usage_time_allocate = TimeDelta(
data_key='usageTimeAllocate',
required=False,
precision=TimeDelta.HOURS,
dump_only=True,
)
class Organize(ActionWithMultipleDevices):
@ -691,26 +793,23 @@ class Trade(ActionWithMultipleDevices):
validate=Length(max=STR_SIZE),
data_key='userToEmail',
missing='',
required=False
required=False,
)
user_to = NestedOn(s_user.User, dump_only=True, data_key='userTo')
user_from_email = SanitizedStr(
validate=Length(max=STR_SIZE),
data_key='userFromEmail',
missing='',
required=False
required=False,
)
user_from = NestedOn(s_user.User, dump_only=True, data_key='userFrom')
code = SanitizedStr(validate=Length(max=STR_SIZE), data_key='code', required=False)
confirm = Boolean(
data_key='confirms',
missing=True,
description="""If you need confirmation of the user you need actevate this field"""
description="""If you need confirmation of the user you need actevate this field""",
)
lot = NestedOn('Lot',
many=False,
required=True,
only_query='id')
lot = NestedOn('Lot', many=False, required=True, only_query='id')
@pre_load
def adding_devices(self, data: dict):

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

@ -1,18 +1,21 @@
""" This is the view for Snapshots """
import os
import json
import os
import shutil
from datetime import datetime
from flask import current_app as app, g
from flask import current_app as app
from flask import g
from flask.json import jsonify
from sqlalchemy.util import OrderedSet
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.action.models import RateComputer, Snapshot
from ereuse_devicehub.resources.device.models import Computer
from ereuse_devicehub.resources.action.rate.v1_0 import CannotRate
from ereuse_devicehub.resources.enums import SnapshotSoftware, Severity
from ereuse_devicehub.resources.action.schemas import Snapshot2
from ereuse_devicehub.resources.device.models import Computer
from ereuse_devicehub.resources.enums import Severity, SnapshotSoftware
from ereuse_devicehub.resources.user.exceptions import InsufficientPermission
@ -59,11 +62,12 @@ def move_json(tmp_snapshots, path_name, user, live=False):
os.remove(path_name)
class SnapshotView():
class SnapshotView:
"""Performs a Snapshot.
See `Snapshot` section in docs for more info.
"""
# Note that if we set the device / components into the snapshot
# model object, when we flush them to the db we will flush
# snapshot, and we want to wait to flush snapshot at the end
@ -74,6 +78,10 @@ class SnapshotView():
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)
if snapshot_json.get('version') in ["14.0.0"]:
self.validate_json(snapshot_json)
self.response = self.build2()
else:
self.snapshot_json = resource_def.schema.load(snapshot_json)
self.response = self.build()
move_json(self.tmp_snapshots, self.path_snapshot, g.user.email)
@ -84,8 +92,12 @@ class SnapshotView():
def build(self):
device = self.snapshot_json.pop('device') # type: Computer
components = None
if self.snapshot_json['software'] == (SnapshotSoftware.Workbench or SnapshotSoftware.WorkbenchAndroid):
components = self.snapshot_json.pop('components', None) # type: List[Component]
if self.snapshot_json['software'] == (
SnapshotSoftware.Workbench or SnapshotSoftware.WorkbenchAndroid
):
components = self.snapshot_json.pop(
'components', None
) # type: List[Component]
if isinstance(device, Computer) and device.hid:
device.add_mac_to_hid(components_snap=components)
snapshot = Snapshot(**self.snapshot_json)
@ -94,7 +106,9 @@ class SnapshotView():
actions_device = set(e for e in device.actions_one)
device.actions_one.clear()
if components:
actions_components = tuple(set(e for e in c.actions_one) for c in components)
actions_components = tuple(
set(e for e in c.actions_one) for c in components
)
for component in components:
component.actions_one.clear()
@ -141,3 +155,12 @@ class SnapshotView():
ret.status_code = 201
db.session.commit()
return ret
def validate_json(self, snapshot_json):
self.schema2 = Snapshot2()
self.snapshot_json = self.schema2.load(snapshot_json)
def build2(self):
res = jsonify("Ok")
res.status_code = 201
return res

288
tests/test_snapshot.py
View File

@ -1,38 +1,48 @@
import os
import json
import os
import shutil
import pytest
import uuid
from datetime import datetime, timedelta, timezone
from requests.exceptions import HTTPError
from operator import itemgetter
from typing import List, Tuple
from uuid import uuid4
import pytest
from boltons import urlutils
from teal.db import UniqueViolation, DBError
from teal.marshmallow import ValidationError
from ereuse_utils.test import ANY
from requests.exceptions import HTTPError
from teal.db import DBError, UniqueViolation
from teal.marshmallow import ValidationError
from ereuse_devicehub.client import UserClient
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.action.models import Action, BenchmarkDataStorage, \
BenchmarkProcessor, EraseSectors, RateComputer, Snapshot, SnapshotRequest, VisualTest, \
EreusePrice, Ready
from ereuse_devicehub.resources.action.models import (
Action,
BenchmarkDataStorage,
BenchmarkProcessor,
EraseSectors,
EreusePrice,
RateComputer,
Ready,
Snapshot,
SnapshotRequest,
VisualTest,
)
from ereuse_devicehub.resources.action.views.snapshot import save_json
from ereuse_devicehub.resources.device import models as m
from ereuse_devicehub.resources.device.exceptions import NeedsId
from ereuse_devicehub.resources.device.models import SolidStateDrive
from ereuse_devicehub.resources.device.sync import MismatchBetweenProperties, \
MismatchBetweenTagsAndHid
from ereuse_devicehub.resources.device.sync import (
MismatchBetweenProperties,
MismatchBetweenTagsAndHid,
)
from ereuse_devicehub.resources.documents import documents
from ereuse_devicehub.resources.enums import ComputerChassis, SnapshotSoftware
from ereuse_devicehub.resources.tag import Tag
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.action.views.snapshot import save_json
from ereuse_devicehub.resources.documents import documents
from tests.conftest import file, yaml2json, json_encode
from tests import conftest
from tests.conftest import file, json_encode, yaml2json
@pytest.mark.mvp
@ -43,18 +53,22 @@ def test_snapshot_model():
"""
device = m.Desktop(serial_number='a1', chassis=ComputerChassis.Tower)
# noinspection PyArgumentList
snapshot = Snapshot(uuid=uuid4(),
snapshot = Snapshot(
uuid=uuid4(),
end_time=datetime.now(timezone.utc),
version='1.0',
software=SnapshotSoftware.DesktopApp,
elapsed=timedelta(seconds=25))
elapsed=timedelta(seconds=25),
)
snapshot.device = device
snapshot.request = SnapshotRequest(request={'foo': 'bar'})
db.session.add(snapshot)
db.session.commit()
device = m.Desktop.query.one() # type: m.Desktop
e1 = device.actions[0]
assert isinstance(e1, Snapshot), 'Creation order must be preserved: 1. snapshot, 2. WR'
assert isinstance(
e1, Snapshot
), 'Creation order must be preserved: 1. snapshot, 2. WR'
db.session.delete(device)
db.session.commit()
assert Snapshot.query.one_or_none() is None
@ -63,7 +77,9 @@ def test_snapshot_model():
assert m.Desktop.query.one_or_none() is None
assert m.Device.query.one_or_none() is None
# Check properties
assert device.url == urlutils.URL('http://localhost/devices/%s' % device.devicehub_id)
assert device.url == urlutils.URL(
'http://localhost/devices/%s' % device.devicehub_id
)
@pytest.mark.mvp
@ -78,13 +94,12 @@ def test_snapshot_post(user: UserClient):
"""Tests the post snapshot endpoint (validation, etc), data correctness,
and relationship correctness.
"""
snapshot = snapshot_and_check(user, yaml2json('basic.snapshot'),
action_types=(
BenchmarkProcessor.t,
VisualTest.t,
RateComputer.t
),
perform_second_snapshot=False)
snapshot = snapshot_and_check(
user,
yaml2json('basic.snapshot'),
action_types=(BenchmarkProcessor.t, VisualTest.t, RateComputer.t),
perform_second_snapshot=False,
)
assert snapshot['software'] == 'Workbench'
assert snapshot['version'] == '11.0'
assert snapshot['uuid'] == 'f5efd26e-8754-46bc-87bf-fbccc39d60d9'
@ -98,8 +113,11 @@ def test_snapshot_post(user: UserClient):
device['components'].sort(key=key)
assert snapshot['components'] == device['components']
assert {c['type'] for c in snapshot['components']} == {m.GraphicCard.t, m.RamModule.t,
m.Processor.t}
assert {c['type'] for c in snapshot['components']} == {
m.GraphicCard.t,
m.RamModule.t,
m.Processor.t,
}
rate = next(e for e in snapshot['actions'] if e['type'] == RateComputer.t)
rate, _ = user.get(res=Action, item=rate['id'])
assert rate['device']['id'] == snapshot['device']['id']
@ -127,20 +145,26 @@ def test_same_device_tow_users(user: UserClient, user2: UserClient):
assert pc['ownerID'] != pc2['ownerID']
assert pc['hid'] == pc2['hid']
@pytest.mark.mvp
def test_snapshot_update_timefield_updated(user: UserClient):
"""
Tests for check if one computer have the time mark updated when one component of it is updated
"""
computer1 = yaml2json('1-device-with-components.snapshot')
snapshot = snapshot_and_check(user,
snapshot = snapshot_and_check(
user,
computer1,
action_types=(BenchmarkProcessor.t,
RateComputer.t),
perform_second_snapshot=False)
action_types=(BenchmarkProcessor.t, RateComputer.t),
perform_second_snapshot=False,
)
computer2 = yaml2json('2-second-device-with-components-of-first.snapshot')
snapshot_and_check(user, computer2, action_types=('Remove', 'RateComputer'),
perform_second_snapshot=False)
snapshot_and_check(
user,
computer2,
action_types=('Remove', 'RateComputer'),
perform_second_snapshot=False,
)
pc1_devicehub_id = snapshot['device']['devicehubID']
pc1, _ = user.get(res=m.Device, item=pc1_devicehub_id)
assert pc1['updated'] != snapshot['device']['updated']
@ -165,7 +189,10 @@ def test_snapshot_power_on_hours(user: UserClient):
test_data_storage = ac
break
assert test_data_storage.lifetime.total_seconds()/3600 == test_data_storage.power_on_hours
assert (
test_data_storage.lifetime.total_seconds() / 3600
== test_data_storage.power_on_hours
)
@pytest.mark.mvp
@ -176,10 +203,7 @@ def test_snapshot_component_add_remove(user: UserClient):
def get_actions_info(actions: List[dict]) -> tuple:
return tuple(
(
e['type'],
[c['serialNumber'] for c in e['components']]
)
(e['type'], [c['serialNumber'] for c in e['components']])
for e in user.get_many(res=Action, resources=actions, key='id')
)
@ -198,7 +222,11 @@ def test_snapshot_component_add_remove(user: UserClient):
pc1, _ = user.get(res=m.Device, item=pc1_devicehub_id)
update1_pc1 = pc1['updated']
# Parent contains components
assert tuple(c['serialNumber'] for c in pc1['components']) == ('p1c1s', 'p1c2s', 'p1c3s')
assert tuple(c['serialNumber'] for c in pc1['components']) == (
'p1c1s',
'p1c2s',
'p1c3s',
)
# Components contain parent
assert all(c['parent'] == pc1_id for c in pc1['components'])
# pc has three actions: Snapshot, BenchmarkProcessor and RateComputer
@ -228,7 +256,12 @@ def test_snapshot_component_add_remove(user: UserClient):
# PC1
assert tuple(c['serialNumber'] for c in pc1['components']) == ('p1c1s', 'p1c3s')
assert all(c['parent'] == pc1_id for c in pc1['components'])
assert tuple(e['type'] for e in pc1['actions']) == ('BenchmarkProcessor', 'Snapshot', 'RateComputer', 'Remove')
assert tuple(e['type'] for e in pc1['actions']) == (
'BenchmarkProcessor',
'Snapshot',
'RateComputer',
'Remove',
)
# PC2
assert tuple(c['serialNumber'] for c in pc2['components']) == ('p1c2s', 'p2c1s')
assert all(c['parent'] == pc2_id for c in pc2['components'])
@ -236,15 +269,24 @@ def test_snapshot_component_add_remove(user: UserClient):
# p1c2s has two Snapshots, a Remove and an Add
p1c2s, _ = user.get(res=m.Device, item=pc2['components'][0]['devicehubID'])
assert tuple(e['type'] for e in p1c2s['actions']) == (
'BenchmarkProcessor', 'Snapshot', 'RateComputer', 'Snapshot', 'Remove', 'RateComputer'
'BenchmarkProcessor',
'Snapshot',
'RateComputer',
'Snapshot',
'Remove',
'RateComputer',
)
# We register the first device again, but removing motherboard
# and moving processor from the second device to the first.
# We have created 1 Remove (from PC2's processor back to PC1)
# PC 0: p1c2s, p1c3s. PC 1: p2c1s
s3 = yaml2json('3-first-device-but-removing-motherboard-and-adding-processor-from-2.snapshot')
snapshot_and_check(user, s3, ('Remove', 'RateComputer'), perform_second_snapshot=False)
s3 = yaml2json(
'3-first-device-but-removing-motherboard-and-adding-processor-from-2.snapshot'
)
snapshot_and_check(
user, s3, ('Remove', 'RateComputer'), perform_second_snapshot=False
)
pc1, _ = user.get(res=m.Device, item=pc1_devicehub_id)
pc2, _ = user.get(res=m.Device, item=pc2_devicehub_id)
# Check if the update_timestamp is updated
@ -263,7 +305,7 @@ def test_snapshot_component_add_remove(user: UserClient):
('RateComputer', ['p1c1s', 'p1c2s', 'p1c3s']),
('Remove', ['p1c2s']), # Remove Processor in Snapshot2
('Snapshot', ['p1c2s', 'p1c3s']), # This Snapshot3
('RateComputer', ['p1c2s', 'p1c3s'])
('RateComputer', ['p1c2s', 'p1c3s']),
)
# PC2
assert tuple(c['serialNumber'] for c in pc2['components']) == ('p2c1s',)
@ -271,7 +313,7 @@ def test_snapshot_component_add_remove(user: UserClient):
assert tuple(e['type'] for e in pc2['actions']) == (
'Snapshot', # Second Snapshot
'RateComputer',
'Remove' # the processor we added in 2.
'Remove', # the processor we added in 2.
)
# p1c2s has Snapshot, Remove and Add
p1c2s, _ = user.get(res=m.Device, item=pc1['components'][0]['devicehubID'])
@ -284,13 +326,17 @@ def test_snapshot_component_add_remove(user: UserClient):
('RateComputer', ['p1c2s', 'p2c1s']),
('Snapshot', ['p1c2s', 'p1c3s']), # The third Snapshot to PC1
('Remove', ['p1c2s']), # ...which caused p1c2 to be removed from PC2
('RateComputer', ['p1c2s', 'p1c3s'])
('RateComputer', ['p1c2s', 'p1c3s']),
)
# We register the first device but without the processor,
# adding a graphic card and adding a new component
s4 = yaml2json('4-first-device-but-removing-processor.snapshot-and-adding-graphic-card')
snapshot4 = snapshot_and_check(user, s4, ('RateComputer',), perform_second_snapshot=False)
s4 = yaml2json(
'4-first-device-but-removing-processor.snapshot-and-adding-graphic-card'
)
snapshot4 = snapshot_and_check(
user, s4, ('RateComputer',), perform_second_snapshot=False
)
pc1, _ = user.get(res=m.Device, item=pc1_devicehub_id)
pc2, _ = user.get(res=m.Device, item=pc2_devicehub_id)
# Check if the update_timestamp is updated
@ -308,6 +354,7 @@ def test_snapshot_component_add_remove(user: UserClient):
assert tuple(c['serialNumber'] for c in pc2['components']) == ('p2c1s',)
assert all(c['parent'] == pc2_id for c in pc2['components'])
@pytest.mark.mvp
def test_snapshot_post_without_hid(user: UserClient):
"""Tests the post snapshot endpoint (validation, etc), data correctness,
@ -338,15 +385,18 @@ def test_snapshot_tag_inner_tag(user: UserClient, tag_id: str, app: Devicehub):
b = yaml2json('basic.snapshot')
b['device']['tags'] = [{'type': 'Tag', 'id': tag_id}]
snapshot_and_check(user, b,
action_types=(RateComputer.t, BenchmarkProcessor.t, VisualTest.t))
snapshot_and_check(
user, b, action_types=(RateComputer.t, BenchmarkProcessor.t, VisualTest.t)
)
with app.app_context():
tag = Tag.query.all()[0] # type: Tag
assert tag.device_id == 3, 'Tag should be linked to the first device'
@pytest.mark.mvp
def test_snapshot_tag_inner_tag_mismatch_between_tags_and_hid(user: UserClient, tag_id: str):
def test_snapshot_tag_inner_tag_mismatch_between_tags_and_hid(
user: UserClient, tag_id: str
):
"""Ensures one device cannot 'steal' the tag from another one."""
pc1 = yaml2json('basic.snapshot')
pc1['device']['tags'] = [{'type': 'Tag', 'id': tag_id}]
@ -396,7 +446,7 @@ def test_snapshot_component_containing_components(user: UserClient):
'type': 'Processor',
'serialNumber': 'foo',
'manufacturer': 'bar',
'model': 'baz'
'model': 'baz',
}
user.post(json_encode(s), res=Snapshot, status=ValidationError)
@ -435,13 +485,15 @@ def test_not_remove_ram_in_same_computer(user: UserClient):
snap1, _ = user.post(json_encode(s), res=Snapshot)
s['uuid'] = '74caa7eb-2bad-4333-94f6-6f1b031d0774'
s['components'].append({
s['components'].append(
{
"actions": [],
"manufacturer": "Intel Corporation",
"model": "NM10/ICH7 Family High Definition Audio Controller",
"serialNumber": "mp2pc",
"type": "SoundCard"
})
"type": "SoundCard",
}
)
dev1 = m.Device.query.filter_by(id=snap1['device']['id']).one()
ram1 = [x.id for x in dev1.components if x.type == 'RamModule'][0]
snap2, _ = user.post(json_encode(s), res=Snapshot)
@ -466,13 +518,18 @@ def test_ereuse_price(user: UserClient):
s = yaml2json('erase-sectors.snapshot')
assert s['components'][0]['actions'][0]['endTime'] == '2018-06-01T09:12:06+02:00'
s['device']['type'] = 'Server'
snapshot = snapshot_and_check(user, s, action_types=(
snapshot = snapshot_and_check(
user,
s,
action_types=(
EraseSectors.t,
BenchmarkDataStorage.t,
BenchmarkProcessor.t,
RateComputer.t,
EreusePrice.t
), perform_second_snapshot=False)
EreusePrice.t,
),
perform_second_snapshot=False,
)
ereuse_price = snapshot['actions'][-1]
assert len(ereuse_price) > 0
@ -487,33 +544,54 @@ def test_erase_privacy_standards_endtime_sort(user: UserClient):
"""
s = yaml2json('erase-sectors.snapshot')
assert s['components'][0]['actions'][0]['endTime'] == '2018-06-01T09:12:06+02:00'
snapshot = snapshot_and_check(user, s, action_types=(
snapshot = snapshot_and_check(
user,
s,
action_types=(
EraseSectors.t,
BenchmarkDataStorage.t,
BenchmarkProcessor.t,
RateComputer.t,
EreusePrice.t
), perform_second_snapshot=False)
EreusePrice.t,
),
perform_second_snapshot=False,
)
# Perform a new snapshot changing the erasure time, as if
# it is a new erasure performed after.
erase = next(e for e in snapshot['actions'] if e['type'] == EraseSectors.t)
assert erase['endTime'] == '2018-06-01T07:12:06+00:00'
s['uuid'] = uuid4()
s['components'][0]['actions'][0]['endTime'] = '2018-06-01T07:14:00+00:00'
snapshot = snapshot_and_check(user, s, action_types=(
snapshot = snapshot_and_check(
user,
s,
action_types=(
EraseSectors.t,
BenchmarkDataStorage.t,
BenchmarkProcessor.t,
RateComputer.t,
EreusePrice.t
), perform_second_snapshot=False)
EreusePrice.t,
),
perform_second_snapshot=False,
)
# The actual test
storage = next(e for e in snapshot['components'] if e['type'] == SolidStateDrive.t)
storage, _ = user.get(res=m.Device, item=storage['devicehubID']) # Let's get storage actions too
storage, _ = user.get(
res=m.Device, item=storage['devicehubID']
) # Let's get storage actions too
# order: endTime ascending
# erasure1/2 have an user defined time and others actions endTime = created
erasure1, erasure2, benchmark_hdd1, _snapshot1, _, _, benchmark_hdd2, _snapshot2 = storage['actions'][:8]
(
erasure1,
erasure2,
benchmark_hdd1,
_snapshot1,
_,
_,
benchmark_hdd2,
_snapshot2,
) = storage['actions'][:8]
assert erasure1['type'] == erasure2['type'] == 'EraseSectors'
assert benchmark_hdd1['type'] == benchmark_hdd2['type'] == 'BenchmarkDataStorage'
assert _snapshot1['type'] == _snapshot2['type'] == 'Snapshot'
@ -555,8 +633,7 @@ def test_test_data_storage(user: UserClient):
s = file('erase-sectors-2-hdd.snapshot')
snapshot, _ = user.post(res=Snapshot, data=s)
incidence_test = next(
ev for ev in snapshot['actions']
if ev.get('reallocatedSectorCount', None) == 15
ev for ev in snapshot['actions'] if ev.get('reallocatedSectorCount', None) == 15
)
assert incidence_test['severity'] == 'Error'
@ -584,10 +661,12 @@ def assert_similar_components(components1: List[dict], components2: List[dict]):
assert_similar_device(c1, c2)
def snapshot_and_check(user: UserClient,
def snapshot_and_check(
user: UserClient,
input_snapshot: dict,
action_types: Tuple[str, ...] = tuple(),
perform_second_snapshot=True) -> dict:
perform_second_snapshot=True,
) -> dict:
"""Performs a Snapshot and then checks if the result is ok:
- There have been performed the types of actions and in the same
@ -610,18 +689,22 @@ def snapshot_and_check(user: UserClient,
if action['type'] == 'Add':
found_add = True
if found_add:
assert action['type'] != 'Receive', 'All Remove actions must be before the Add ones'
assert (
action['type'] != 'Receive'
), 'All Remove actions must be before the Add ones'
assert input_snapshot['device']
assert_similar_device(input_snapshot['device'], snapshot['device'])
if input_snapshot.get('components', None):
assert_similar_components(input_snapshot['components'], snapshot['components'])
assert all(c['parent'] == snapshot['device']['id'] for c in snapshot['components']), \
'Components must be in their parent'
assert all(
c['parent'] == snapshot['device']['id'] for c in snapshot['components']
), 'Components must be in their parent'
if perform_second_snapshot:
if 'uuid' in input_snapshot:
input_snapshot['uuid'] = uuid4()
return snapshot_and_check(user, input_snapshot, action_types,
perform_second_snapshot=False)
return snapshot_and_check(
user, input_snapshot, action_types, perform_second_snapshot=False
)
else:
return snapshot
@ -642,12 +725,12 @@ def test_erase_changing_hdd_between_pcs(user: UserClient):
db.session.commit()
assert dev2.components[1].actions[2].parent == dev1
doc1, response = user.get(res=documents.DocumentDef.t,
item='erasures/{}'.format(dev1.id),
accept=ANY)
doc2, response = user.get(res=documents.DocumentDef.t,
item='erasures/{}'.format(dev2.id),
accept=ANY)
doc1, response = user.get(
res=documents.DocumentDef.t, item='erasures/{}'.format(dev1.id), accept=ANY
)
doc2, response = user.get(
res=documents.DocumentDef.t, item='erasures/{}'.format(dev2.id), accept=ANY
)
assert 'dev1' in doc2
assert 'dev2' in doc2
@ -667,10 +750,9 @@ def test_pc_2(user: UserClient):
snapshot, _ = user.post(res=Snapshot, data=s)
@pytest.mark.mvp
def test_save_snapshot_in_file(app: Devicehub, user: UserClient):
""" This test check if works the function save_snapshot_in_file """
"""This test check if works the function save_snapshot_in_file"""
snapshot_no_hid = yaml2json('basic.snapshot.nohid')
tmp_snapshots = app.config['TMP_SNAPSHOTS']
path_dir_base = os.path.join(tmp_snapshots, user.user['email'], 'errors')
@ -696,7 +778,7 @@ def test_save_snapshot_in_file(app: Devicehub, user: UserClient):
@pytest.mark.mvp
def test_action_no_snapshot_without_save_file(app: Devicehub, user: UserClient):
""" This test check if the function save_snapshot_in_file not work when we
"""This test check if the function save_snapshot_in_file not work when we
send one other action different to snapshot
"""
s = file('laptop-hp_255_g3_notebook-hewlett-packard-cnd52270fw.snapshot')
@ -712,9 +794,10 @@ def test_action_no_snapshot_without_save_file(app: Devicehub, user: UserClient):
assert os.path.exists(tmp_snapshots) == False
@pytest.mark.mvp
def test_save_snapshot_with_debug(app: Devicehub, user: UserClient):
""" This test check if works the function save_snapshot_in_file """
"""This test check if works the function save_snapshot_in_file"""
snapshot_file = yaml2json('basic.snapshot.with_debug')
debug = snapshot_file['debug']
user.post(res=Snapshot, data=json_encode(snapshot_file))
@ -738,7 +821,7 @@ def test_save_snapshot_with_debug(app: Devicehub, user: UserClient):
@pytest.mark.mvp
def test_backup_snapshot_with_errors(app: Devicehub, user: UserClient):
""" This test check if the file snapshot is create when some snapshot is wrong """
"""This test check if the file snapshot is create when some snapshot is wrong"""
tmp_snapshots = app.config['TMP_SNAPSHOTS']
path_dir_base = os.path.join(tmp_snapshots, user.user['email'], 'errors')
snapshot_no_hid = yaml2json('basic.snapshot.badly_formed')
@ -763,7 +846,7 @@ def test_backup_snapshot_with_errors(app: Devicehub, user: UserClient):
@pytest.mark.mvp
def test_snapshot_failed_missing_cpu_benchmark(app: Devicehub, user: UserClient):
""" This test check if the file snapshot is create when some snapshot is wrong """
"""This test check if the file snapshot is create when some snapshot is wrong"""
tmp_snapshots = app.config['TMP_SNAPSHOTS']
path_dir_base = os.path.join(tmp_snapshots, user.user['email'], 'errors')
snapshot_error = yaml2json('failed.snapshot.500.missing-cpu-benchmark')
@ -788,7 +871,7 @@ def test_snapshot_failed_missing_cpu_benchmark(app: Devicehub, user: UserClient)
@pytest.mark.mvp
def test_snapshot_failed_missing_hdd_benchmark(app: Devicehub, user: UserClient):
""" This test check if the file snapshot is create when some snapshot is wrong """
"""This test check if the file snapshot is create when some snapshot is wrong"""
tmp_snapshots = app.config['TMP_SNAPSHOTS']
path_dir_base = os.path.join(tmp_snapshots, user.user['email'], 'errors')
snapshot_error = yaml2json('failed.snapshot.500.missing-hdd-benchmark')
@ -813,7 +896,7 @@ def test_snapshot_failed_missing_hdd_benchmark(app: Devicehub, user: UserClient)
@pytest.mark.mvp
def test_snapshot_not_failed_null_chassis(app: Devicehub, user: UserClient):
""" This test check if the file snapshot is create when some snapshot is wrong """
"""This test check if the file snapshot is create when some snapshot is wrong"""
tmp_snapshots = app.config['TMP_SNAPSHOTS']
path_dir_base = os.path.join(tmp_snapshots, user.user['email'], 'errors')
snapshot_error = yaml2json('desktop-9644w8n-lenovo-0169622.snapshot')
@ -831,7 +914,7 @@ def test_snapshot_not_failed_null_chassis(app: Devicehub, user: UserClient):
@pytest.mark.mvp
def test_snapshot_failed_missing_chassis(app: Devicehub, user: UserClient):
""" This test check if the file snapshot is create when some snapshot is wrong """
"""This test check if the file snapshot is create when some snapshot is wrong"""
tmp_snapshots = app.config['TMP_SNAPSHOTS']
path_dir_base = os.path.join(tmp_snapshots, user.user['email'], 'errors')
snapshot_error = yaml2json('failed.snapshot.422.missing-chassis')
@ -856,7 +939,7 @@ def test_snapshot_failed_missing_chassis(app: Devicehub, user: UserClient):
@pytest.mark.mvp
def test_snapshot_failed_end_time_bug(app: Devicehub, user: UserClient):
""" This test check if the end_time = 0001-01-01 00:00:00+00:00
"""This test check if the end_time = 0001-01-01 00:00:00+00:00
and then we get a /devices, this create a crash
"""
snapshot_file = file('asus-end_time_bug88.snapshot')
@ -870,9 +953,10 @@ def test_snapshot_failed_end_time_bug(app: Devicehub, user: UserClient):
tmp_snapshots = app.config['TMP_SNAPSHOTS']
shutil.rmtree(tmp_snapshots)
@pytest.mark.mvp
def test_snapshot_not_failed_end_time_bug(app: Devicehub, user: UserClient):
""" This test check if the end_time != 0001-01-01 00:00:00+00:00
"""This test check if the end_time != 0001-01-01 00:00:00+00:00
and then we get a /devices, this create a crash
"""
snapshot_file = yaml2json('asus-end_time_bug88.snapshot')
@ -891,7 +975,7 @@ def test_snapshot_not_failed_end_time_bug(app: Devicehub, user: UserClient):
@pytest.mark.mvp
def test_snapshot_bug_smallint_hdd(app: Devicehub, user: UserClient):
""" This test check if the end_time != 0001-01-01 00:00:00+00:00
"""This test check if the end_time != 0001-01-01 00:00:00+00:00
and then we get a /devices, this create a crash
"""
snapshot_file = file('asus-eee-1000h.snapshot.bug1857')
@ -907,7 +991,7 @@ def test_snapshot_bug_smallint_hdd(app: Devicehub, user: UserClient):
@pytest.mark.mvp
def test_snapshot_mobil(app: Devicehub, user: UserClient):
""" This test check if the end_time != 0001-01-01 00:00:00+00:00
"""This test check if the end_time != 0001-01-01 00:00:00+00:00
and then we get a /devices, this create a crash
"""
snapshot_file = file('mobil')
@ -926,3 +1010,19 @@ def test_bug_141(user: UserClient):
"""
dev = file('2021-5-4-13-41_time_out_test_datastorage')
user.post(dev, res=Snapshot)
@pytest.mark.mvp
def test_min_validate_fields(user: UserClient):
"""This test check the minimum validation of json that come from snapshot"""
snapshot = {
"type": "Snapshot",
"uuid": "d1b70cb8-8929-4f36-99b7-fe052cec0abd",
"version": "14.0.0",
"endTime": "2016-11-03T17:17:17.266543+00:00",
"dmidecode": '',
"smartctl": '',
}
body, res = user.post(snapshot, res=Snapshot)
assert body == 'Ok'
assert res.status == '201 CREATED'