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

resolve conflicts

Browse source
This commit is contained in:
Cayo Puigdefabregas 2021-06-07 12:08:55 +02:00
parent 2a66fb94a3 289126a8a7
commit f3f65efeaf
37 changed files with 1322 additions and 256 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
CHANGELOG.md
View file

@ -6,12 +6,16 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
ml).
## master
[1.0.5-beta]
## testing
[1.0.6-beta]
## testing
[1.0.7-beta]
## [1.0.7-beta]
- [addend] #140 adding endpoint for download the settings for usb workbench
## [1.0.6-beta]
- [bugfix] #143 biginteger instead of integer in TestDataStorage
## [1.0.5-beta]
- [addend] #124 adding endpoint for extract the internal stats of use

2
ereuse_devicehub/__init__.py
View file

@ -1 +1 @@
__version__ = "1.0.6-beta"
__version__ = "1.0.7-beta"

9
ereuse_devicehub/auth.py
View file

@ -3,12 +3,17 @@ from teal.auth import TokenAuth
from teal.db import ResourceNotFound
from werkzeug.exceptions import Unauthorized
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.user.models import User, Session
class Auth(TokenAuth):
def authenticate(self, token: str, *args, **kw) -> User:
try:
return User.query.filter_by(token=token).one()
user = User.query.filter_by(token=token).first()
if user:
return user
ses = Session.query.filter_by(token=token).one()
return ses.user
except (ResourceNotFound, DataError):
raise Unauthorized('Provide a suitable token.')

1
ereuse_devicehub/config.py
View file

@ -39,6 +39,7 @@ class DevicehubConfig(Config):
DB_PASSWORD = config('DB_PASSWORD', 'ereuse')
DB_HOST = config('DB_HOST', 'localhost')
DB_DATABASE = config('DB_DATABASE', 'devicehub')
DB_SCHEMA = config('DB_SCHEMA', 'dbtest')
SQLALCHEMY_DATABASE_URI = 'postgresql://{user}:{pw}@{host}/{db}'.format(
user=DB_USER,
pw=DB_PASSWORD,

6
ereuse_devicehub/dummy/dummy.py
View file

@ -17,6 +17,8 @@ from ereuse_devicehub.resources.device.models import Device
from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.tag.model import Tag
from ereuse_devicehub.resources.user import User
from ereuse_devicehub.resources.user.models import Session
from ereuse_devicehub.resources.enums import SessionType
class Dummy:
@ -194,6 +196,10 @@ class Dummy:
user.individuals.add(Person(name=name))
db.session.add(user)
session_external = Session(user=user, type=SessionType.External)
session_internal = Session(user=user, type=SessionType.Internal)
db.session.add(session_internal)
db.session.add(session_external)
db.session.commit()
client = UserClient(self.app, user.email, password,

62
ereuse_devicehub/migrations/versions/21afd375a654_session_table.py Normal file
View file

@ -0,0 +1,62 @@
"""session_table
Revision ID: 21afd375a654
Revises: 6a2a939d5668
Create Date: 2021-04-13 11:18:27.720567
"""
from alembic import context
from alembic import op
from sqlalchemy.dialects import postgresql
import sqlalchemy as sa
import sqlalchemy_utils
import citext
import teal
from ereuse_devicehub.resources.enums import SessionType
# revision identifiers, used by Alembic.
revision = '21afd375a654'
down_revision = '398826453b39'
branch_labels = None
depends_on = None
comment_update = 'The last time Devicehub recorded a change for this thing.\n'
def get_inv():
INV = context.get_x_argument(as_dictionary=True).get('inventory')
if not INV:
raise ValueError("Inventory value is not specified")
return INV
def upgrade():
op.create_table('session',
sa.Column('updated', sa.TIMESTAMP(timezone=True),
server_default=sa.text('CURRENT_TIMESTAMP'),
nullable=False,
comment=comment_update),
sa.Column('created', sa.TIMESTAMP(timezone=True),
server_default=sa.text('CURRENT_TIMESTAMP'),
nullable=False, comment='When Devicehub created this.'),
sa.Column('id', sa.BigInteger(), nullable=False),
sa.Column('expired', sa.BigInteger(), nullable=True),
sa.Column('token', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('type', teal.db.IntEnum(SessionType), nullable=False),
sa.Column('user_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.ForeignKeyConstraint(['user_id'], ['common.user.id'], ),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('token'),
schema='common'
)
op.create_index(op.f('ix_session_created'), 'session', ['created'], unique=False, schema='common')
op.create_index(op.f('ix_session_updated'), 'session', ['updated'], unique=False, schema='common')
op.create_index(op.f('ix_session_token'), 'session', ['token'], unique=True, schema='common')
def downgrade():
op.drop_table('trade', schema=f'{get_inv()}')
op.drop_index(op.f('ix_session_created'), table_name='session', schema='common')
op.drop_index(op.f('ix_session_updated'), table_name='session', schema='common')
op.drop_index(op.f('ix_session_token'), table_name='session', schema='common')

30
ereuse_devicehub/migrations/versions/398826453b39_change_command_timeout_of_.py Normal file
View file

@ -0,0 +1,30 @@
"""change command_timeout of TestDataStorage Action
Revision ID: 398826453b39
Revises: 8d34480c82c4
Create Date: 2021-05-12 12:41:02.808311
"""
from alembic import op, context
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '398826453b39'
down_revision = '8d34480c82c4'
branch_labels = None
depends_on = None
def get_inv():
INV = context.get_x_argument(as_dictionary=True).get('inventory')
if not INV:
raise ValueError("Inventory value is not specified")
return INV
def upgrade():
op.alter_column('test_data_storage', 'command_timeout', type_=sa.BigInteger(), schema=f'{get_inv()}')
def downgrade():
op.alter_column('test_data_storage', 'command_timeout', type_=sa.Integer(), schema=f'{get_inv()}')

2
ereuse_devicehub/migrations/versions/51439cf24be8_change_trade_action.py
View file

@ -14,7 +14,7 @@ import citext
# revision identifiers, used by Alembic.
revision = '51439cf24be8'
down_revision = '8d34480c82c4'
down_revision = '21afd375a654'
branch_labels = None
depends_on = None

2
ereuse_devicehub/migrations/versions/8d34480c82c4_add_code_device_search.py
View file

@ -10,7 +10,7 @@ from alembic import context
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from ereuse_devicehub.resources.device.search import DeviceSearch
# from ereuse_devicehub.resources.device.search import DeviceSearch
# revision identifiers, used by Alembic.

2
ereuse_devicehub/resources/action/models.py
View file

@ -757,7 +757,7 @@ class TestDataStorage(TestMixin, Test):
reallocated_sector_count = Column(SmallInteger)
power_cycle_count = Column(SmallInteger)
_reported_uncorrectable_errors = Column('reported_uncorrectable_errors', Integer)
command_timeout = Column(Integer)
command_timeout = Column(BigInteger)
current_pending_sector_count = Column(Integer)
offline_uncorrectable = Column(Integer)
remaining_lifetime_percentage = Column(SmallInteger)

45
ereuse_devicehub/resources/action/schemas.py
View file

@ -640,13 +640,26 @@ class Trade(ActionWithMultipleDevices):
__doc__ = m.Trade.__doc__
date = DateTime(data_key='date', required=False)
price = Float(required=False, data_key='price')
user_to_id = SanitizedStr(validate=Length(max=STR_SIZE), data_key='userTo', missing='',
required=False)
user_from_id = SanitizedStr(validate=Length(max=STR_SIZE), data_key='userFrom', missing='',
required=False)
user_to_email = SanitizedStr(
validate=Length(max=STR_SIZE),
data_key='userToEmail',
missing='',
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
)
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(missing=False, description="""If you need confirmation of the user
you need actevate this field""")
confirm = Boolean(
data_key='confirms',
missing=True,
description="""If you need confirmation of the user you need actevate this field"""
)
lot = NestedOn('Lot',
many=False,
required=True,
@ -654,7 +667,7 @@ class Trade(ActionWithMultipleDevices):
@validates_schema
def validate_lot(self, data: dict):
if not g.user.email in [data['user_from_id'], data['user_to_id']]:
if not g.user.email in [data['user_from_email'], data['user_to_email']]:
txt = "you need to be one of the users of involved in the Trade"
raise ValidationError(txt)
@ -676,7 +689,7 @@ class Trade(ActionWithMultipleDevices):
data['documents'] = data['lot'].documents
@validates_schema
def validate_user_to_id(self, data: dict):
def validate_user_to_email(self, data: dict):
"""
- if user_to exist
* confirmation
@ -685,15 +698,14 @@ class Trade(ActionWithMultipleDevices):
* without confirmation
"""
if data['user_to_id']:
user_to = User.query.filter_by(email=data['user_to_id']).one()
data['user_to_id'] = user_to.id
if data['user_to_email']:
user_to = User.query.filter_by(email=data['user_to_email']).one()
data['user_to'] = user_to
else:
data['confirm'] = False
@validates_schema
def validate_user_from_id(self, data: dict):
def validate_user_from_email(self, data: dict):
"""
- if user_from exist
* confirmation
@ -702,13 +714,12 @@ class Trade(ActionWithMultipleDevices):
* without confirmation
"""
if not (data['user_from_id'] or data['user_to_id']):
if not (data['user_from_email'] or data['user_to_email']):
txt = "you need one user from or user to for to do a offer"
raise ValidationError(txt)
if data['user_from_id']:
user_from = User.query.filter_by(email=data['user_from_id']).one()
data['user_from_id'] = user_from.id
if data['user_from_email']:
user_from = User.query.filter_by(email=data['user_from_email']).one()
data['user_from'] = user_from
else:
data['confirm'] = False
@ -716,7 +727,7 @@ class Trade(ActionWithMultipleDevices):
@validates_schema
def validate_code(self, data: dict):
"""If the user not exist, you need a code to be able to do the traceability"""
if data['user_from_id'] and data['user_to_id']:
if data['user_from_email'] and data['user_to_email']:
return
if not data.get('code'):

169
ereuse_devicehub/resources/action/views/trade.py
View file

@ -7,6 +7,7 @@ from teal.marshmallow import ValidationError
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.action.models import Trade, Confirm, ConfirmRevoke, Revoke
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.lot.views import delete_from_trade
class TradeView():
@ -29,6 +30,8 @@ class TradeView():
def __init__(self, data, resource_def, schema):
self.schema = schema
a = resource_def.schema.load(data)
a.pop('user_to_email', '')
a.pop('user_from_email', '')
self.trade = Trade(**a)
self.create_phantom_account()
db.session.add(self.trade)
@ -63,7 +66,9 @@ class TradeView():
# check than the user than want to do the action is one of the users
# involved in the action
assert g.user.id in [self.trade.user_from_id, self.trade.user_to_id]
if not g.user in [self.trade.user_from, self.trade.user_to]:
txt = "You do not participate in this trading"
raise ValidationError(txt)
if self.trade.user_from == g.user or self.trade.user_from.phantom:
confirm_from = Confirm(user=self.trade.user_from,
@ -87,12 +92,12 @@ class TradeView():
The same if exist to but not from
"""
if self.trade.user_from_id and self.trade.user_to_id:
if self.trade.user_from and self.trade.user_to:
return
if self.trade.user_from_id and not self.trade.user_to_id:
assert g.user.id == self.trade.user_from_id
email = "{}_{}@dhub.com".format(str(self.trade.user_from_id), self.trade.code)
if self.trade.user_from and not self.trade.user_to:
assert g.user == self.trade.user_from
email = "{}_{}@dhub.com".format(str(self.trade.user_from.id), self.trade.code)
users = User.query.filter_by(email=email)
if users.first():
user = users.first()
@ -103,8 +108,8 @@ class TradeView():
db.session.add(user)
self.trade.user_to = user
if not self.trade.user_from_id and self.trade.user_to_id:
email = "{}_{}@dhub.com".format(str(self.trade.user_to_id), self.trade.code)
if not self.trade.user_from and self.trade.user_to:
email = "{}_{}@dhub.com".format(str(self.trade.user_to.id), self.trade.code)
users = User.query.filter_by(email=email)
if users.first():
user = users.first()
@ -176,41 +181,16 @@ class ConfirmView(ConfirmMixin):
"""
real_devices = []
for dev in data['devices']:
actions = copy.copy(dev.actions)
actions.reverse()
for ac in actions:
if ac == data['action']:
# If device have the last action the action Trade
real_devices.append(dev)
break
if ac.t == Confirm.t and not ac.user == g.user:
# If device is confirmed but is not for g.user, then need confirm
real_devices.append(dev)
break
if ac.t == 'Revoke' and not ac.user == g.user:
# If device is revoke before from other user
# it's not possible confirm now
break
if ac.t == 'ConfirmRevoke' and ac.user == g.user:
# if the last action is a ConfirmRevoke this mean than not there are
# other confirmation from the real owner of the trade
break
if ac.t == Confirm.t and ac.user == g.user:
# If device is confirmed we don't need confirmed again
break
ac = dev.last_action_trading
if ac.type == Confirm.t and not ac.user == g.user:
real_devices.append(dev)
data['devices'] = OrderedSet(real_devices)
# Change the owner for every devices
for dev in data['devices']:
dev.owner = data['action'].user_to
if hasattr(dev, 'components'):
for c in dev.components:
c.owner = data['action'].user_to
user_to = data['action'].user_to
dev.change_owner(user_to)
class RevokeView(ConfirmMixin):
@ -226,28 +206,54 @@ class RevokeView(ConfirmMixin):
Model = Revoke
def __init__(self, data, resource_def, schema):
self.schema = schema
a = resource_def.schema.load(data)
self.validate(a)
def validate(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
"""
real_devices = []
"""All devices need to have the status of DoubleConfirmation."""
### check ###
if not data['devices']:
raise ValidationError('Devices not exist.')
for dev in data['devices']:
actions = copy.copy(dev.actions)
actions.reverse()
for ac in actions:
if ac == data['action']:
# data['action'] is a Trade action, if this is the first action
# to find mean that this devices dont have a confirmation
break
if not dev.trading == 'TradeConfirmed':
txt = 'Some of devices do not have enough to confirm for to do a revoke'
ValidationError(txt)
### End check ###
if ac.t == 'Revoke' and ac.user == g.user:
break
ids = {d.id for d in data['devices']}
lot = data['action'].lot
# import pdb; pdb.set_trace()
self.model = delete_from_trade(lot, ids)
if ac.t == Confirm.t and ac.user == g.user:
real_devices.append(dev)
break
# devices = set(data['devices'])
# without_confirms = set() # set of devs without confirms of user2
data['devices'] = OrderedSet(real_devices)
# if g.user == lot.trade.author:
# for dev in devices:
# ac = dev.last_action_trading
# if ac.type == 'Confirm' and ac.user == g.user:
# without_confirms.add(dev)
# # we need to mark one revoke for every devs
# revoke = Revoke(action=lot.trade, user=g.user, devices=devices)
# db.session.add(revoke)
# if without_confirms:
# confirm_revoke = ConfirmRevoke(
# action=revoke,
# user=g.user,
# devices=without_confirms
# )
# db.session.add(confirm_revoke)
# lot.devices.difference_update(without_confirms)
# lot.trade.devices = lot.devices
# self.model = revoke
class ConfirmRevokeView(ConfirmMixin):
@ -264,46 +270,25 @@ class ConfirmRevokeView(ConfirmMixin):
Model = ConfirmRevoke
def validate(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
"""
real_devices = []
"""All devices need to have the status of revoke."""
if not data['action'].type == 'Revoke':
txt = 'Error: this action is not a revoke action'
ValidationError(txt)
for dev in data['devices']:
actions = copy.copy(dev.actions)
actions.reverse()
for ac in actions:
if ac == data['action']:
# If device have the last action the action for confirm
real_devices.append(dev)
break
if not dev.trading == 'Revoke':
txt = 'Some of devices do not have revoke to confirm'
ValidationError(txt)
if ac.t == 'Revoke' and not ac.user == g.user:
# If device is revoke before you can Confirm now
# and revoke is an action of one other user
real_devices.append(dev)
break
if ac.t == ConfirmRevoke.t and ac.user == g.user:
# If device is confirmed we don't need confirmed again
break
if ac.t == Confirm.t:
# if onwer of trade confirm again before than this user Confirm the
# revoke, then is not possible confirm the revoke
#
# If g.user confirm the trade before do a ConfirmRevoke
# then g.user can not to do the ConfirmRevoke more
break
data['devices'] = OrderedSet(real_devices)
devices = OrderedSet(data['devices'])
data['devices'] = devices
# Change the owner for every devices
trade = data['action']
for dev in data['devices']:
# TODO @cayop this should be the author of confirm actions instead of
# author of trade
dev.owner = trade.author
if hasattr(dev, 'components'):
for c in dev.components:
c.owner = trade.author
# data['action'] == 'Revoke'
trade = data['action'].action
for dev in devices:
dev.reset_owner()
trade.lot.devices.difference_update(devices)

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

@ -1,5 +1,6 @@
import pathlib
import copy
from flask import g
from contextlib import suppress
from fractions import Fraction
from itertools import chain
@ -253,14 +254,100 @@ class Device(Thing):
from ereuse_devicehub.resources.action.models import Price
return self.last_action_of(Price)
@property
def last_action_trading(self):
"""which is the last action trading"""
from ereuse_devicehub.resources.device import states
with suppress(LookupError, ValueError):
return self.last_action_of(*states.Trading.actions())
@property
def trading(self):
"""The actual trading state, or None if no Trade action has
ever been performed to this device."""
"""The trading state, or None if no Trade action has
ever been performed to this device. This extract the posibilities for to do"""
# trade = 'Trade'
confirm = 'Confirm'
need_confirm = 'NeedConfirmation'
double_confirm = 'TradeConfirmed'
revoke = 'Revoke'
revoke_pending = 'RevokePending'
confirm_revoke = 'ConfirmRevoke'
# revoke_confirmed = 'RevokeConfirmed'
# return the correct status of trade depending of the user
##### CASES #####
## User1 == owner of trade (This user have automatic Confirmation)
## =======================
## if the last action is => only allow to do
## ==========================================
## Confirmation not User1 => Revoke
## Confirmation User1 => Revoke
## Revoke not User1 => ConfirmRevoke
## Revoke User1 => RevokePending
## RevokeConfirmation => RevokeConfirmed
##
##
## User2 == Not owner of trade
## =======================
## if the last action is => only allow to do
## ==========================================
## Confirmation not User2 => Confirm
## Confirmation User2 => Revoke
## Revoke not User2 => ConfirmRevoke
## Revoke User2 => RevokePending
## RevokeConfirmation => RevokeConfirmed
ac = self.last_action_trading
if not ac:
return
first_owner = self.which_user_put_this_device_in_trace()
if ac.type == confirm_revoke:
# can to do revoke_confirmed
return confirm_revoke
if ac.type == revoke:
if ac.user == g.user:
# can todo revoke_pending
return revoke_pending
else:
# can to do confirm_revoke
return revoke
if ac.type == confirm:
if not first_owner:
return
if ac.user == first_owner:
if first_owner == g.user:
# can to do revoke
return confirm
else:
# can to do confirm
return need_confirm
else:
# can to do revoke
return double_confirm
@property
def revoke(self):
"""If the actual trading state is an revoke action, this property show
the id of that revoke"""
from ereuse_devicehub.resources.device import states
with suppress(LookupError, ValueError):
action = self.last_action_of(*states.Trading.actions())
return states.Trading(action.__class__)
if action.type == 'Revoke':
return action.id
@property
def confirm_status(self):
"""The actual state of confirmation of one Trade, or None if no Trade action
has ever been performed to this device."""
# TODO @cayop we need implement this functionality
return None
@property
def physical(self):
@ -347,12 +434,37 @@ class Device(Thing):
"""
try:
# noinspection PyTypeHints
actions = self.actions
actions = copy.copy(self.actions)
actions.sort(key=lambda x: x.created)
return next(e for e in reversed(actions) if isinstance(e, types))
except StopIteration:
raise LookupError('{!r} does not contain actions of types {}.'.format(self, types))
def which_user_put_this_device_in_trace(self):
"""which is the user than put this device in this trade"""
actions = copy.copy(self.actions)
actions.sort(key=lambda x: x.created)
actions.reverse()
last_ac = None
# search the automatic Confirm
for ac in actions:
if ac.type == 'Trade':
return last_ac.user
if ac.type == 'Confirm':
last_ac = ac
def change_owner(self, new_user):
"""util for change the owner one device"""
self.owner = new_user
if hasattr(self, 'components'):
for c in self.components:
c.owner = new_user
def reset_owner(self):
"""Change the owner with the user put the device into the trade"""
user = self.which_user_put_this_device_in_trace()
self.change_owner(user)
def _warning_actions(self, actions):
return sorted(ev for ev in actions if ev.severity >= Severity.Warning)

2
ereuse_devicehub/resources/device/schemas.py
View file

@ -51,9 +51,11 @@ class Device(Thing):
rate = NestedOn('Rate', dump_only=True, description=m.Device.rate.__doc__)
price = NestedOn('Price', dump_only=True, description=m.Device.price.__doc__)
trading = EnumField(states.Trading, dump_only=True, description=m.Device.trading.__doc__)
trading = SanitizedStr(dump_only=True, description='')
physical = EnumField(states.Physical, dump_only=True, description=m.Device.physical.__doc__)
traking= EnumField(states.Traking, dump_only=True, description=m.Device.physical.__doc__)
usage = EnumField(states.Usage, dump_only=True, description=m.Device.physical.__doc__)
revoke = UUID(dump_only=True)
physical_possessor = NestedOn('Agent', dump_only=True, data_key='physicalPossessor')
production_date = DateTime('iso',
description=m.Device.updated.comment,

3
ereuse_devicehub/resources/device/states.py
View file

@ -35,6 +35,9 @@ class Trading(State):
"""
Reserved = e.Reserve
Trade = e.Trade
Confirm = e.Confirm
Revoke = e.Revoke
ConfirmRevoke = e.ConfirmRevoke
Cancelled = e.CancelTrade
Sold = e.Sell
Donated = e.Donate

12
ereuse_devicehub/resources/device/views.py
View file

@ -24,6 +24,7 @@ from ereuse_devicehub.resources.device.models import Device, Manufacturer, Compu
from ereuse_devicehub.resources.device.search import DeviceSearch
from ereuse_devicehub.resources.enums import SnapshotSoftware
from ereuse_devicehub.resources.lot.models import LotDeviceDescendants
from ereuse_devicehub.resources.action.models import Trade
from ereuse_devicehub.resources.tag.model import Tag
@ -150,7 +151,16 @@ class DeviceView(View):
)
def query(self, args):
query = Device.query.filter((Device.owner_id == g.user.id)).distinct()
trades = Trade.query.filter(
(Trade.user_from == g.user) | (Trade.user_to == g.user)
).distinct()
trades_dev_ids = {d.id for t in trades for d in t.devices}
query = Device.query.filter(
(Device.owner_id == g.user.id) | (Device.id.in_(trades_dev_ids))
).distinct()
search_p = args.get('search', None)
if search_p:
properties = DeviceSearch.properties

35
ereuse_devicehub/resources/documents/documents.py
View file

@ -1,7 +1,8 @@
import csv
import datetime
import enum
import uuid
import datetime
import pathlib
from collections import OrderedDict
from io import StringIO
from typing import Callable, Iterable, Tuple
@ -18,7 +19,9 @@ from flask.json import jsonify
from teal.cache import cache
from teal.resource import Resource, View
from ereuse_devicehub import auth
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.enums import SessionType
from ereuse_devicehub.resources.action import models as evs
from ereuse_devicehub.resources.device import models as devs
from ereuse_devicehub.resources.deliverynote.models import Deliverynote
@ -314,6 +317,30 @@ class InternalStatsView(DeviceView):
return output
class WbConfDocumentView(DeviceView):
def get(self, wbtype: str):
if not wbtype.lower() in ['usodyrate', 'usodywipe']:
return jsonify('')
data = {'token': self.get_token(),
'host': app.config['DB_HOST'],
'inventory': app.config['DB_SCHEMA']
}
data['erase'] = False
# data['erase'] = True if wbtype == 'usodywipe' else False
env = flask.render_template('documents/wbSettings.ini', **data)
output = make_response(env)
output.headers['Content-Disposition'] = 'attachment; filename=settings.ini'
output.headers['Content-type'] = 'text/plain'
return output
def get_token(self):
tk = [s.token for s in g.user.sessions if s.type == SessionType.Internal][0]
token = auth.Auth.encode(tk)
return token
class DocumentDef(Resource):
__type__ = 'Document'
SCHEMA = None
@ -380,3 +407,9 @@ class DocumentDef(Resource):
auth=app.auth)
actions_view = app.auth.requires_auth(actions_view)
self.add_url_rule('/actions/', defaults=d, view_func=actions_view, methods=get)
wbconf_view = WbConfDocumentView.as_view('WbConfDocumentView',
definition=self,
auth=app.auth)
wbconf_view = app.auth.requires_auth(wbconf_view)
self.add_url_rule('/wbconf/<string:wbtype>', view_func=wbconf_view, methods=get)

17
ereuse_devicehub/resources/documents/templates/documents/wbSettings.ini Normal file
View file

@ -0,0 +1,17 @@
[settings]
DH_TOKEN="{{token}}"
DH_HOST="{{host}}"
DH_DATABASE="{{inventory}}"
DEVICEHUB_URL=https://${DB_HOST}/${DB_DATABASE}/
WB_BENCHMARK = False
WB_STRESS_TEST = 0
WB_SMART_TEST = ""
WB_ERASE = {{erase}}
WB_ERASE_STEPS = 1
WB_ERASE_LEADING_ZEROS = False
WB_DEBUG = True

21
ereuse_devicehub/resources/enums.py
View file

@ -393,3 +393,24 @@ class TransferState(IntEnum):
def __str__(self):
return self.name
@unique
class SessionType(IntEnum):
"""
Enumaration of types of sessions:
* Internal: permanent Session for internal user. This is used in usb of WorkBench.
* External: permanent Session for external users. This is used in usb of WorkBench.
* Session: This is used for keep more than one session in the app frontend.
Devicehub specially raises user awareness when an action
has a Severity of ``Warning`` or greater.
"""
Internal = 0
External = 1
Session = 2
def __str__(self):
return self.name

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

@ -99,6 +99,10 @@ class Lot(Thing):
def descendants(self):
return self.descendantsq(self.id)
@property
def is_temporary(self):
return False if self.trade else True
@classmethod
def descendantsq(cls, id):
_id = UUIDLtree.convert(id)

2
ereuse_devicehub/resources/lot/schemas.py
View file

@ -4,6 +4,7 @@ from teal.marshmallow import SanitizedStr, URL, EnumField
from ereuse_devicehub.marshmallow import NestedOn
from ereuse_devicehub.resources.deliverynote import schemas as s_deliverynote
from ereuse_devicehub.resources.device import schemas as s_device
from ereuse_devicehub.resources.action import schemas as s_action
from ereuse_devicehub.resources.enums import TransferState
from ereuse_devicehub.resources.lot import models as m
from ereuse_devicehub.resources.models import STR_SIZE
@ -26,3 +27,4 @@ class Lot(Thing):
transfer_state = EnumField(TransferState, description=m.Lot.transfer_state.comment)
receiver_address = SanitizedStr(validate=f.validate.Length(max=42))
deliverynote = NestedOn(s_deliverynote.Deliverynote, dump_only=True)
trade = NestedOn(s_action.Trade, dump_only=True)

75
ereuse_devicehub/resources/lot/views.py
View file

@ -12,9 +12,8 @@ from teal.resource import View
from ereuse_devicehub.db import db
from ereuse_devicehub.query import things_response
from ereuse_devicehub.resources.deliverynote.models import Deliverynote
from ereuse_devicehub.resources.device.models import Device, Computer
from ereuse_devicehub.resources.action.models import Confirm, Revoke
from ereuse_devicehub.resources.action.models import Trade, Confirm, Revoke, ConfirmRevoke
from ereuse_devicehub.resources.lot.models import Lot, Path
@ -98,9 +97,9 @@ class LotView(View):
return jsonify(ret)
def visibility_filter(self, query):
query = query.outerjoin(Deliverynote) \
.filter(or_(Deliverynote.receiver_address == g.user.email,
Deliverynote.supplier_email == g.user.email,
query = query.outerjoin(Trade) \
.filter(or_(Trade.user_from == g.user,
Trade.user_to == g.user,
Lot.owner_id == g.user.id))
return query
@ -109,7 +108,7 @@ class LotView(View):
return query
def delete(self, id):
lot = Lot.query.filter_by(id=id).one()
lot = Lot.query.filter_by(id=id, owner=g.user).one()
lot.delete()
db.session.commit()
return Response(status=204)
@ -253,20 +252,62 @@ class LotDeviceView(LotBaseChildrenView):
if not ids.issubset({x.id for x in lot.devices}):
return
users = [g.user.id]
if lot.trade:
# all users involved in the trade action can modify the lot
trade_users = [lot.trade.user_from.id, lot.trade.user_to.id]
if g.user in trade_users:
users = trade_users
return delete_from_trade(lot, ids)
if not g.user in lot.owner:
txt = 'This is not your trade'
raise ma.ValidationError(txt)
devices = set(Device.query.filter(Device.id.in_(ids)).filter(
Device.owner_id.in_(users)))
Device.owner_id.in_(g.user.id)))
lot.devices.difference_update(devices)
if lot.trade:
lot.trade.devices = lot.devices
if g.user in [lot.trade.user_from, lot.trade.user_to]:
revoke = Revoke(action=lot.trade, user=g.user, devices=devices)
db.session.add(revoke)
def delete_from_trade(lot: Lot, ids: Set[int]):
users = [lot.trade.user_from.id, lot.trade.user_to.id]
if not g.user.id in users:
# theoretically this case is impossible
txt = 'This is not your trade'
raise ma.ValidationError(txt)
# import pdb; pdb.set_trace()
devices = set(Device.query.filter(Device.id.in_(ids)).filter(
Device.owner_id.in_(users)))
# Now we need to know which devices we need extract of the lot
without_confirms = set() # set of devs without confirms of user2
# if the trade need confirmation, then extract all devs than
# have only one confirmation and is from the same user than try to do
# now the revoke action
if lot.trade.confirm:
for dev in devices:
# if have only one confirmation
# then can be revoked and deleted of the lot
# Confirm of dev.trading mean that there are only one confirmation
# and the first user than put this device in trade is the actual g.user
if dev.trading == 'Confirm':
without_confirms.add(dev)
dev.reset_owner()
# we need to mark one revoke for every devs
revoke = Revoke(action=lot.trade, user=g.user, devices=devices)
db.session.add(revoke)
if not lot.trade.confirm:
# if the trade is with phantom account
without_confirms = devices
if without_confirms:
confirm_revoke = ConfirmRevoke(
action=revoke,
user=g.user,
devices=without_confirms
)
db.session.add(confirm_revoke)
lot.devices.difference_update(without_confirms)
lot.trade.devices = lot.devices
return revoke

4
ereuse_devicehub/resources/user/__init__.py
View file

@ -7,7 +7,7 @@ from teal.resource import Converters, Resource
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.user import schemas
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.user.views import UserView, login
from ereuse_devicehub.resources.user.views import UserView, login, logout
class UserDef(Resource):
@ -23,6 +23,8 @@ class UserDef(Resource):
super().__init__(app, import_name, static_folder, static_url_path, template_folder,
url_prefix, subdomain, url_defaults, root_path, cli_commands)
self.add_url_rule('/login/', view_func=login, methods={'POST'})
logout_view = app.auth.requires_auth(logout)
self.add_url_rule('/logout/', view_func=logout_view, methods={'GET'})
@argument('email')
@option('-i', '--inventory',

19
ereuse_devicehub/resources/user/models.py
View file

@ -2,13 +2,15 @@ from uuid import uuid4
from citext import CIText
from flask import current_app as app
from sqlalchemy import Column, Boolean
from sqlalchemy import Column, Boolean, BigInteger, Sequence
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy_utils import EmailType, PasswordType
from teal.db import IntEnum
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.inventory.model import Inventory
from ereuse_devicehub.resources.models import STR_SIZE, Thing
from ereuse_devicehub.resources.enums import SessionType
class User(Thing):
@ -63,3 +65,18 @@ class UserInventory(db.Model):
__table_args__ = {'schema': 'common'}
user_id = db.Column(db.UUID(as_uuid=True), db.ForeignKey(User.id), primary_key=True)
inventory_id = db.Column(db.Unicode(), db.ForeignKey(Inventory.id), primary_key=True)
class Session(Thing):
__table_args__ = {'schema': 'common'}
id = Column(BigInteger, Sequence('device_seq'), primary_key=True)
expired = Column(BigInteger, default=0)
token = Column(UUID(as_uuid=True), default=uuid4, unique=True, nullable=False)
type = Column(IntEnum(SessionType), default=SessionType.Internal, nullable=False)
user_id = db.Column(db.UUID(as_uuid=True), db.ForeignKey(User.id))
user = db.relationship(User,
backref=db.backref('sessions', lazy=True, collection_class=set),
collection_class=set)
def __str__(self) -> str:
return '{0.token}'.format(self)

4
ereuse_devicehub/resources/user/schemas.py
View file

@ -9,6 +9,10 @@ from ereuse_devicehub.resources.inventory.schema import Inventory
from ereuse_devicehub.resources.schemas import Thing
class Session(Thing):
token = String(dump_only=True)
class User(Thing):
id = UUID(dump_only=True)
email = Email(required=True)

13
ereuse_devicehub/resources/user/views.py
View file

@ -1,11 +1,12 @@
from uuid import UUID
from uuid import UUID, uuid4
from flask import g, request
from flask.json import jsonify
from teal.resource import View
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.user.exceptions import WrongCredentials
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.user.schemas import User as UserS
class UserView(View):
@ -24,3 +25,11 @@ def login():
return schema_with_token.jsonify(user)
else:
raise WrongCredentials()
def logout():
# We use custom schema as we only want to parse a subset of user
g.user.token = uuid4()
db.session.add(g.user)
db.session.commit()
return jsonify('Ok')

6
tests/conftest.py
View file

@ -17,6 +17,8 @@ from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.agent.models import Person
from ereuse_devicehub.resources.tag import Tag
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.user.models import Session
from ereuse_devicehub.resources.enums import SessionType
STARTT = datetime(year=2000, month=1, day=1, hour=1)
"""A dummy starting time to use in tests."""
@ -112,7 +114,11 @@ def user2(app: Devicehub) -> UserClient:
def create_user(email='foo@foo.com', password='foo') -> User:
user = User(email=email, password=password)
user.individuals.add(Person(name='Timmy'))
session_external = Session(user=user, type=SessionType.External)
session_internal = Session(user=user, type=SessionType.Internal)
db.session.add(user)
db.session.add(session_internal)
db.session.add(session_external)
db.session.commit()
return user

1
tests/files/2021-5-4-13-41_time_out_test_datastorage.yaml Normal file
View file

File diff suppressed because one or more lines are too long

2
tests/files/basic.csv
View file

File diff suppressed because one or more lines are too long

4
tests/files/proposal_extended_csv_report.csv
View file

File diff suppressed because one or more lines are too long

648
tests/test_action.py
View file

@ -387,7 +387,7 @@ def test_live_without_TestDataStorage(user: UserClient, client: Client, app: Dev
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
@ -420,7 +420,7 @@ def test_live_without_hdd_1(user: UserClient, client: Client, app: Devicehub):
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
@ -451,7 +451,7 @@ def test_live_without_hdd_2(user: UserClient, client: Client, app: Devicehub):
acer['components'] = components
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
@ -482,7 +482,7 @@ def test_live_without_hdd_3(user: UserClient, client: Client, app: Devicehub):
acer['components'] = components
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
@ -515,7 +515,7 @@ def test_live_with_hdd_with_old_time(user: UserClient, client: Client, app: Devi
acer = file('acer.happy.battery.snapshot')
snapshot, _ = user.post(acer, res=models.Snapshot)
device_id = snapshot['device']['id']
db_device = Device.query.filter_by(id=1).one()
db_device = Device.query.filter_by(id=device_id).one()
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
"devices": [device_id], "description": "aaa",
"finalUserCode": "abcdefjhi",
@ -765,6 +765,227 @@ def test_trade_endpoint(user: UserClient, user2: UserClient):
device2, _ = user2.get(res=Device, item=device['id'])
assert device2['id'] == device['id']
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_offer_without_to(user: UserClient):
"""Test one offer with automatic confirmation and without user to"""
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
device = Device.query.filter_by(id=snapshot['device']['id']).one()
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
user.post({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=[('id', device.id)])
# check the owner of the device
assert device.owner.email == user.email
for c in device.components:
assert c.owner.email == user.email
request_post = {
'type': 'Trade',
'devices': [device.id],
'userFromEmail': user.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'documentID': '1',
'lot': lot['id'],
'confirms': False,
'code': 'MAX'
}
user.post(res=models.Action, data=request_post)
trade = models.Trade.query.one()
assert device in trade.devices
# assert trade.confirm_transfer
users = [ac.user for ac in trade.acceptances]
assert trade.user_to == device.owner
assert request_post['code'].lower() in device.owner.email
assert device.owner.active == False
assert device.owner.phantom == True
assert trade.user_to in users
assert trade.user_from in users
assert device.owner.email != user.email
for c in device.components:
assert c.owner.email != user.email
# check if the user_from is owner of the devices
request_post = {
'type': 'Trade',
'devices': [device.id],
'userFromEmail': user.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'documentID': '1',
'lot': lot['id'],
'confirms': False,
'code': 'MAX'
}
user.post(res=models.Action, data=request_post, status=422)
trade = models.Trade.query.one()
# Check if the new phantom account is reused and not duplicated
computer = file('1-device-with-components.snapshot')
snapshot2, _ = user.post(computer, res=models.Snapshot)
device2 = Device.query.filter_by(id=snapshot2['device']['id']).one()
lot2 = Lot('MyLot2')
lot2.owner_id = user.user['id']
lot2.devices.add(device2)
db.session.add(lot2)
db.session.flush()
request_post2 = {
'type': 'Trade',
'devices': [device2.id],
'userFromEmail': user.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'documentID': '1',
'lot': lot2.id,
'confirms': False,
'code': 'MAX'
}
user.post(res=models.Action, data=request_post2)
assert User.query.filter_by(email=device.owner.email).count() == 1
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_offer_without_from(user: UserClient, user2: UserClient):
"""Test one offer without confirmation and without user from"""
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
lot = Lot('MyLot')
lot.owner_id = user.user['id']
device = Device.query.filter_by(id=snapshot['device']['id']).one()
# check the owner of the device
assert device.owner.email == user.email
assert device.owner.email != user2.email
lot.devices.add(device)
db.session.add(lot)
db.session.flush()
request_post = {
'type': 'Trade',
'devices': [device.id],
'userToEmail': user2.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'documentID': '1',
'lot': lot.id,
'confirms': False,
'code': 'MAX'
}
action, _ = user2.post(res=models.Action, data=request_post, status=422)
request_post['userToEmail'] = user.email
action, _ = user.post(res=models.Action, data=request_post)
trade = models.Trade.query.one()
phantom_user = trade.user_from
assert request_post['code'].lower() in phantom_user.email
assert phantom_user.active == False
assert phantom_user.phantom == True
# assert trade.confirm_transfer
users = [ac.user for ac in trade.acceptances]
assert trade.user_to in users
assert trade.user_from in users
assert user.email in trade.devices[0].owner.email
assert device.owner.email != user2.email
assert device.owner.email == user.email
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_offer_without_users(user: UserClient):
"""Test one offer with doble confirmation"""
user2 = User(email='baz@baz.cxm', password='baz')
user2.individuals.add(Person(name='Tommy'))
db.session.add(user2)
db.session.commit()
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
lot = Lot('MyLot')
lot.owner_id = user.user['id']
device = Device.query.filter_by(id=snapshot['device']['id']).one()
lot.devices.add(device)
db.session.add(lot)
db.session.flush()
request_post = {
'type': 'Trade',
'devices': [device.id],
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'documentID': '1',
'lot': lot.id,
'confirms': False,
'code': 'MAX'
}
action, response = user.post(res=models.Action, data=request_post, status=422)
txt = 'you need one user from or user to for to do a offer'
assert txt in action['message']['_schema']
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_offer(user: UserClient):
"""Test one offer with doble confirmation"""
user2 = User(email='baz@baz.cxm', password='baz')
user2.individuals.add(Person(name='Tommy'))
db.session.add(user2)
db.session.commit()
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
lot = Lot('MyLot')
lot.owner_id = user.user['id']
device = Device.query.filter_by(id=snapshot['device']['id']).one()
assert device.owner.email == user.email
assert device.owner.email != user2.email
lot.devices.add(device)
db.session.add(lot)
db.session.flush()
request_post = {
'type': 'Trade',
'devices': [],
'userFromEmail': user.email,
'userToEmail': user2.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'documentID': '1',
'lot': lot.id,
'confirms': True,
}
action, _ = user.post(res=models.Action, data=request_post)
# no there are transfer of devices
assert device.owner.email == user.email
assert device.owner.email != user2.email
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_offer_without_devices(user: UserClient):
"""Test one offer with doble confirmation"""
user2 = User(email='baz@baz.cxm', password='baz')
user2.individuals.add(Person(name='Tommy'))
db.session.add(user2)
db.session.commit()
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
request_post = {
'type': 'Trade',
'devices': [],
'userFromEmail': user.email,
'userToEmail': user2.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'documentID': '1',
'lot': lot['id'],
'confirms': True,
}
user.post(res=models.Action, data=request_post)
# no there are transfer of devices
>>>>>>> feature/endpoint-confirm
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.auth_app_context.__name__)
def test_price_custom():
@ -815,3 +1036,420 @@ def test_erase_physical():
db.session.add(erasure)
db.session.commit()
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_endpoint_confirm(user: UserClient, user2: UserClient):
"""Check the normal creation and visualization of one confirmation trade"""
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
device_id = snapshot['device']['id']
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
user.post({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=[('id', device_id)])
request_post = {
'type': 'Trade',
'devices': [device_id],
'userFromEmail': user.email,
'userToEmail': user2.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'documentID': '1',
'lot': lot['id'],
'confirms': True,
}
user.post(res=models.Action, data=request_post)
trade = models.Trade.query.one()
assert trade.devices[0].owner.email == user.email
request_confirm = {
'type': 'Confirm',
'action': trade.id,
'devices': [device_id]
}
user2.post(res=models.Action, data=request_confirm)
user2.post(res=models.Action, data=request_confirm, status=422)
assert len(trade.acceptances) == 2
assert trade.devices[0].owner.email == user2.email
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_confirm_revoke(user: UserClient, user2: UserClient):
"""Check the normal revoke of one confirmation"""
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
device_id = snapshot['device']['id']
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
user.post({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=[('id', device_id)])
request_post = {
'type': 'Trade',
'devices': [device_id],
'userFromEmail': user.email,
'userToEmail': user2.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'documentID': '1',
'lot': lot['id'],
'confirms': True,
}
user.post(res=models.Action, data=request_post)
trade = models.Trade.query.one()
request_confirm = {
'type': 'Confirm',
'action': trade.id,
'devices': [device_id]
}
request_revoke = {
'type': 'Revoke',
'action': trade.id,
'devices': [device_id],
}
# Normal confirmation
user2.post(res=models.Action, data=request_confirm)
# Normal revoke
user2.post(res=models.Action, data=request_revoke)
# Error for try duplicate revoke
user2.post(res=models.Action, data=request_revoke, status=422)
assert len(trade.acceptances) == 3
# You can not to do one confirmation next of one revoke
user2.post(res=models.Action, data=request_confirm, status=422)
assert len(trade.acceptances) == 3
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_usecase_confirmation(user: UserClient, user2: UserClient):
"""Example of one usecase about confirmation"""
# the pRp (manatest_usecase_confirmationger) creates a temporary lot
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
# The manager add 7 device into the lot
snap1, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
snap2, _ = user.post(file('acer.happy.battery.snapshot'), res=models.Snapshot)
snap3, _ = user.post(file('asus-1001pxd.snapshot'), res=models.Snapshot)
snap4, _ = user.post(file('desktop-9644w8n-lenovo-0169622.snapshot'), res=models.Snapshot)
snap5, _ = user.post(file('laptop-hp_255_g3_notebook-hewlett-packard-cnd52270fw.snapshot'), res=models.Snapshot)
snap6, _ = user.post(file('1-device-with-components.snapshot'), res=models.Snapshot)
snap7, _ = user.post(file('asus-eee-1000h.snapshot.11'), res=models.Snapshot)
snap8, _ = user.post(file('complete.export.snapshot'), res=models.Snapshot)
snap9, _ = user.post(file('real-hp-quad-core.snapshot.11'), res=models.Snapshot)
snap10, _ = user.post(file('david.lshw.snapshot'), res=models.Snapshot)
devices = [('id', snap1['device']['id']),
('id', snap2['device']['id']),
('id', snap3['device']['id']),
('id', snap4['device']['id']),
('id', snap5['device']['id']),
('id', snap6['device']['id']),
('id', snap7['device']['id']),
('id', snap8['device']['id']),
('id', snap9['device']['id']),
('id', snap10['device']['id']),
]
lot, _ = user.post({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=devices[:7])
# the manager shares the temporary lot with the SCRAP as an incoming lot
# for the SCRAP to confirm it
request_post = {
'type': 'Trade',
'devices': [],
'userFromEmail': user2.email,
'userToEmail': user.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'documentID': '1',
'lot': lot['id'],
'confirms': True,
}
user.post(res=models.Action, data=request_post)
trade = models.Trade.query.one()
# l_after, _ = user.get(res=Lot, item=lot['id'])
# import pdb; pdb.set_trace()
# the SCRAP confirms 3 of the 10 devices in its outgoing lot
request_confirm = {
'type': 'Confirm',
'action': trade.id,
'devices': [snap1['device']['id'], snap2['device']['id'], snap3['device']['id']]
}
assert trade.devices[0].actions[-2].t == 'Trade'
assert trade.devices[0].actions[-1].t == 'Confirm'
assert trade.devices[0].actions[-1].user == trade.user_to
user2.post(res=models.Action, data=request_confirm)
assert trade.devices[0].actions[-1].t == 'Confirm'
assert trade.devices[0].actions[-1].user == trade.user_from
n_actions = len(trade.devices[0].actions)
# check validation error
request_confirm = {
'type': 'Confirm',
'action': trade.id,
'devices': [
snap10['device']['id']
]
}
user2.post(res=models.Action, data=request_confirm, status=422)
# The manager add 3 device more into the lot
lot, _ = user.post({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=devices[7:])
assert trade.devices[-1].actions[-2].t == 'Trade'
assert trade.devices[-1].actions[-1].t == 'Confirm'
assert trade.devices[-1].actions[-1].user == trade.user_to
assert len(trade.devices[0].actions) == n_actions
# the SCRAP confirms the rest of devices
request_confirm = {
'type': 'Confirm',
'action': trade.id,
'devices': [
snap1['device']['id'],
snap2['device']['id'],
snap3['device']['id'],
snap4['device']['id'],
snap5['device']['id'],
snap6['device']['id'],
snap7['device']['id'],
snap8['device']['id'],
snap9['device']['id'],
snap10['device']['id']
]
}
user2.post(res=models.Action, data=request_confirm)
assert trade.devices[-1].actions[-3].t == 'Trade'
assert trade.devices[-1].actions[-1].t == 'Confirm'
assert trade.devices[-1].actions[-1].user == trade.user_from
assert len(trade.devices[0].actions) == n_actions
# The manager remove one device of the lot and automaticaly
# is create one revoke action
device_10 = trade.devices[-1]
lot, _ = user.delete({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=devices[-1:], status=200)
assert len(trade.lot.devices) == len(trade.devices) == 9
assert not device_10 in trade.devices
assert device_10.actions[-1].t == 'Revoke'
lot, _ = user.delete({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=devices[-1:], status=200)
assert device_10.actions[-1].t == 'Revoke'
assert device_10.actions[-2].t == 'Confirm'
# the SCRAP confirms the revoke action
request_confirm_revoke = {
'type': 'ConfirmRevoke',
'action': device_10.actions[-1].id,
'devices': [
snap10['device']['id']
]
}
user2.post(res=models.Action, data=request_confirm_revoke)
assert device_10.actions[-1].t == 'ConfirmRevoke'
assert device_10.actions[-2].t == 'Revoke'
# check validation error
request_confirm_revoke = {
'type': 'ConfirmRevoke',
'action': device_10.actions[-1].id,
'devices': [
snap9['device']['id']
]
}
user2.post(res=models.Action, data=request_confirm_revoke, status=422)
# The manager add again device_10
assert len(trade.devices) == 9
lot, _ = user.post({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=devices[-1:])
assert device_10.actions[-1].t == 'Confirm'
assert device_10 in trade.devices
assert len(trade.devices) == 10
# the SCRAP confirms the action trade for device_10
request_reconfirm = {
'type': 'Confirm',
'action': trade.id,
'devices': [
snap10['device']['id']
]
}
# import pdb; pdb.set_trace()
user2.post(res=models.Action, data=request_reconfirm)
assert device_10.actions[-1].t == 'Confirm'
assert device_10.actions[-1].user == trade.user_from
assert device_10.actions[-2].t == 'Confirm'
assert device_10.actions[-2].user == trade.user_to
assert device_10.actions[-3].t == 'ConfirmRevoke'
assert len(device_10.actions) == 13
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_confirmRevoke(user: UserClient, user2: UserClient):
"""Example of one usecase about confirmation"""
# the pRp (manatest_usecase_confirmationger) creates a temporary lot
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
# The manager add 7 device into the lot
snap1, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
snap2, _ = user.post(file('acer.happy.battery.snapshot'), res=models.Snapshot)
snap3, _ = user.post(file('asus-1001pxd.snapshot'), res=models.Snapshot)
snap4, _ = user.post(file('desktop-9644w8n-lenovo-0169622.snapshot'), res=models.Snapshot)
snap5, _ = user.post(file('laptop-hp_255_g3_notebook-hewlett-packard-cnd52270fw.snapshot'), res=models.Snapshot)
snap6, _ = user.post(file('1-device-with-components.snapshot'), res=models.Snapshot)
snap7, _ = user.post(file('asus-eee-1000h.snapshot.11'), res=models.Snapshot)
snap8, _ = user.post(file('complete.export.snapshot'), res=models.Snapshot)
snap9, _ = user.post(file('real-hp-quad-core.snapshot.11'), res=models.Snapshot)
snap10, _ = user.post(file('david.lshw.snapshot'), res=models.Snapshot)
devices = [('id', snap1['device']['id']),
('id', snap2['device']['id']),
('id', snap3['device']['id']),
('id', snap4['device']['id']),
('id', snap5['device']['id']),
('id', snap6['device']['id']),
('id', snap7['device']['id']),
('id', snap8['device']['id']),
('id', snap9['device']['id']),
('id', snap10['device']['id']),
]
lot, _ = user.post({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=devices)
# the manager shares the temporary lot with the SCRAP as an incoming lot
# for the CRAP to confirm it
request_post = {
'type': 'Trade',
'devices': [],
'userFromEmail': user2.email,
'userToEmail': user.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'documentID': '1',
'lot': lot['id'],
'confirms': True,
}
user.post(res=models.Action, data=request_post)
trade = models.Trade.query.one()
# the SCRAP confirms all of devices
request_confirm = {
'type': 'Confirm',
'action': trade.id,
'devices': [
snap1['device']['id'],
snap2['device']['id'],
snap3['device']['id'],
snap4['device']['id'],
snap5['device']['id'],
snap6['device']['id'],
snap7['device']['id'],
snap8['device']['id'],
snap9['device']['id'],
snap10['device']['id']
]
}
user2.post(res=models.Action, data=request_confirm)
assert trade.devices[-1].actions[-3].t == 'Trade'
assert trade.devices[-1].actions[-1].t == 'Confirm'
assert trade.devices[-1].actions[-1].user == trade.user_from
# The manager remove one device of the lot and automaticaly
# is create one revoke action
device_10 = trade.devices[-1]
lot, _ = user.delete({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=devices[-1:], status=200)
assert len(trade.lot.devices) == len(trade.devices) == 9
assert not device_10 in trade.devices
assert device_10.actions[-1].t == 'Revoke'
lot, _ = user.delete({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=devices[-1:], status=200)
assert device_10.actions[-1].t == 'Revoke'
assert device_10.actions[-2].t == 'Confirm'
# The manager add again device_10
assert len(trade.devices) == 9
lot, _ = user.post({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=devices[-1:])
assert device_10.actions[-1].t == 'Confirm'
assert device_10 in trade.devices
assert len(trade.devices) == 10
# the SCRAP confirms the revoke action
request_confirm_revoke = {
'type': 'ConfirmRevoke',
'action': device_10.actions[-2].id,
'devices': [
snap10['device']['id']
]
}
# check validation error
user2.post(res=models.Action, data=request_confirm_revoke, status=422)
# the SCRAP confirms the action trade for device_10
request_reconfirm = {
'type': 'Confirm',
'action': trade.id,
'devices': [
snap10['device']['id']
]
}
user2.post(res=models.Action, data=request_reconfirm)
assert device_10.actions[-1].t == 'Confirm'
assert device_10.actions[-1].user == trade.user_from
assert device_10.actions[-2].t == 'Confirm'
assert device_10.actions[-2].user == trade.user_to
assert device_10.actions[-3].t == 'Revoke'
>>>>>>> feature/endpoint-confirm

4
tests/test_basic.py
View file

@ -40,6 +40,7 @@ def test_api_docs(client: Client):
'/documents/erasures/',
'/documents/devices/',
'/documents/stamps/',
'/documents/wbconf/{wbtype}',
'/documents/internalstats/',
'/documents/stock/',
'/documents/check/',
@ -56,7 +57,8 @@ def test_api_docs(client: Client):
'/tags/{tag_id}/device/{device_id}',
'/trade-documents/',
'/users/',
'/users/login/'
'/users/login/',
'/users/logout/',
# '/devices/{dev1_id}/merge/{dev2_id}',
# '/batteries/{dev1_id}/merge/{dev2_id}',
# '/bikes/{dev1_id}/merge/{dev2_id}',

133
tests/test_device.py
View file

@ -65,13 +65,13 @@ def test_device_model():
gcard = d.GraphicCard.query.one()
db.session.delete(pc)
db.session.flush()
assert pc.id == 1
assert pc.id == 3
assert d.Desktop.query.first() is None
db.session.commit()
assert d.Desktop.query.first() is None
assert network_adapter.id == 2
assert network_adapter.id == 4
assert d.NetworkAdapter.query.first() is not None, 'We removed the network adaptor'
assert gcard.id == 3, 'We should still hold a reference to a zombie graphic card'
assert gcard.id == 5, 'We should still hold a reference to a zombie graphic card'
assert d.GraphicCard.query.first() is None, 'We should have deleted it –it was inside the pc'
@ -396,74 +396,73 @@ def test_sync_execute_register_mismatch_between_tags_and_hid():
@pytest.mark.mvp
def test_get_device(app: Devicehub, user: UserClient):
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_get_device(user: UserClient):
"""Checks GETting a d.Desktop with its components."""
with app.app_context():
pc = d.Desktop(model='p1mo',
manufacturer='p1ma',
serial_number='p1s',
chassis=ComputerChassis.Tower,
owner_id=user.user['id'])
pc.components = OrderedSet([
d.NetworkAdapter(model='c1mo', manufacturer='c1ma', serial_number='c1s',
owner_id=user.user['id']),
d.GraphicCard(model='c2mo', manufacturer='c2ma', memory=1500, owner_id=user.user['id'])
])
db.session.add(pc)
# todo test is an abstract class. replace with another one
db.session.add(TestConnectivity(device=pc,
severity=Severity.Info,
agent=Person(name='Timmy'),
author=User(email='bar@bar.com')))
db.session.commit()
devicehub_id = pc.devicehub_id
pc, _ = user.get(res=d.Device, item=devicehub_id)
assert len(pc['actions']) == 1
assert pc['actions'][0]['type'] == 'TestConnectivity'
assert pc['actions'][0]['device'] == 1
assert pc['actions'][0]['severity'] == 'Info'
assert UUID(pc['actions'][0]['author'])
assert 'actions_components' not in pc, 'actions_components are internal use only'
assert 'actions_one' not in pc, 'they are internal use only'
assert 'author' not in pc
assert tuple(c['id'] for c in pc['components']) == (2, 3)
assert pc['hid'] == 'desktop-p1ma-p1mo-p1s'
assert pc['model'] == 'p1mo'
assert pc['manufacturer'] == 'p1ma'
assert pc['serialNumber'] == 'p1s'
assert pc['type'] == d.Desktop.t
pc = d.Desktop(model='p1mo',
manufacturer='p1ma',
serial_number='p1s',
chassis=ComputerChassis.Tower,
owner_id=user.user['id'])
pc.components = OrderedSet([
d.NetworkAdapter(model='c1mo', manufacturer='c1ma', serial_number='c1s',
owner_id=user.user['id']),
d.GraphicCard(model='c2mo', manufacturer='c2ma', memory=1500, owner_id=user.user['id'])
])
db.session.add(pc)
# todo test is an abstract class. replace with another one
db.session.add(TestConnectivity(device=pc,
severity=Severity.Info,
agent=Person(name='Timmy'),
author=User(email='bar@bar.com')))
db.session.commit()
pc_api, _ = user.get(res=d.Device, item=pc.devicehub_id)
assert len(pc_api['actions']) == 1
assert pc_api['actions'][0]['type'] == 'TestConnectivity'
assert pc_api['actions'][0]['device'] == pc.id
assert pc_api['actions'][0]['severity'] == 'Info'
assert UUID(pc_api['actions'][0]['author'])
assert 'actions_components' not in pc_api, 'actions_components are internal use only'
assert 'actions_one' not in pc_api, 'they are internal use only'
assert 'author' not in pc_api
assert tuple(c['id'] for c in pc_api['components']) == tuple(c.id for c in pc.components)
assert pc_api['hid'] == 'desktop-p1ma-p1mo-p1s'
assert pc_api['model'] == 'p1mo'
assert pc_api['manufacturer'] == 'p1ma'
assert pc_api['serialNumber'] == 'p1s'
assert pc_api['type'] == d.Desktop.t
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_get_devices(app: Devicehub, user: UserClient):
"""Checks GETting multiple devices."""
with app.app_context():
pc = d.Desktop(model='p1mo',
manufacturer='p1ma',
serial_number='p1s',
chassis=ComputerChassis.Tower,
owner_id=user.user['id'])
pc.components = OrderedSet([
d.NetworkAdapter(model='c1mo', manufacturer='c1ma', serial_number='c1s',
owner_id=user.user['id']),
d.GraphicCard(model='c2mo', manufacturer='c2ma', memory=1500,
owner_id=user.user['id'])
])
pc1 = d.Desktop(model='p2mo',
manufacturer='p2ma',
serial_number='p2s',
chassis=ComputerChassis.Tower,
owner_id=user.user['id'])
pc2 = d.Laptop(model='p3mo',
manufacturer='p3ma',
serial_number='p3s',
chassis=ComputerChassis.Netbook,
owner_id=user.user['id'])
db.session.add_all((pc, pc1, pc2))
db.session.commit()
pc = d.Desktop(model='p1mo',
manufacturer='p1ma',
serial_number='p1s',
chassis=ComputerChassis.Tower,
owner_id=user.user['id'])
pc.components = OrderedSet([
d.NetworkAdapter(model='c1mo', manufacturer='c1ma', serial_number='c1s',
owner_id=user.user['id']),
d.GraphicCard(model='c2mo', manufacturer='c2ma', memory=1500,
owner_id=user.user['id'])
])
pc1 = d.Desktop(model='p2mo',
manufacturer='p2ma',
serial_number='p2s',
chassis=ComputerChassis.Tower,
owner_id=user.user['id'])
pc2 = d.Laptop(model='p3mo',
manufacturer='p3ma',
serial_number='p3s',
chassis=ComputerChassis.Netbook,
owner_id=user.user['id'])
db.session.add_all((pc, pc1, pc2))
db.session.commit()
devices, _ = user.get(res=d.Device)
assert tuple(dev['id'] for dev in devices['items']) == (1, 2, 3, 4, 5)
ids = (pc.id, pc1.id, pc2.id, pc.components[0].id, pc.components[1].id)
assert tuple(dev['id'] for dev in devices['items']) == ids
assert tuple(dev['type'] for dev in devices['items']) == (
d.Desktop.t, d.Desktop.t, d.Laptop.t, d.NetworkAdapter.t, d.GraphicCard.t
)
@ -536,7 +535,7 @@ def test_device_properties_format(app: Devicehub, user: UserClient):
user.post(file('asus-eee-1000h.snapshot.11'), res=m.Snapshot)
with app.app_context():
pc = d.Laptop.query.one() # type: d.Laptop
assert format(pc) == 'Laptop 1: model 1000h, S/N 94oaaq021116'
assert format(pc) == 'Laptop 3: model 1000h, S/N 94oaaq021116'
assert format(pc, 't') == 'Netbook 1000h'
assert format(pc, 's') == '(asustek computer inc.) S/N 94OAAQ021116'
assert pc.ram_size == 1024
@ -544,12 +543,12 @@ def test_device_properties_format(app: Devicehub, user: UserClient):
assert pc.graphic_card_model == 'mobile 945gse express integrated graphics controller'
assert pc.processor_model == 'intel atom cpu n270 @ 1.60ghz'
net = next(c for c in pc.components if isinstance(c, d.NetworkAdapter))
assert format(net) == 'NetworkAdapter 2: model ar8121/ar8113/ar8114 ' \
assert format(net) == 'NetworkAdapter 4: model ar8121/ar8113/ar8114 ' \
'gigabit or fast ethernet, S/N 00:24:8c:7f:cf:2d'
assert format(net, 't') == 'NetworkAdapter ar8121/ar8113/ar8114 gigabit or fast ethernet'
assert format(net, 's') == 'qualcomm atheros 00:24:8C:7F:CF:2D – 100 Mbps'
hdd = next(c for c in pc.components if isinstance(c, d.DataStorage))
assert format(hdd) == 'HardDrive 7: model st9160310as, S/N 5sv4tqa6'
assert format(hdd) == 'HardDrive 9: model st9160310as, S/N 5sv4tqa6'
assert format(hdd, 't') == 'HardDrive st9160310as'
assert format(hdd, 's') == 'seagate 5SV4TQA6 – 152 GB'

10
tests/test_device_find.py
View file

@ -176,10 +176,10 @@ def test_device_query_filter_lots(user: UserClient):
@pytest.mark.mvp
def test_device_query(user: UserClient):
"""Checks result of inventory."""
snap, _ = user.post(conftest.file('basic.snapshot'), res=Snapshot)
snapshot, _ = user.post(conftest.file('basic.snapshot'), res=Snapshot)
i, _ = user.get(res=Device)
assert i['url'] == '/devices/'
assert i['items'][0]['url'] == '/devices/%s' % snap['device']['devicehubID']
assert i['items'][0]['url'] == '/devices/%s' % snapshot['device']['devicehubID']
pc = next(d for d in i['items'] if d['type'] == 'Desktop')
assert len(pc['actions']) == 4
assert len(pc['components']) == 3
@ -241,16 +241,16 @@ def test_device_search_regenerate_table(app: DeviceSearch, user: UserClient):
@pytest.mark.mvp
def test_device_query_search(user: UserClient):
# todo improve
user.post(file('basic.snapshot'), res=Snapshot)
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)
user.post(file('computer-monitor.snapshot'), res=Snapshot)
user.post(file('real-eee-1001pxd.snapshot.11'), res=Snapshot)
i, _ = user.get(res=Device, query=[('search', 'desktop')])
assert i['items'][0]['id'] == 1
assert i['items'][0]['id'] == snapshot['device']['id']
i, _ = user.get(res=Device, query=[('search', 'intel')])
assert len(i['items']) == 1
i, _ = user.get(res=Device, query=[('search', i['items'][0]['devicehubID'])])
assert len(i['items']) == 1
i, _ = user.get(res=Device, query=[('search', '1')])
i, _ = user.get(res=Device, query=[('search', snapshot['device']['id'])])
assert len(i['items']) == 1

27
tests/test_documents.py
View file

@ -10,14 +10,17 @@ from werkzeug.exceptions import Unauthorized
import teal.marshmallow
from ereuse_utils.test import ANY
from ereuse_devicehub import auth
from ereuse_devicehub.client import Client, UserClient
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.user.models import Session
from ereuse_devicehub.resources.action.models import Snapshot, Allocate, Live
from ereuse_devicehub.resources.documents import documents
from ereuse_devicehub.resources.device import models as d
from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.tag.model import Tag
from ereuse_devicehub.resources.hash_reports import ReportHash
from ereuse_devicehub.resources.enums import SessionType
from ereuse_devicehub.db import db
from tests import conftest
from tests.conftest import file
@ -262,6 +265,7 @@ def test_export_extended(app: Devicehub, user: UserClient):
pc.tags.add(tag)
db.session.add(pc)
db.session.commit()
csv_str, _ = user.get(res=documents.DocumentDef.t,
item='devices/',
accept='text/csv',
@ -619,7 +623,7 @@ def test_verify_stamp_erasure_certificate(user: UserClient, client: Client):
@pytest.mark.mvp
def test_get_document_internal_stats(user: UserClient, user2: UserClient):
"""Tests for get teh internal stats."""
"""Tests for get the internal stats."""
# csv_str, _ = user.get(res=documents.DocumentDef.t,
# item='internalstats/')
@ -644,3 +648,24 @@ def test_get_document_internal_stats(user: UserClient, user2: UserClient):
export_csv = list(obj_csv)
assert csv_str.strip() == '""'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_get_wbconf(user: UserClient):
"""Tests for get env file for usb wb."""
env, _ = user.get(res=documents.DocumentDef.t, item='wbconf/usodyrate', accept=ANY)
assert 'WB_ERASE = False' in env
env, _ = user.get(res=documents.DocumentDef.t, item='wbconf/usodywipe', accept=ANY)
assert 'WB_ERASE = False' in env
# assert 'WB_ERASE = True' in env
session = Session.query.filter_by(user_id=user.user['id'],
type=SessionType.Internal).first()
token = session.token
token = auth.Auth.encode(session.token)
assert token in env
user.user['token'] = token
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)

12
tests/test_snapshot.py
View file

@ -37,6 +37,7 @@ from tests import conftest
@pytest.mark.mvp
@pytest.mark.usefixtures('auth_app_context')
# cayop
def test_snapshot_model():
"""Tests creating a Snapshot with its relationships ensuring correct
DB mapping.
@ -318,7 +319,7 @@ def test_snapshot_tag_inner_tag(user: UserClient, tag_id: str, app: Devicehub):
action_types=(RateComputer.t, BenchmarkProcessor.t, VisualTest.t))
with app.app_context():
tag = Tag.query.one() # type: Tag
assert tag.device_id == 1, 'Tag should be linked to the first device'
assert tag.device_id == 3, 'Tag should be linked to the first device'
@pytest.mark.mvp
@ -838,3 +839,12 @@ def test_snapshot_mobil(app: Devicehub, user: UserClient):
tmp_snapshots = app.config['TMP_SNAPSHOTS']
shutil.rmtree(tmp_snapshots)
@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
"""
user.post(file('2021-5-4-13-41_time_out_test_datastorage'), res=Snapshot)

28
tests/test_workbench.py
View file

@ -32,20 +32,24 @@ def test_workbench_server_condensed(user: UserClient):
user.post({'id': t['id']}, res=Tag)
snapshot, _ = user.post(res=em.Snapshot, data=s)
pc_id = snapshot['device']['id']
cpu_id = snapshot['components'][3]['id']
ssd_id= snapshot['components'][4]['id']
hdd_id = snapshot['components'][5]['id']
actions = snapshot['actions']
assert {(action['type'], action['device']) for action in actions} == {
('BenchmarkProcessorSysbench', 5),
('StressTest', 1),
('EraseSectors', 6),
('EreusePrice', 1),
('BenchmarkRamSysbench', 1),
('BenchmarkProcessor', 5),
('Install', 6),
('EraseSectors', 7),
('BenchmarkDataStorage', 6),
('BenchmarkDataStorage', 7),
('TestDataStorage', 6),
('RateComputer', 1)
('BenchmarkProcessorSysbench', cpu_id),
('StressTest', pc_id),
('EraseSectors', ssd_id),
('EreusePrice', pc_id),
('BenchmarkRamSysbench', pc_id),
('BenchmarkProcessor', cpu_id),
('Install', ssd_id),
('EraseSectors', hdd_id),
('BenchmarkDataStorage', ssd_id),
('BenchmarkDataStorage', hdd_id),
('TestDataStorage', ssd_id),
('RateComputer', pc_id)
}
assert snapshot['closed']
assert snapshot['severity'] == 'Info'