From 18883b2b8de4f26fdd38ff8f1844b8e05cb15a17 Mon Sep 17 00:00:00 2001 From: Cayo Puigdefabregas Date: Fri, 18 Mar 2022 12:07:22 +0100 Subject: [PATCH] validate fields for the new snapshot --- ereuse_devicehub/resources/action/schemas.py | 355 +++++++++++------- .../resources/action/views/snapshot.py | 43 ++- tests/test_snapshot.py | 350 +++++++++++------ 3 files changed, 485 insertions(+), 263 deletions(-) diff --git a/ereuse_devicehub/resources/action/schemas.py b/ereuse_devicehub/resources/action/schemas.py index df93b107..dad9c208 100644 --- a/ereuse_devicehub/resources/action/schemas.py +++ b/ereuse_devicehub/resources/action/schemas.py @@ -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, - many=True, - required=True, # todo test ensuring len(devices) >= 1 - only_query='id', - collection_class=OrderedSet) + documents = NestedOn( + s_document.TradeDocument, + many=True, + required=True, # todo test ensuring len(devices) >= 1 + only_query='id', + collection_class=OrderedSet, + ) class ActionWithMultipleDevices(Action): __doc__ = m.ActionWithMultipleDevices.__doc__ - devices = NestedOn(s_device.Device, - many=True, - required=True, # todo test ensuring len(devices) >= 1 - only_query='id', - collection_class=OrderedSet) + devices = NestedOn( + s_device.Device, + many=True, + required=True, # todo test ensuring len(devices) >= 1 + only_query='id', + 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", - 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), - 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")]) + 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), + 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")], + ) @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), - required=False, - description='The code used from the owner for \ - relation with external tool.') + 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.', + ) @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), - 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._appearance.comment) - functionality = Integer(validate=Range(enums.R_NEGATIVE), - dump_only=True, - description=m.Rate._functionality.comment) - rating_range = EnumField(RatingRange, - dump_only=True, - data_key='ratingRange', - description=m.Rate.rating_range.__doc__) + 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._appearance.comment, + ) + functionality = Integer( + validate=Range(enums.R_NEGATIVE), + dump_only=True, + description=m.Rate._functionality.comment, + ) + rating_range = EnumField( + RatingRange, + dump_only=True, + data_key='ratingRange', + 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, - rounding=m.Price.ROUND, - required=True, - description=m.Price.price.comment) + price = Decimal( + places=m.Price.SCALE, + rounding=m.Price.ROUND, + required=True, + 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), - required=True, - description='The name of the OS installed.') + name = SanitizedStr( + validate=Length(min=4, max=STR_BIG_SIZE), + required=True, + 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, - required=True, - description='The software that generated this Snapshot.') + software = EnumField( + SnapshotSoftware, + required=True, + 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, - 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.') + 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.', + ) @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, - many=True, - required=False, # todo test ensuring len(devices) >= 1 - only_query='id', - collection_class=OrderedSet) - documents = NestedOn(s_document.TradeDocument, - many=True, - required=False, # todo test ensuring len(devices) >= 1 - only_query='id', - collection_class=OrderedSet) + 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, + many=True, + required=False, # todo test ensuring len(devices) >= 1 + only_query='id', + 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, - required=True, - description='The software that generated this Snapshot.') + software = EnumField( + SnapshotSoftware, + required=True, + 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, - 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) + 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, + ) class Organize(ActionWithMultipleDevices): @@ -570,7 +672,7 @@ class Revoke(ActionWithMultipleDevices): @validates_schema def validate_documents(self, data): """Check if there are or no one before confirmation, - This is not checked in the view becouse the list of documents is inmutable + This is not checked in the view becouse the list of documents is inmutable """ if not data['devices'] == OrderedSet(): @@ -610,7 +712,7 @@ class ConfirmDocument(ActionWithMultipleDocuments): @validates_schema def validate_documents(self, data): """If there are one device than have one confirmation, - then remove the list this device of the list of devices of this action + then remove the list this device of the list of devices of this action """ if data['documents'] == OrderedSet(): return @@ -636,7 +738,7 @@ class RevokeDocument(ActionWithMultipleDocuments): @validates_schema def validate_documents(self, data): """Check if there are or no one before confirmation, - This is not checked in the view becouse the list of documents is inmutable + This is not checked in the view becouse the list of documents is inmutable """ if data['documents'] == OrderedSet(): @@ -663,7 +765,7 @@ class ConfirmRevokeDocument(ActionWithMultipleDocuments): @validates_schema def validate_documents(self, data): """Check if there are or no one before confirmation, - This is not checked in the view becouse the list of documents is inmutable + This is not checked in the view becouse the list of documents is inmutable """ if data['documents'] == OrderedSet(): @@ -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): diff --git a/ereuse_devicehub/resources/action/views/snapshot.py b/ereuse_devicehub/resources/action/views/snapshot.py index 556211e3..de9564ba 100644 --- a/ereuse_devicehub/resources/action/views/snapshot.py +++ b/ereuse_devicehub/resources/action/views/snapshot.py @@ -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,8 +78,12 @@ 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) - self.snapshot_json = resource_def.schema.load(snapshot_json) - self.response = self.build() + 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) def post(self): @@ -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 diff --git a/tests/test_snapshot.py b/tests/test_snapshot.py index 7e84e4cb..28e22ac3 100644 --- a/tests/test_snapshot.py +++ b/tests/test_snapshot.py @@ -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(), - end_time=datetime.now(timezone.utc), - version='1.0', - software=SnapshotSoftware.DesktopApp, - elapsed=timedelta(seconds=25)) + snapshot = Snapshot( + uuid=uuid4(), + end_time=datetime.now(timezone.utc), + version='1.0', + software=SnapshotSoftware.DesktopApp, + 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, - computer1, - action_types=(BenchmarkProcessor.t, - RateComputer.t), - perform_second_snapshot=False) + snapshot = snapshot_and_check( + user, + computer1, + 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({ - "actions": [], - "manufacturer": "Intel Corporation", - "model": "NM10/ICH7 Family High Definition Audio Controller", - "serialNumber": "mp2pc", - "type": "SoundCard" - }) + s['components'].append( + { + "actions": [], + "manufacturer": "Intel Corporation", + "model": "NM10/ICH7 Family High Definition Audio Controller", + "serialNumber": "mp2pc", + "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=( - EraseSectors.t, - BenchmarkDataStorage.t, - BenchmarkProcessor.t, - RateComputer.t, - EreusePrice.t - ), perform_second_snapshot=False) + snapshot = snapshot_and_check( + user, + s, + action_types=( + EraseSectors.t, + BenchmarkDataStorage.t, + BenchmarkProcessor.t, + RateComputer.t, + 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=( - EraseSectors.t, - BenchmarkDataStorage.t, - BenchmarkProcessor.t, - RateComputer.t, - EreusePrice.t - ), perform_second_snapshot=False) + snapshot = snapshot_and_check( + user, + s, + action_types=( + EraseSectors.t, + BenchmarkDataStorage.t, + BenchmarkProcessor.t, + RateComputer.t, + 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=( - EraseSectors.t, - BenchmarkDataStorage.t, - BenchmarkProcessor.t, - RateComputer.t, - EreusePrice.t - ), perform_second_snapshot=False) + snapshot = snapshot_and_check( + user, + s, + action_types=( + EraseSectors.t, + BenchmarkDataStorage.t, + BenchmarkProcessor.t, + RateComputer.t, + 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,22 +661,24 @@ def assert_similar_components(components1: List[dict], components2: List[dict]): assert_similar_device(c1, c2) -def snapshot_and_check(user: UserClient, - input_snapshot: dict, - action_types: Tuple[str, ...] = tuple(), - perform_second_snapshot=True) -> dict: +def snapshot_and_check( + user: UserClient, + input_snapshot: dict, + action_types: Tuple[str, ...] = tuple(), + 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 - order as described in the passed-in ``action_types``. - - The inputted devices are similar to the resulted ones. - - There is no Remove action after the first Add. - - All input components are now inside the parent device. + - There have been performed the types of actions and in the same + order as described in the passed-in ``action_types``. + - The inputted devices are similar to the resulted ones. + - There is no Remove action after the first Add. + - All input components are now inside the parent device. - Optionally, it can perform a second Snapshot which should - perform an exact result, except for the actions. + Optionally, it can perform a second Snapshot which should + perform an exact result, except for the actions. - :return: The last resulting snapshot. + :return: The last resulting snapshot. """ snapshot, _ = user.post(res=Snapshot, data=json_encode(input_snapshot)) assert all(e['type'] in action_types for e in snapshot['actions']) @@ -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') @@ -921,8 +1005,24 @@ def test_snapshot_mobil(app: Devicehub, user: UserClient): @pytest.mark.mvp def test_bug_141(user: UserClient): """This test check one bug that create a problem when try to up one snapshot - with a big number in the parameter command_timeout of the DataStorage + with a big number in the parameter command_timeout of the DataStorage """ 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'