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

Merge branch 'testing' of https://github.com/eReuse/devicehub-teal into testing

This commit is contained in:
Cayo Puigdefabregas 2021-06-09 11:47:20 +02:00
parent 8c43e24a34 a0d26a104f
commit 9d89d4fe4a
23 changed files with 1790 additions and 218 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
ereuse_devicehub/cli.py
View File

@ -7,6 +7,27 @@ import flask.cli
from ereuse_devicehub.config import DevicehubConfig
from ereuse_devicehub.devicehub import Devicehub
import sys
sys.ps1 = '\001\033[92m\002>>> \001\033[0m\002'
sys.ps2= '\001\033[94m\002... \001\033[0m\002'
import os, readline, rlcompleter, atexit
history_file = os.path.join(os.environ['HOME'], '.python_history')
try:
readline.read_history_file(history_file)
except IOError:
pass
readline.parse_and_bind("tab: complete")
readline.parse_and_bind('"\e[5~": history-search-backward')
readline.parse_and_bind('"\e[6~": history-search-forward')
readline.parse_and_bind('"\e[5C": forward-word')
readline.parse_and_bind('"\e[5D": backward-word')
readline.parse_and_bind('"\e\e[C": forward-word')
readline.parse_and_bind('"\e\e[D": backward-word')
readline.parse_and_bind('"\e[1;5C": forward-word')
readline.parse_and_bind('"\e[1;5D": backward-word')
readline.set_history_length(100000)
atexit.register(readline.write_history_file, history_file)
class DevicehubGroup(flask.cli.FlaskGroup):
# todo users cannot make cli to use a custom db this way!

125
ereuse_devicehub/migrations/versions/51439cf24be8_change_trade_action.py Normal file
View File

@ -0,0 +1,125 @@
"""change trade action
Revision ID: 51439cf24be8
Revises: eca457d8b2a4
Create Date: 2021-03-15 17:40:34.410408
"""
from alembic import op
from alembic import context
from sqlalchemy.dialects import postgresql
import sqlalchemy as sa
import citext
# revision identifiers, used by Alembic.
revision = '51439cf24be8'
down_revision = '21afd375a654'
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_data():
con = op.get_bind()
sql = "update common.user set active='t';"
con.execute(sql)
sql = "update common.user set phantom='f';"
con.execute(sql)
def upgrade():
## Trade
currency = sa.Enum('AFN', 'ARS', 'AWG', 'AUD', 'AZN', 'BSD', 'BBD', 'BDT', 'BYR', 'BZD', 'BMD',
'BOB', 'BAM', 'BWP', 'BGN', 'BRL', 'BND', 'KHR', 'CAD', 'KYD', 'CLP', 'CNY',
'COP', 'CRC', 'HRK', 'CUP', 'CZK', 'DKK', 'DOP', 'XCD', 'EGP', 'SVC', 'EEK',
'EUR', 'FKP', 'FJD', 'GHC', 'GIP', 'GTQ', 'GGP', 'GYD', 'HNL', 'HKD', 'HUF',
'ISK', 'INR', 'IDR', 'IRR', 'IMP', 'ILS', 'JMD', 'JPY', 'JEP', 'KZT', 'KPW',
'KRW', 'KGS', 'LAK', 'LVL', 'LBP', 'LRD', 'LTL', 'MKD', 'MYR', 'MUR', 'MXN',
'MNT', 'MZN', 'NAD', 'NPR', 'ANG', 'NZD', 'NIO', 'NGN', 'NOK', 'OMR', 'PKR',
'PAB', 'PYG', 'PEN', 'PHP', 'PLN', 'QAR', 'RON', 'RUB', 'SHP', 'SAR', 'RSD',
'SCR', 'SGD', 'SBD', 'SOS', 'ZAR', 'LKR', 'SEK', 'CHF', 'SRD', 'SYP', 'TWD',
'THB', 'TTD', 'TRY', 'TRL', 'TVD', 'UAH', 'GBP', 'USD', 'UYU', 'UZS', 'VEF', name='currency', create_type=False, checkfirst=True, schema=f'{get_inv()}')
op.drop_table('trade', schema=f'{get_inv()}')
op.create_table('trade',
sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('price', sa.Float(decimal_return_scale=4), nullable=True),
sa.Column('lot_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('date', sa.TIMESTAMP(timezone=True), nullable=True),
sa.Column('user_from_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('user_to_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('document_id', citext.CIText(), nullable=True),
sa.Column('confirm', sa.Boolean(), nullable=True),
sa.Column('code', citext.CIText(), default='', nullable=True,
comment = "This code is used for traceability"),
sa.ForeignKeyConstraint(['id'], [f'{get_inv()}.action.id'], ),
sa.ForeignKeyConstraint(['user_from_id'], ['common.user.id'], ),
sa.ForeignKeyConstraint(['user_to_id'], ['common.user.id'], ),
sa.ForeignKeyConstraint(['lot_id'], [f'{get_inv()}.lot.id'], ),
sa.PrimaryKeyConstraint('id'),
schema=f'{get_inv()}'
)
op.add_column("trade", sa.Column("currency", currency, nullable=False), schema=f'{get_inv()}')
op.create_table('confirm',
sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('user_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('action_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.ForeignKeyConstraint(['id'], [f'{get_inv()}.action.id'], ),
sa.ForeignKeyConstraint(['action_id'], [f'{get_inv()}.action.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['common.user.id'], ),
sa.PrimaryKeyConstraint('id'),
schema=f'{get_inv()}'
)
# ## User
op.add_column('user', sa.Column('active', sa.Boolean(), default=True, nullable=True),
schema='common')
op.add_column('user', sa.Column('phantom', sa.Boolean(), default=False, nullable=True),
schema='common')
upgrade_data()
op.alter_column('user', 'active', nullable=False, schema='common')
op.alter_column('user', 'phantom', nullable=False, schema='common')
def downgrade():
op.drop_table('confirm', schema=f'{get_inv()}')
op.drop_table('trade', schema=f'{get_inv()}')
op.create_table('trade',
sa.Column('shipping_date', sa.TIMESTAMP(timezone=True), nullable=True,
comment='When are the devices going to be ready \n \
for shipping?\n '),
sa.Column('invoice_number', citext.CIText(), nullable=True,
comment='The id of the invoice so they can be linked.'),
sa.Column('price_id', postgresql.UUID(as_uuid=True), nullable=True,
comment='The price set for this trade. \n \
If no price is set it is supposed that the trade was\n \
not payed, usual in donations.\n '),
sa.Column('to_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('confirms_id', postgresql.UUID(as_uuid=True), nullable=True,
comment='An organize action that this association confirms. \
\n \n For example, a ``Sell`` or ``Rent``\n \
can confirm a ``Reserve`` action.\n '),
sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
sa.ForeignKeyConstraint(['confirms_id'], [f'{get_inv()}.organize.id'], ),
sa.ForeignKeyConstraint(['id'], [f'{get_inv()}.action.id'], ),
sa.ForeignKeyConstraint(['price_id'], [f'{get_inv()}.price.id'], ),
sa.ForeignKeyConstraint(['to_id'], [f'{get_inv()}.agent.id'], ),
sa.PrimaryKeyConstraint('id'),
schema=f'{get_inv()}'
)
op.drop_column('user', 'active', schema='common')
op.drop_column('user', 'phantom', schema='common')

17
ereuse_devicehub/resources/action/__init__.py
View File

@ -3,7 +3,7 @@ from typing import Callable, Iterable, Tuple
from teal.resource import Converters, Resource
from ereuse_devicehub.resources.action import schemas
from ereuse_devicehub.resources.action.views import (ActionView, AllocateView, DeallocateView,
from ereuse_devicehub.resources.action.views.views import (ActionView, AllocateView, DeallocateView,
LiveView)
from ereuse_devicehub.resources.device.sync import Sync
@ -250,6 +250,21 @@ class MakeAvailable(ActionDef):
SCHEMA = schemas.MakeAvailable
class ConfirmDef(ActionDef):
VIEW = None
SCHEMA = schemas.Confirm
class ConfirmRevokeDef(ActionDef):
VIEW = None
SCHEMA = schemas.ConfirmRevoke
class RevokeDef(ActionDef):
VIEW = None
SCHEMA = schemas.Revoke
class TradeDef(ActionDef):
VIEW = None
SCHEMA = schemas.Trade

103
ereuse_devicehub/resources/action/models.py
View File

@ -1433,6 +1433,46 @@ class CancelReservation(Organize):
"""The act of cancelling a reservation."""
class Confirm(JoinedTableMixin, ActionWithMultipleDevices):
"""Users confirm the one action trade this confirmation it's link to trade
and the devices that confirm
"""
user_id = db.Column(UUID(as_uuid=True),
db.ForeignKey(User.id),
nullable=False,
default=lambda: g.user.id)
user = db.relationship(User, primaryjoin=user_id == User.id)
user_comment = """The user that accept the offer."""
action_id = db.Column(UUID(as_uuid=True),
db.ForeignKey('action.id'),
nullable=False)
action = db.relationship('Action',
backref=backref('acceptances',
uselist=True,
lazy=True,
order_by=lambda: Action.end_time,
collection_class=list),
primaryjoin='Confirm.action_id == Action.id')
def __repr__(self) -> str:
if self.action.t in ['Trade']:
origin = 'To'
if self.user == self.action.user_from:
origin = 'From'
return '<{0.t} {0.id} accepted by {1}>'.format(self, origin)
class Revoke(Confirm):
"""Users can revoke one confirmation of one action trade"""
class ConfirmRevoke(Confirm):
"""Users can confirm and accept one action revoke"""
def __repr__(self) -> str:
return '<{0.t} {0.id} accepted by {0.user}>'.format(self)
class Trade(JoinedTableMixin, ActionWithMultipleDevices):
"""Trade actions log the political exchange of devices between users.
Every time a trade action is performed, the old user looses its
@ -1446,34 +1486,41 @@ class Trade(JoinedTableMixin, ActionWithMultipleDevices):
This class and its inheritors
extend `Schema's Trade <http://schema.org/TradeAction>`_.
"""
shipping_date = Column(db.TIMESTAMP(timezone=True))
shipping_date.comment = """When are the devices going to be ready
for shipping?
"""
invoice_number = Column(CIText())
invoice_number.comment = """The id of the invoice so they can be linked."""
price_id = Column(UUID(as_uuid=True), ForeignKey(Price.id))
price = relationship(Price,
backref=backref('trade', lazy=True, uselist=False),
primaryjoin=price_id == Price.id)
price_id.comment = """The price set for this trade.
If no price is set it is supposed that the trade was
not payed, usual in donations.
"""
to_id = Column(UUID(as_uuid=True), ForeignKey(Agent.id), nullable=False)
# todo compute the org
to = relationship(Agent,
backref=backref('actions_to', lazy=True, **_sorted_actions),
primaryjoin=to_id == Agent.id)
to_comment = """The agent that gets the device due this deal."""
confirms_id = Column(UUID(as_uuid=True), ForeignKey(Organize.id))
confirms = relationship(Organize,
backref=backref('confirmation', lazy=True, uselist=False),
primaryjoin=confirms_id == Organize.id)
confirms_id.comment = """An organize action that this association confirms.
For example, a ``Sell`` or ``Rent``
can confirm a ``Reserve`` action.
"""
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
user_from_id = db.Column(UUID(as_uuid=True),
db.ForeignKey(User.id),
nullable=False)
user_from = db.relationship(User, primaryjoin=user_from_id == User.id)
user_from_comment = """The user that offers the device due this deal."""
user_to_id = db.Column(UUID(as_uuid=True),
db.ForeignKey(User.id),
nullable=False)
user_to = db.relationship(User, primaryjoin=user_to_id == User.id)
user_to_comment = """The user that gets the device due this deal."""
price = Column(Float(decimal_return_scale=2), nullable=True)
currency = Column(DBEnum(Currency), nullable=False, default=Currency.EUR.name)
currency.comment = """The currency of this price as for ISO 4217."""
date = Column(db.TIMESTAMP(timezone=True))
document_id = Column(CIText())
document_id.comment = """The id of one document like invoice so they can be linked."""
confirm = Column(Boolean, default=False, nullable=False)
confirm.comment = """If you need confirmation of the user, you need actevate this field"""
code = Column(CIText(), nullable=True)
code.comment = """If the user not exist, you need a code to be able to do the traceability"""
lot_id = db.Column(UUID(as_uuid=True),
db.ForeignKey('lot.id',
use_alter=True,
name='lot_trade'),
nullable=True)
lot = relationship('Lot',
backref=backref('trade',
lazy=True,
uselist=False,
cascade=CASCADE_OWN),
primaryjoin='Trade.lot_id == Lot.id')
def __repr__(self) -> str:
return '<{0.t} {0.id} executed by {0.author}>'.format(self)
class InitTransfer(Trade):

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

@ -1,6 +1,7 @@
import copy
from datetime import datetime, timedelta
from dateutil.tz import tzutc
from flask import current_app as app
from flask import current_app as app, g
from marshmallow import Schema as MarshmallowSchema, ValidationError, fields as f, validates_schema
from marshmallow.fields import Boolean, DateTime, Decimal, Float, Integer, Nested, String, \
TimeDelta, UUID
@ -21,6 +22,7 @@ from ereuse_devicehub.resources.enums import AppearanceRange, BiosAccessRange, F
from ereuse_devicehub.resources.models import STR_BIG_SIZE, STR_SIZE
from ereuse_devicehub.resources.schemas import Thing
from ereuse_devicehub.resources.user import schemas as s_user
from ereuse_devicehub.resources.user.models import User
class Action(Thing):
@ -455,13 +457,146 @@ class CancelReservation(Organize):
__doc__ = m.CancelReservation.__doc__
class Confirm(ActionWithMultipleDevices):
__doc__ = m.Confirm.__doc__
action = NestedOn('Action', only_query='id')
@validates_schema
def validate_revoke(self, data: dict):
for dev in data['devices']:
# if device not exist in the Trade, then this query is wrong
if not dev in data['action'].devices:
txt = "Device {} not exist in the trade".format(dev.devicehub_id)
raise ValidationError(txt)
class Revoke(ActionWithMultipleDevices):
__doc__ = m.Revoke.__doc__
action = NestedOn('Action', only_query='id')
@validates_schema
def validate_revoke(self, data: dict):
for dev in data['devices']:
# if device not exist in the Trade, then this query is wrong
if not dev in data['action'].devices:
txt = "Device {} not exist in the trade".format(dev.devicehub_id)
raise ValidationError(txt)
class ConfirmRevoke(ActionWithMultipleDevices):
__doc__ = m.ConfirmRevoke.__doc__
action = NestedOn('Action', only_query='id')
@validates_schema
def validate_revoke(self, data: dict):
# import pdb; pdb.set_trace()
for dev in data['devices']:
# if device not exist in the Trade, then this query is wrong
if not dev in data['action'].devices:
txt = "Device {} not exist in the revoke action".format(dev.devicehub_id)
raise ValidationError(txt)
class Trade(ActionWithMultipleDevices):
__doc__ = m.Trade.__doc__
shipping_date = DateTime(data_key='shippingDate')
invoice_number = SanitizedStr(validate=Length(max=STR_SIZE), data_key='invoiceNumber')
price = NestedOn(Price)
to = NestedOn(s_agent.Agent, only_query='id', required=True, comment=m.Trade.to_comment)
confirms = NestedOn(Organize)
document_id = SanitizedStr(validate=Length(max=STR_SIZE), data_key='documentID', required=False)
date = DateTime(data_key='date', required=False)
price = Float(required=False, data_key='price')
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(
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,
only_query='id')
@validates_schema
def validate_lot(self, data: dict):
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)
for dev in data['lot'].devices:
if not dev.owner == g.user:
txt = "you need to be the owner of the devices for to do a trade"
raise ValidationError(txt)
if not data['lot'].owner == g.user:
txt = "you need to be the owner of the lot for to do a trade"
raise ValidationError(txt)
data['devices'] = data['lot'].devices
@validates_schema
def validate_user_to_email(self, data: dict):
"""
- if user_to exist
* confirmation
* without confirmation
- if user_to don't exist
* without confirmation
"""
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_email(self, data: dict):
"""
- if user_from exist
* confirmation
* without confirmation
- if user_from don't exist
* without confirmation
"""
if data['user_from_email']:
user_from = User.query.filter_by(email=data['user_from_email']).one()
data['user_from'] = user_from
@validates_schema
def validate_email_users(self, data: dict):
"""We need at least one user"""
if not (data['user_from_email'] or data['user_to_email']):
txt = "you need one user from or user to for to do a trade"
raise ValidationError(txt)
if not g.user.email in [data['user_from_email'], data['user_to_email']]:
txt = "you need to be one of participate of the action"
raise ValidationError(txt)
@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_email'] and data['user_to_email']:
data['confirm'] = True
return
if not data['confirm'] and not data.get('code'):
txt = "you need a code to be able to do the traceability"
raise ValidationError(txt)
data['code'] = data['code'].replace('@', '_')
class InitTransfer(Trade):

0
ereuse_devicehub/resources/action/views/__init__.py Normal file
View File

145
ereuse_devicehub/resources/action/views/snapshot.py Normal file
View File

@ -0,0 +1,145 @@
""" This is the view for Snapshots """
import os
import json
import shutil
from datetime import datetime
from flask import current_app as app, g
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.user.exceptions import InsufficientPermission
def save_json(req_json, tmp_snapshots, user, live=False):
"""
This function allow save a snapshot in json format un a TMP_SNAPSHOTS directory
The file need to be saved with one name format with the stamptime and uuid joins
"""
uuid = req_json.get('uuid', '')
now = datetime.now()
year = now.year
month = now.month
day = now.day
hour = now.hour
minutes = now.minute
name_file = f"{year}-{month}-{day}-{hour}-{minutes}_{user}_{uuid}.json"
path_dir_base = os.path.join(tmp_snapshots, user)
if live:
path_dir_base = tmp_snapshots
path_errors = os.path.join(path_dir_base, 'errors')
path_fixeds = os.path.join(path_dir_base, 'fixeds')
path_name = os.path.join(path_errors, name_file)
if not os.path.isdir(path_dir_base):
os.system(f'mkdir -p {path_errors}')
os.system(f'mkdir -p {path_fixeds}')
with open(path_name, 'w') as snapshot_file:
snapshot_file.write(json.dumps(req_json))
return path_name
def move_json(tmp_snapshots, path_name, user, live=False):
"""
This function move the json than it's correct
"""
path_dir_base = os.path.join(tmp_snapshots, user)
if live:
path_dir_base = tmp_snapshots
if os.path.isfile(path_name):
shutil.copy(path_name, path_dir_base)
os.remove(path_name)
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
def __init__(self, snapshot_json: dict, resource_def, schema):
self.schema = schema
self.snapshot_json = snapshot_json
self.resource_def = resource_def
self.tmp_snapshots = app.config['TMP_SNAPSHOTS']
self.path_snapshot = save_json(snapshot_json, self.tmp_snapshots, g.user.email)
snapshot_json.pop('debug', None)
self.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):
return self.response
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 isinstance(device, Computer) and device.hid:
device.add_mac_to_hid(components_snap=components)
snapshot = Snapshot(**self.snapshot_json)
# Remove new actions from devices so they don't interfere with sync
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)
for component in components:
component.actions_one.clear()
assert not device.actions_one
assert all(not c.actions_one for c in components) if components else True
db_device, remove_actions = self.resource_def.sync.run(device, components)
del device # Do not use device anymore
snapshot.device = db_device
snapshot.actions |= remove_actions | actions_device # Set actions to snapshot
# commit will change the order of the components by what
# the DB wants. Let's get a copy of the list so we preserve order
ordered_components = OrderedSet(x for x in snapshot.components)
# Add the new actions to the db-existing devices and components
db_device.actions_one |= actions_device
if components:
for component, actions in zip(ordered_components, actions_components):
component.actions_one |= actions
snapshot.actions |= actions
if snapshot.software == SnapshotSoftware.Workbench:
# Check ownership of (non-component) device to from current.user
if db_device.owner_id != g.user.id:
raise InsufficientPermission()
# Compute ratings
try:
rate_computer, price = RateComputer.compute(db_device)
except CannotRate:
pass
else:
snapshot.actions.add(rate_computer)
if price:
snapshot.actions.add(price)
elif snapshot.software == SnapshotSoftware.WorkbenchAndroid:
pass # TODO try except to compute RateMobile
# Check if HID is null and add Severity:Warning to Snapshot
if snapshot.device.hid is None:
snapshot.severity = Severity.Warning
db.session.add(snapshot)
db.session().final_flush()
ret = self.schema.jsonify(snapshot) # transform it back
ret.status_code = 201
db.session.commit()
return ret

263
ereuse_devicehub/resources/action/views/trade.py Normal file
View File

@ -0,0 +1,263 @@
import copy
from flask import g
from sqlalchemy.util import OrderedSet
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():
"""Handler for manager the trade action register from post
request_post = {
'type': 'Trade',
'devices': [device_id],
'userFrom': user2.email,
'userTo': user.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'documentID': '1',
'lot': lot['id'],
'confirm': True,
}
"""
def __init__(self, data, resource_def, schema):
self.schema = schema
self.data = resource_def.schema.load(data)
self.data.pop('user_to_email', '')
self.data.pop('user_from_email', '')
self.create_phantom_account()
self.trade = Trade(**self.data)
db.session.add(self.trade)
self.create_confirmations()
self.create_automatic_trade()
def post(self):
db.session().final_flush()
ret = self.schema.jsonify(self.trade)
ret.status_code = 201
db.session.commit()
return ret
def create_confirmations(self) -> None:
"""Do the first confirmation for the user than do the action"""
# if the confirmation is mandatory, do automatic confirmation only for
# owner of the lot
if self.trade.confirm:
confirm = Confirm(user=g.user,
action=self.trade,
devices=self.trade.devices)
db.session.add(confirm)
return
# check than the user than want to do the action is one of the users
# involved in the action
if not g.user in [self.trade.user_from, self.trade.user_to]:
txt = "You do not participate in this trading"
raise ValidationError(txt)
confirm_from = Confirm(user=self.trade.user_from,
action=self.trade,
devices=self.trade.devices)
confirm_to = Confirm(user=self.trade.user_to,
action=self.trade,
devices=self.trade.devices)
db.session.add(confirm_from)
db.session.add(confirm_to)
def create_phantom_account(self) -> None:
"""
If exist both users not to do nothing
If exist from but not to:
search if exist in the DB
if exist use it
else create new one
The same if exist to but not from
"""
user_from = self.data.get('user_from')
user_to = self.data.get('user_to')
code = self.data.get('code')
if user_from and user_to:
return
if self.data['confirm']:
return
if user_from and not user_to:
assert g.user == user_from
email = "{}_{}@dhub.com".format(str(user_from.id), code)
users = User.query.filter_by(email=email)
if users.first():
user = users.first()
self.data['user_to'] = user
return
user = User(email=email, password='', active=False, phantom=True)
db.session.add(user)
self.data['user_to'] = user
if not user_from and user_to:
email = "{}_{}@dhub.com".format(str(user_to.id), code)
users = User.query.filter_by(email=email)
if users.first():
user = users.first()
self.data['user_from'] = user
return
user = User(email=email, password='', active=False, phantom=True)
db.session.add(user)
self.data['user_from'] = user
def create_automatic_trade(self) -> None:
# not do nothing if it's neccesary confirmation explicity
if self.trade.confirm:
return
# Change the owner for every devices
for dev in self.trade.devices:
dev.change_owner(self.trade.user_to)
class ConfirmMixin():
"""
Very Important:
==============
All of this Views than inherit of this class is executed for users
than is not owner of the Trade action.
The owner of Trade action executed this actions of confirm and revoke from the
lot
"""
Model = None
def __init__(self, data, resource_def, schema):
self.schema = schema
a = resource_def.schema.load(data)
self.validate(a)
if not a['devices']:
raise ValidationError('Devices not exist.')
self.model = self.Model(**a)
def post(self):
db.session().final_flush()
ret = self.schema.jsonify(self.model)
ret.status_code = 201
db.session.commit()
return ret
class ConfirmView(ConfirmMixin):
"""Handler for manager the Confirmation register from post
request_confirm = {
'type': 'Confirm',
'action': trade.id,
'devices': [device_id]
}
"""
Model = Confirm
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
"""
# import pdb; pdb.set_trace()
real_devices = []
for dev in data['devices']:
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']:
user_to = data['action'].user_to
dev.change_owner(user_to)
class RevokeView(ConfirmMixin):
"""Handler for manager the Revoke register from post
request_revoke = {
'type': 'Revoke',
'action': trade.id,
'devices': [device_id],
}
"""
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):
"""All devices need to have the status of DoubleConfirmation."""
### check ###
if not data['devices']:
raise ValidationError('Devices not exist.')
for dev in data['devices']:
if not dev.trading == 'TradeConfirmed':
txt = 'Some of devices do not have enough to confirm for to do a revoke'
ValidationError(txt)
### End check ###
ids = {d.id for d in data['devices']}
lot = data['action'].lot
# import pdb; pdb.set_trace()
self.model = delete_from_trade(lot, ids)
class ConfirmRevokeView(ConfirmMixin):
"""Handler for manager the Confirmation register from post
request_confirm_revoke = {
'type': 'ConfirmRevoke',
'action': action_revoke.id,
'devices': [device_id]
}
"""
Model = ConfirmRevoke
def validate(self, data):
"""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']:
if not dev.trading == 'Revoke':
txt = 'Some of devices do not have revoke to confirm'
ValidationError(txt)
devices = OrderedSet(data['devices'])
data['devices'] = devices
# Change the owner for every devices
# data['action'] == 'Revoke'
trade = data['action'].action
for dev in devices:
dev.reset_owner()
trade.lot.devices.difference_update(devices)

159
ereuse_devicehub/resources/action/views.py → ereuse_devicehub/resources/action/views/views.py
View File

@ -1,73 +1,26 @@
""" This is the view for Snapshots """
import os
import json
import shutil
from datetime import datetime, timedelta
from datetime import timedelta
from distutils.version import StrictVersion
from uuid import UUID
from flask import current_app as app, request, g
from sqlalchemy.util import OrderedSet
from teal.marshmallow import ValidationError
from teal.resource import View
from teal.db import ResourceNotFound
from ereuse_devicehub.db import db
from ereuse_devicehub.query import things_response
from ereuse_devicehub.resources.action.models import (Action, RateComputer, Snapshot, VisualTest,
InitTransfer, Live, Allocate, Deallocate)
from ereuse_devicehub.resources.action.models import (Action, Snapshot, VisualTest,
InitTransfer, Live, Allocate, Deallocate,
Trade, Confirm, ConfirmRevoke, Revoke)
from ereuse_devicehub.resources.device.models import Device, Computer, DataStorage
from ereuse_devicehub.resources.action.rate.v1_0 import CannotRate
from ereuse_devicehub.resources.enums import SnapshotSoftware, Severity
from ereuse_devicehub.resources.user.exceptions import InsufficientPermission
from ereuse_devicehub.resources.enums import Severity
from ereuse_devicehub.resources.action.views import trade as trade_view
from ereuse_devicehub.resources.action.views.snapshot import SnapshotView, save_json, move_json
SUPPORTED_WORKBENCH = StrictVersion('11.0')
def save_json(req_json, tmp_snapshots, user, live=False):
"""
This function allow save a snapshot in json format un a TMP_SNAPSHOTS directory
The file need to be saved with one name format with the stamptime and uuid joins
"""
uuid = req_json.get('uuid', '')
now = datetime.now()
year = now.year
month = now.month
day = now.day
hour = now.hour
minutes = now.minute
name_file = f"{year}-{month}-{day}-{hour}-{minutes}_{user}_{uuid}.json"
path_dir_base = os.path.join(tmp_snapshots, user)
if live:
path_dir_base = tmp_snapshots
path_errors = os.path.join(path_dir_base, 'errors')
path_fixeds = os.path.join(path_dir_base, 'fixeds')
path_name = os.path.join(path_errors, name_file)
if not os.path.isdir(path_dir_base):
os.system(f'mkdir -p {path_errors}')
os.system(f'mkdir -p {path_fixeds}')
with open(path_name, 'w') as snapshot_file:
snapshot_file.write(json.dumps(req_json))
return path_name
def move_json(tmp_snapshots, path_name, user, live=False):
"""
This function move the json than it's correct
"""
path_dir_base = os.path.join(tmp_snapshots, user)
if live:
path_dir_base = tmp_snapshots
if os.path.isfile(path_name):
shutil.copy(path_name, path_dir_base)
os.remove(path_name)
class AllocateMix():
model = None
@ -223,18 +176,32 @@ class ActionView(View):
# defs
resource_def = app.resources[json['type']]
if json['type'] == Snapshot.t:
tmp_snapshots = app.config['TMP_SNAPSHOTS']
path_snapshot = save_json(json, tmp_snapshots, g.user.email)
json.pop('debug', None)
a = resource_def.schema.load(json)
response = self.snapshot(a, resource_def)
move_json(tmp_snapshots, path_snapshot, g.user.email)
return response
snapshot = SnapshotView(json, resource_def, self.schema)
return snapshot.post()
if json['type'] == VisualTest.t:
pass
# TODO JN add compute rate with new visual test and old components device
if json['type'] == InitTransfer.t:
return self.transfer_ownership()
if json['type'] == Trade.t:
trade = trade_view.TradeView(json, resource_def, self.schema)
return trade.post()
if json['type'] == Confirm.t:
confirm = trade_view.ConfirmView(json, resource_def, self.schema)
return confirm.post()
if json['type'] == Revoke.t:
revoke = trade_view.RevokeView(json, resource_def, self.schema)
return revoke.post()
if json['type'] == ConfirmRevoke.t:
confirm_revoke = trade_view.ConfirmRevokeView(json, resource_def, self.schema)
return confirm_revoke.post()
a = resource_def.schema.load(json)
Model = db.Model._decl_class_registry.data[json['type']]()
action = Model(**a)
@ -250,75 +217,7 @@ class ActionView(View):
action = Action.query.filter_by(id=id).one()
return self.schema.jsonify(action)
def snapshot(self, snapshot_json: dict, resource_def):
"""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
device = snapshot_json.pop('device') # type: Computer
components = None
if snapshot_json['software'] == (SnapshotSoftware.Workbench or SnapshotSoftware.WorkbenchAndroid):
components = 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(**snapshot_json)
# Remove new actions from devices so they don't interfere with sync
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)
for component in components:
component.actions_one.clear()
assert not device.actions_one
assert all(not c.actions_one for c in components) if components else True
db_device, remove_actions = resource_def.sync.run(device, components)
del device # Do not use device anymore
snapshot.device = db_device
snapshot.actions |= remove_actions | actions_device # Set actions to snapshot
# commit will change the order of the components by what
# the DB wants. Let's get a copy of the list so we preserve order
ordered_components = OrderedSet(x for x in snapshot.components)
# Add the new actions to the db-existing devices and components
db_device.actions_one |= actions_device
if components:
for component, actions in zip(ordered_components, actions_components):
component.actions_one |= actions
snapshot.actions |= actions
if snapshot.software == SnapshotSoftware.Workbench:
# Check ownership of (non-component) device to from current.user
if db_device.owner_id != g.user.id:
raise InsufficientPermission()
# Compute ratings
try:
rate_computer, price = RateComputer.compute(db_device)
except CannotRate:
pass
else:
snapshot.actions.add(rate_computer)
if price:
snapshot.actions.add(price)
elif snapshot.software == SnapshotSoftware.WorkbenchAndroid:
pass # TODO try except to compute RateMobile
# Check if HID is null and add Severity:Warning to Snapshot
if snapshot.device.hid is None:
snapshot.severity = Severity.Warning
db.session.add(snapshot)
db.session().final_flush()
ret = self.schema.jsonify(snapshot) # transform it back
ret.status_code = 201
db.session.commit()
return ret
def transfer_ownership(self):
"""Perform a InitTransfer action to change author_id of device"""
pass

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,

5
ereuse_devicehub/resources/device/states.py
View File

@ -23,6 +23,7 @@ class Trading(State):
"""Trading states.
:cvar Reserved: The device has been reserved.
:cvar Trade: The devices has been changed of owner.
:cvar Cancelled: The device has been cancelled.
:cvar Sold: The device has been sold.
:cvar Donated: The device is donated.
@ -33,6 +34,10 @@ class Trading(State):
from the facility. It does not mean end-of-life.
"""
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

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)

99
ereuse_devicehub/resources/lot/views.py
View File

@ -12,8 +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 Trade, Confirm, Revoke, ConfirmRevoke
from ereuse_devicehub.resources.lot.models import Lot, Path
@ -97,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
@ -108,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)
@ -224,7 +224,92 @@ class LotDeviceView(LotBaseChildrenView):
id = ma.fields.List(ma.fields.Integer())
def _post(self, lot: Lot, ids: Set[int]):
lot.devices.update(Device.query.filter(Device.id.in_(ids)))
# get only new devices
ids -= {x.id for x in lot.devices}
if not ids:
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
devices = set(Device.query.filter(Device.id.in_(ids)).filter(
Device.owner_id.in_(users)))
lot.devices.update(devices)
if lot.trade:
lot.trade.devices = lot.devices
if g.user in [lot.trade.user_from, lot.trade.user_to]:
confirm = Confirm(action=lot.trade, user=g.user, devices=devices)
db.session.add(confirm)
def _delete(self, lot: Lot, ids: Set[int]):
lot.devices.difference_update(Device.query.filter(Device.id.in_(ids)))
# if there are some devices in ids than not exist now in the lot, then exit
if not ids.issubset({x.id for x in lot.devices}):
return
if lot.trade:
return delete_from_trade(lot, ids)
# import pdb; pdb.set_trace()
if not g.user == lot.owner:
txt = 'This is not your lot'
raise ma.ValidationError(txt)
devices = set(Device.query.filter(Device.id.in_(ids)).filter(
Device.owner_id == g.user.id))
lot.devices.difference_update(devices)
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

12
ereuse_devicehub/resources/user/models.py
View File

@ -2,7 +2,7 @@ from uuid import uuid4
from citext import CIText
from flask import current_app as app
from sqlalchemy import Column, BigInteger, Sequence
from sqlalchemy import Column, Boolean, BigInteger, Sequence
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy_utils import EmailType, PasswordType
from teal.db import IntEnum
@ -23,6 +23,8 @@ class User(Thing):
**kwargs
)))
token = Column(UUID(as_uuid=True), default=uuid4, unique=True, nullable=False)
active = Column(Boolean, default=True, nullable=False)
phantom = Column(Boolean, default=False, nullable=False)
inventories = db.relationship(Inventory,
backref=db.backref('users', lazy=True, collection_class=set),
secondary=lambda: UserInventory.__table__,
@ -30,16 +32,20 @@ class User(Thing):
# todo set restriction that user has, at least, one active db
def __init__(self, email, password=None, inventories=None) -> None:
def __init__(self, email, password=None, inventories=None, active=True, phantom=False) -> None:
"""Creates an user.
:param email:
:param password:
:param inventories: A set of Inventory where the user has
access to. If none, the user is granted access to the current
inventory.
:param active: allow active and deactive one account without delete the account
:param phantom: it's util for identify the phantom accounts
create during the trade actions
"""
inventories = inventories or {Inventory.current}
super().__init__(email=email, password=password, inventories=inventories)
super().__init__(email=email, password=password, inventories=inventories,
active=active, phantom=phantom)
def __repr__(self) -> str:
return '<User {0.email}>'.format(self)

2
ereuse_devicehub/resources/user/views.py
View File

@ -19,7 +19,7 @@ def login():
user_s = g.resource_def.SCHEMA(only=('email', 'password')) # type: UserS
# noinspection PyArgumentList
u = request.get_json(schema=user_s)
user = User.query.filter_by(email=u['email']).one_or_none()
user = User.query.filter_by(email=u['email'], active=True, phantom=False).one_or_none()
if user and user.password == u['password']:
schema_with_token = g.resource_def.SCHEMA(exclude=set())
return schema_with_token.jsonify(user)

680
tests/test_action.py
View File

@ -9,6 +9,8 @@ from datetime import datetime, timedelta
from dateutil.tz import tzutc
from decimal import Decimal
from typing import Tuple, Type
from pytest import raises
from json.decoder import JSONDecodeError
from flask import current_app as app, g
from sqlalchemy.util import OrderedSet
@ -18,6 +20,9 @@ from ereuse_devicehub.db import db
from ereuse_devicehub.client import UserClient, Client
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources import enums
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.agent.models import Person
from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.action import models
from ereuse_devicehub.resources.device import states
from ereuse_devicehub.resources.device.models import Desktop, Device, GraphicCard, HardDrive, \
@ -740,34 +745,245 @@ def test_deallocate_bad_dates(user: UserClient):
@pytest.mark.mvp
@pytest.mark.parametrize('action_model_state',
(pytest.param(ams, id=ams[0].__name__)
for ams in [
(models.MakeAvailable, states.Trading.Available),
(models.Sell, states.Trading.Sold),
(models.Donate, states.Trading.Donated),
(models.Rent, states.Trading.Renting),
(models.DisposeProduct, states.Trading.ProductDisposed)
]))
def test_trade(action_model_state: Tuple[Type[models.Action], states.Trading], user: UserClient):
"""Tests POSTing all Trade actions."""
# todo missing None states.Trading for after cancelling renting, for example
# Remove this test
action_model, state = action_model_state
@pytest.mark.xfail(reason='Old functionality')
def test_trade_endpoint(user: UserClient, user2: UserClient):
"""Tests POST one simple Trade between 2 users of the system."""
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
action = {
'type': action_model.t,
device, _ = user.get(res=Device, item=snapshot['device']['id'])
assert device['id'] == snapshot['device']['id']
request_post = {
'userTo': user2.user['email'],
'price': 1.0,
'date': "2020-12-01T02:00:00+00:00",
'devices': [snapshot['device']['id']]
}
if issubclass(action_model, models.Trade):
action['to'] = user.user['individuals'][0]['id']
action['shippingDate'] = '2018-06-29T12:28:54'
action['invoiceNumber'] = 'ABC'
action, _ = user.post(action, res=models.Action)
assert action['devices'][0]['id'] == snapshot['device']['id']
device, _ = user.get(res=Device, item=snapshot['device']['devicehubID'])
assert device['actions'][-1]['id'] == action['id']
assert device['trading'] == state.name
action, _ = user.post(res=models.Trade, data=request_post)
with raises(JSONDecodeError):
device1, _ = user.get(res=Device, item=device['id'])
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 trade'
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
@pytest.mark.mvp
@ -819,3 +1035,417 @@ 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)
# 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'])
# 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)
# import pdb; pdb.set_trace()
assert len(trade.lot.devices) == len(trade.devices) == 10
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'
# 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'
# assert len(trade.lot.devices) == len(trade.devices) == 9
# assert not device_10 in trade.devices
# 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'

2
tests/test_basic.py
View File

@ -121,4 +121,4 @@ def test_api_docs(client: Client):
'scheme': 'basic',
'name': 'Authorization'
}
assert len(docs['definitions']) == 118
assert len(docs['definitions']) == 121

33
tests/test_lot.py
View File

@ -1,9 +1,13 @@
import pytest
from flask import g
from pytest import raises
from json.decoder import JSONDecodeError
from ereuse_devicehub.client import UserClient
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.agent.models import Person
from ereuse_devicehub.resources.device.models import Desktop, Device, GraphicCard
from ereuse_devicehub.resources.enums import ComputerChassis
from ereuse_devicehub.resources.lot.models import Lot, LotDevice
@ -384,6 +388,35 @@ def test_lot_post_add_remove_device_view(app: Devicehub, user: UserClient):
assert not len(lot['devices'])
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_lot_error_add_device_from_other_user(user: UserClient):
"""Tests adding a device to a lot using POST and
removing it with DELETE.
"""
user2 = User(email='baz@baz.cxm', password='baz')
user2.individuals.add(Person(name='Tommy'))
db.session.add(user2)
db.session.commit()
device = Desktop(serial_number='foo',
model='bar',
manufacturer='foobar',
chassis=ComputerChassis.Lunchbox,
owner_id=user2.id)
db.session.add(device)
db.session.commit()
device_id = device.id
parent, _ = user.post(({'name': 'lot'}), res=Lot)
lot, _ = user.post({},
res=Lot,
item='{}/devices'.format(parent['id']),
query=[('id', device_id)])
assert lot['devices'] == [], 'Lot contains device'
assert len(lot['devices']) == 0
@pytest.mark.mvp
def test_get_multiple_lots(user: UserClient):
"""Tests submitting and retreiving multiple lots."""

2
tests/test_snapshot.py
View File

@ -29,7 +29,7 @@ from ereuse_devicehub.resources.device.sync import MismatchBetweenProperties, \
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 import save_json
from ereuse_devicehub.resources.action.views.snapshot import save_json
from ereuse_devicehub.resources.documents import documents
from tests.conftest import file
from tests import conftest

33
tests/test_user.py
View File

@ -87,6 +87,39 @@ def test_login_success(client: Client, app: Devicehub):
assert user['inventories'][0]['id'] == 'test'
@pytest.mark.mvp
@pytest.mark.usefixtures(app_context.__name__)
def test_login_active_phantom(client: Client):
"""Tests successfully performing login.
This checks that:
- User is returned if is active and is not phantom.
"""
dbuser = User(email='foo@foo.com', password='foo')
dbuser1 = User(email='foo1@foo.com', password='foo', active=True, phantom=False)
dbuser2 = User(email='foo2@foo.com', password='foo', active=False, phantom=False)
dbuser3 = User(email='foo3@foo.com', password='foo', active=True, phantom=True)
dbuser4 = User(email='foo4@foo.com', password='foo', active=False, phantom=True)
db.session.add(dbuser)
db.session.add(dbuser1)
db.session.add(dbuser2)
db.session.add(dbuser3)
db.session.add(dbuser4)
db.session.commit()
db.session.flush()
assert dbuser.active
assert not dbuser.phantom
uri = '/users/login/'
client.post({'email': 'foo@foo.com', 'password': 'foo'}, uri=uri, status=200)
client.post({'email': 'foo1@foo.com', 'password': 'foo'}, uri=uri, status=200)
client.post({'email': 'foo2@foo.com', 'password': 'foo'}, uri=uri, status=401)
client.post({'email': 'foo3@foo.com', 'password': 'foo'}, uri=uri, status=401)
client.post({'email': 'foo4@foo.com', 'password': 'foo'}, uri=uri, status=401)
@pytest.mark.mvp
def test_login_failure(client: Client, app: Devicehub):
"""Tests performing wrong login."""