Merge pull request #173 from eReuse/feature/new-metrix

Feature/new metrix
This commit is contained in:
cayop 2021-10-20 15:08:01 +02:00 committed by GitHub
commit 9972c1666f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 485 additions and 73 deletions

View File

@ -49,6 +49,7 @@ from ereuse_devicehub.resources.enums import AppearanceRange, BatteryHealth, Bio
from ereuse_devicehub.resources.models import STR_SM_SIZE, Thing from ereuse_devicehub.resources.models import STR_SM_SIZE, Thing
from ereuse_devicehub.resources.user.models import User from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.tradedocument.models import TradeDocument from ereuse_devicehub.resources.tradedocument.models import TradeDocument
from ereuse_devicehub.resources.device.metrics import TradeMetrics
class JoinedTableMixin: class JoinedTableMixin:
@ -1632,6 +1633,16 @@ class Trade(JoinedTableMixin, ActionWithMultipleTradeDocuments):
cascade=CASCADE_OWN), cascade=CASCADE_OWN),
primaryjoin='Trade.lot_id == Lot.id') primaryjoin='Trade.lot_id == Lot.id')
def get_metrics(self):
"""
This method get a list of values for calculate a metrics from a spreadsheet
"""
metrics = []
for doc in self.documents:
m = TradeMetrics(document=doc, Trade=self)
metrics.extend(m.get_metrics())
return metrics
def __repr__(self) -> str: def __repr__(self) -> str:
return '<{0.t} {0.id} executed by {0.author}>'.format(self) return '<{0.t} {0.id} executed by {0.author}>'.format(self)

View File

@ -0,0 +1,242 @@
import copy
class MetricsMix:
"""we want get the data metrics of one device"""
def __init__(self, *args, **kwargs):
self.actions.sort(key=lambda x: x.created)
self.rows = []
self.lifetime = 0
self.last_trade = None
self.action_create_by = 'Receiver'
self.status_receiver = 'Use'
self.status_supplier = ''
self.act = None
self.end_users = 0
self.final_user_code = ''
def get_template_row(self):
"""
This is a template of a row.
"""
return {'type': '',
'action_type': 'Status',
'document_name': '',
'status_receiver': self.status_receiver,
'status_supplier': self.status_supplier,
'status_receiver_created': '',
'status_supplier_created': '',
'trade_supplier': '',
'trade_receiver': self.act.author.email,
'trade_confirmed': '',
'trade_weight': 0,
'action_create_by': self.action_create_by,
'devicehubID': self.devicehub_id,
'hid': self.hid,
'finalUserCode': '',
'numEndUsers': 0,
'liveCreate': 0,
'usageTimeHdd': self.lifetime,
'created': self.act.created,
'start': '',
'usageTimeAllocate': 0}
def get_metrics(self):
"""
This method get a list of values for calculate a metrics from a spreadsheet
"""
return self.rows
class Metrics(MetricsMix):
"""we want get the data metrics of one device"""
def __init__(self, *args, **kwargs):
self.device = kwargs.pop('device')
self.actions = copy.copy(self.device.actions)
super().__init__(*args, **kwargs)
self.hid = self.device.hid
self.devicehub_id = self.device.devicehub_id
def get_action_status(self):
"""
Mark the status of one device.
If exist one trade before this action, then modify the trade action
else, create one row new.
"""
self.status_receiver = self.act.type
self.status_supplier = ''
if self.act.author != self.act.rol_user:
# It is neccesary exist one trade action before
self.last_trade['status_supplier'] = self.act.type
self.last_trade['status_supplier_created'] = self.act.created
return
self.action_create_by = 'Receiver'
if self.last_trade:
# if exist one trade action before
self.last_trade['status_receiver'] = self.act.type
self.last_trade['status_receiver_created'] = self.act.created
return
# If not exist any trade action for this device
row = self.get_template_row()
row['status_receiver_created'] = self.act.created
self.rows.append(row)
def get_snapshot(self):
"""
If there are one snapshot get the last lifetime for to do a calcul of time of use.
"""
lifestimes = self.act.get_last_lifetimes()
if lifestimes:
self.lifetime = lifestimes[0]['lifetime']
def get_allocate(self):
"""
If the action is one Allocate, need modify the row base.
"""
self.end_users = self.act.end_users
self.final_user_code = self.act.final_user_code
row = self.get_template_row()
row['type'] = 'Allocate'
row['trade_supplier'] = ''
row['finalUserCode'] = self.final_user_code
row['numEndUsers'] = self.end_users
row['start'] = self.act.start_time
row['usageTimeAllocate'] = self.lifetime
self.rows.append(row)
def get_live(self):
"""
If the action is one Live, need modify the row base.
"""
row = self.get_template_row()
row['type'] = 'Live'
row['finalUserCode'] = self.final_user_code
row['numEndUsers'] = self.end_users
row['start'] = self.act.start_time
row['usageTimeAllocate'] = self.lifetime
row['liveCreate'] = self.act.created
if self.act.usage_time_hdd:
row['usageTimeHdd'] = self.act.usage_time_hdd.total_seconds() / 3600
self.rows.append(row)
def get_deallocate(self):
"""
If the action is one Dellocate, need modify the row base.
"""
row = self.get_template_row()
row['type'] = 'Deallocate'
row['start'] = self.act.start_time
self.rows.append(row)
def get_confirms(self):
"""
if the action is one trade action, is possible than have a list of confirmations.
Get the doble confirm for to know if this trade is confirmed or not.
"""
if hasattr(self.act, 'acceptances'):
accept = self.act.acceptances[-1]
if accept.t == 'Confirm' and accept.user == self.act.user_to:
return True
return False
def get_trade(self):
"""
If this action is a trade action modify the base row.
"""
if self.act.author == self.act.user_from:
self.action_create_by = 'Supplier'
row = self.get_template_row()
self.last_trade = row
row['type'] = 'Trade'
row['action_type'] = 'Trade'
row['trade_supplier'] = self.act.user_from.email
row['trade_receiver'] = self.act.user_to.email
row['self.status_receiver'] = self.status_receiver
row['self.status_supplier'] = self.status_supplier
row['trade_confirmed'] = self.get_confirms()
self.rows.append(row)
def get_metrics(self):
"""
This method get a list of values for calculate a metrics from a spreadsheet
"""
for act in self.actions:
self.act = act
if act.type in ['Use', 'Refurbish', 'Recycling', 'Management']:
self.get_action_status()
continue
if act.type == 'Snapshot':
self.get_snapshot()
continue
if act.type == 'Allocate':
self.get_allocate()
continue
if act.type == 'Live':
self.get_live()
continue
if act.type == 'Deallocate':
self.get_deallocate()
continue
if act.type == 'Trade':
self.get_trade()
continue
return self.rows
class TradeMetrics(MetricsMix):
"""we want get the data metrics of one device"""
def __init__(self, *args, **kwargs):
self.document = kwargs.pop('document')
self.actions = copy.copy(self.document.actions)
self.hid = self.document.file_hash
self.devicehub_id = ''
super().__init__(*args, **kwargs)
def get_metrics(self):
self.last_trade = next(x for x in self.actions if x.t == 'Trade')
self.act = self.last_trade
row = self.get_template_row()
row['type'] = 'Trade-Document'
row['action_type'] = 'Trade-Document'
if self.document.weight:
row['type'] = 'Trade-Container'
row['action_type'] = 'Trade-Container'
row['document_name'] = self.document.file_name
row['trade_supplier'] = self.last_trade.user_from.email
row['trade_receiver'] = self.last_trade.user_to.email
row['trade_confirmed'] = self.get_confirms()
row['status_receiver'] = ''
row['status_supplier'] = ''
row['trade_weight'] = self.document.weight
if self.last_trade.author == self.last_trade.user_from:
row['action_create_by'] = 'Supplier'
elif self.last_trade.author == self.last_trade.user_to:
row['action_create_by'] = 'Receiver'
self.rows.append(row)
return self.rows
def get_confirms(self):
"""
if the action is one trade action, is possible than have a list of confirmations.
Get the doble confirm for to know if this trade is confirmed or not.
"""
if hasattr(self.last_trade, 'acceptances_document'):
accept = self.last_trade.acceptances_document[-1]
if accept.t == 'Confirm' and accept.user == self.last_trade.user_to:
return True
return False

View File

@ -34,6 +34,7 @@ from ereuse_devicehub.resources.enums import BatteryTechnology, CameraFacing, Co
DataStorageInterface, DisplayTech, PrinterTechnology, RamFormat, RamInterface, Severity, TransferState DataStorageInterface, DisplayTech, PrinterTechnology, RamFormat, RamInterface, Severity, TransferState
from ereuse_devicehub.resources.models import STR_SM_SIZE, Thing, listener_reset_field_updated_in_actual_time from ereuse_devicehub.resources.models import STR_SM_SIZE, Thing, listener_reset_field_updated_in_actual_time
from ereuse_devicehub.resources.user.models import User from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.device.metrics import Metrics
def create_code(context): def create_code(context):
@ -505,52 +506,8 @@ class Device(Thing):
""" """
This method get a list of values for calculate a metrics from a spreadsheet This method get a list of values for calculate a metrics from a spreadsheet
""" """
actions = copy.copy(self.actions) metrics = Metrics(device=self)
actions.sort(key=lambda x: x.created) return metrics.get_metrics()
allocates = []
lifetime = 0
for act in actions:
if act.type == 'Snapshot':
snapshot = act
lifestimes = snapshot.get_last_lifetimes()
lifetime = 0
if lifestimes:
lifetime = lifestimes[0]['lifetime']
if act.type == 'Allocate':
allo = {'type': 'Allocate',
'devicehubID': self.devicehub_id,
'finalUserCode': act.final_user_code,
'numEndUsers': act.end_users,
'hid': self.hid,
'liveCreate': 0,
'usageTimeHdd': 0,
'start': act.start_time,
'usageTimeAllocate': lifetime}
allocates.append(allo)
if act.type == 'Live':
allocate = copy.copy(allo)
allocate['type'] = 'Live'
allocate['liveCreate'] = act.created
allocate['usageTimeHdd'] = 0
if act.usage_time_hdd:
allocate['usageTimeHdd'] = act.usage_time_hdd.total_seconds()/3600
allocates.append(allocate)
if act.type == 'Deallocate':
deallo = {'type': 'Deallocate',
'devicehubID': self.devicehub_id,
'finalUserCode': '',
'numEndUsers': '',
'hid': self.hid,
'liveCreate': 0,
'usageTimeHdd': lifetime,
'start': act.start_time,
'usageTimeAllocate': 0}
allocates.append(deallo)
return allocates
def __lt__(self, other): def __lt__(self, other):
return self.id < other.id return self.id < other.id

View File

@ -414,6 +414,8 @@ def none2str(string):
return '' return ''
return format(string) return format(string)
def get_action(component, action): def get_action(component, action):
""" Filter one action from a component or return None """ """ Filter one action from a component or return None """
result = [a for a in component.actions if a.type == action] result = [a for a in component.actions if a.type == action]
@ -427,9 +429,21 @@ class ActionRow(OrderedDict):
# General information about allocates, deallocate and lives # General information about allocates, deallocate and lives
self['DHID'] = allocate['devicehubID'] self['DHID'] = allocate['devicehubID']
self['Hid'] = allocate['hid'] self['Hid'] = allocate['hid']
self['Start'] = allocate['start'] self['Document-Name'] = allocate['document_name']
self['FinalUserCode'] = allocate['finalUserCode'] self['Action-Type'] = allocate['action_type']
self['NumEndUsers'] = allocate['numEndUsers'] self['Action-User-LastOwner-Supplier'] = allocate['trade_supplier']
self['Action-User-LastOwner-Receiver'] = allocate['trade_receiver']
self['Action-Create-By'] = allocate['action_create_by']
self['Trade-Confirmed'] = allocate['trade_confirmed']
self['Status-Supplier'] = allocate['status_supplier']
self['Status-Receiver'] = allocate['status_receiver']
self['Status Supplier Created Date'] = allocate['status_supplier_created']
self['Status Receiver Created Date'] = allocate['status_receiver_created']
self['Trade-Weight'] = allocate['trade_weight']
self['Action-Create'] = allocate['created']
self['Allocate-Start'] = allocate['start']
self['Allocate-User-Code'] = allocate['finalUserCode']
self['Allocate-NumUsers'] = allocate['numEndUsers']
self['UsageTimeAllocate'] = allocate['usageTimeAllocate'] self['UsageTimeAllocate'] = allocate['usageTimeAllocate']
self['Type'] = allocate['type'] self['Type'] = allocate['type']
self['LiveCreate'] = allocate['liveCreate'] self['LiveCreate'] = allocate['liveCreate']

View File

@ -3,11 +3,9 @@ import enum
import uuid import uuid
import time import time
import datetime import datetime
import pathlib
from collections import OrderedDict from collections import OrderedDict
from io import StringIO from io import StringIO
from typing import Callable, Iterable, Tuple from typing import Callable, Iterable, Tuple
from decouple import config
import boltons import boltons
import flask import flask
@ -32,6 +30,8 @@ from ereuse_devicehub.resources.documents.device_row import (DeviceRow, StockRow
InternalStatsRow) InternalStatsRow)
from ereuse_devicehub.resources.lot import LotView from ereuse_devicehub.resources.lot import LotView
from ereuse_devicehub.resources.lot.models import Lot from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.action.models import Trade
from ereuse_devicehub.resources.device.models import Device
from ereuse_devicehub.resources.hash_reports import insert_hash, ReportHash, verify_hash from ereuse_devicehub.resources.hash_reports import insert_hash, ReportHash, verify_hash
@ -90,7 +90,6 @@ class DocumentView(DeviceView):
res = flask.make_response(template) res = flask.make_response(template)
return res return res
@staticmethod @staticmethod
def erasure(query: db.Query): def erasure(query: db.Query):
def erasures(): def erasures():
@ -151,7 +150,7 @@ class DevicesDocumentView(DeviceView):
class ActionsDocumentView(DeviceView): class ActionsDocumentView(DeviceView):
@cache(datetime.timedelta(minutes=1)) @cache(datetime.timedelta(minutes=1))
def find(self, args: dict): def find(self, args: dict):
query = (x for x in self.query(args) if x.owner_id == g.user.id) query = (x for x in self.query(args))
return self.generate_post_csv(query) return self.generate_post_csv(query)
def generate_post_csv(self, query): def generate_post_csv(self, query):
@ -159,13 +158,26 @@ class ActionsDocumentView(DeviceView):
data = StringIO() data = StringIO()
cw = csv.writer(data, delimiter=';', lineterminator="\n", quotechar='"') cw = csv.writer(data, delimiter=';', lineterminator="\n", quotechar='"')
first = True first = True
devs_id = []
for device in query: for device in query:
devs_id.append(device.id)
for allocate in device.get_metrics(): for allocate in device.get_metrics():
d = ActionRow(allocate) d = ActionRow(allocate)
if first: if first:
cw.writerow(d.keys()) cw.writerow(d.keys())
first = False first = False
cw.writerow(d.values()) cw.writerow(d.values())
query_trade = Trade.query.filter(Trade.devices.any(Device.id.in_(devs_id))).all()
for trade in query_trade:
data_rows = trade.get_metrics()
for row in data_rows:
d = ActionRow(row)
if first:
cw.writerow(d.keys())
first = False
cw.writerow(d.values())
bfile = data.getvalue().encode('utf-8') bfile = data.getvalue().encode('utf-8')
output = make_response(bfile) output = make_response(bfile)
insert_hash(bfile) insert_hash(bfile)
@ -185,11 +197,11 @@ class LotsDocumentView(LotView):
cw = csv.writer(data) cw = csv.writer(data)
first = True first = True
for lot in query: for lot in query:
l = LotRow(lot) _lot = LotRow(lot)
if first: if first:
cw.writerow(l.keys()) cw.writerow(_lot.keys())
first = False first = False
cw.writerow(l.values()) cw.writerow(_lot.values())
bfile = data.getvalue().encode('utf-8') bfile = data.getvalue().encode('utf-8')
output = make_response(bfile) output = make_response(bfile)
insert_hash(bfile) insert_hash(bfile)
@ -275,7 +287,7 @@ class StampsView(View):
ok = '100% coincidence. The attached file contains data 100% existing in \ ok = '100% coincidence. The attached file contains data 100% existing in \
to our backend' to our backend'
result = ('Bad', bad) result = ('Bad', bad)
mime = ['text/csv', 'application/pdf', 'text/plain','text/markdown', mime = ['text/csv', 'application/pdf', 'text/plain', 'text/markdown',
'image/jpeg', 'image/png', 'text/html', 'image/jpeg', 'image/png', 'text/html',
'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
'application/vnd.oasis.opendocument.spreadsheet', 'application/vnd.oasis.opendocument.spreadsheet',
@ -304,9 +316,9 @@ class InternalStatsView(DeviceView):
create = '{}-{}'.format(ac.created.year, ac.created.month) create = '{}-{}'.format(ac.created.year, ac.created.month)
user = ac.author.email user = ac.author.email
if not user in d: if user not in d:
d[user] = {} d[user] = {}
if not create in d[user]: if create not in d[user]:
d[user][create] = [] d[user][create] = []
d[user][create].append(ac) d[user][create].append(ac)
@ -434,4 +446,3 @@ class DocumentDef(Resource):
auth=app.auth) auth=app.auth)
wbconf_view = app.auth.requires_auth(wbconf_view) wbconf_view = app.auth.requires_auth(wbconf_view)
self.add_url_rule('/wbconf/<string:wbtype>', view_func=wbconf_view, methods=get) self.add_url_rule('/wbconf/<string:wbtype>', view_func=wbconf_view, methods=get)

View File

@ -2,6 +2,9 @@ import pytest
from ereuse_devicehub.client import UserClient from ereuse_devicehub.client import UserClient
from ereuse_devicehub.resources.action import models as ma from ereuse_devicehub.resources.action import models as ma
from ereuse_devicehub.resources.documents import documents
from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.tradedocument.models import TradeDocument
from tests import conftest from tests import conftest
from tests.conftest import file, yaml2json, json_encode from tests.conftest import file, yaml2json, json_encode
@ -20,8 +23,7 @@ def test_simple_metrics(user: UserClient):
"finalUserCode": "abcdefjhi", "finalUserCode": "abcdefjhi",
"devices": [device_id], "description": "aaa", "devices": [device_id], "description": "aaa",
"startTime": "2020-11-01T02:00:00+00:00", "startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00" "endTime": "2020-12-01T02:00:00+00:00"}
}
# Create Allocate # Create Allocate
user.post(res=ma.Allocate, data=post_request) user.post(res=ma.Allocate, data=post_request)
@ -65,8 +67,7 @@ def test_second_hdd_metrics(user: UserClient):
"finalUserCode": "abcdefjhi", "finalUserCode": "abcdefjhi",
"devices": [device_id], "description": "aaa", "devices": [device_id], "description": "aaa",
"startTime": "2020-11-01T02:00:00+00:00", "startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00" "endTime": "2020-12-01T02:00:00+00:00"}
}
# Create Allocate # Create Allocate
user.post(res=ma.Allocate, data=post_request) user.post(res=ma.Allocate, data=post_request)
@ -109,8 +110,7 @@ def test_metrics_with_live_null(user: UserClient):
"finalUserCode": "abcdefjhi", "finalUserCode": "abcdefjhi",
"devices": [device_id], "description": "aaa", "devices": [device_id], "description": "aaa",
"startTime": "2020-11-01T02:00:00+00:00", "startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00" "endTime": "2020-12-01T02:00:00+00:00"}
}
# Create Allocate # Create Allocate
user.post(res=ma.Allocate, data=post_request) user.post(res=ma.Allocate, data=post_request)
@ -120,3 +120,180 @@ def test_metrics_with_live_null(user: UserClient):
res, _ = user.get("/metrics/") res, _ = user.get("/metrics/")
assert res == metrics assert res == metrics
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_metrics_action_status(user: UserClient, user2: UserClient):
""" Checks one standard query of metrics."""
# Insert computer
lenovo = yaml2json('desktop-9644w8n-lenovo-0169622.snapshot')
snap, _ = user.post(json_encode(lenovo), res=ma.Snapshot)
action = {'type': ma.Use.t, 'devices': [snap['device']['id']]}
action_use, _ = user.post(action, res=ma.Action)
csv_str, _ = user.get(res=documents.DocumentDef.t,
item='actions/',
accept='text/csv',
query=[('filter', {'type': ['Computer']})])
head = 'DHID;Hid;Document-Name;Action-Type;Action-User-LastOwner-Supplier;Action-User-LastOwner-Receiver;Action-Create-By;Trade-Confirmed;Status-Supplier;Status-Receiver;Status Supplier Created Date;Status Receiver Created Date;Trade-Weight;Action-Create;Allocate-Start;Allocate-User-Code;Allocate-NumUsers;UsageTimeAllocate;Type;LiveCreate;UsageTimeHdd\n'
body = '93652;desktop-lenovo-9644w8n-0169622-00:1a:6b:5e:7f:10;;Status;;foo@foo.com;Receiver;;;Use;;'
assert head in csv_str
assert body in csv_str
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_complet_metrics_with_trade(user: UserClient, user2: UserClient):
""" Checks one standard query of metrics in a trade enviroment."""
# Insert computer
lenovo = yaml2json('desktop-9644w8n-lenovo-0169622.snapshot')
acer = yaml2json('acer.happy.battery.snapshot')
snap1, _ = user.post(json_encode(lenovo), res=ma.Snapshot)
snap2, _ = user.post(json_encode(acer), res=ma.Snapshot)
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
devices = [('id', snap1['device']['id']),
('id', snap2['device']['id'])]
lot, _ = user.post({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=devices)
request_post = {
'type': 'Trade',
'devices': [snap1['device']['id'], snap2['device']['id']],
'userFromEmail': user.email,
'userToEmail': user2.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'lot': lot['id'],
'confirms': True,
}
user.post(res=ma.Action, data=request_post)
action = {'type': ma.Refurbish.t, 'devices': [snap1['device']['id']]}
action_use, _ = user.post(action, res=ma.Action)
csv_str, _ = user.get(res=documents.DocumentDef.t,
item='actions/',
accept='text/csv',
query=[('filter', {'type': ['Computer']})])
body1_lenovo = '93652;desktop-lenovo-9644w8n-0169622-00:1a:6b:5e:7f:10;;Trade;foo@foo.com;foo2@foo.com;Supplier;False;Refurbish;Use;'
body2_lenovo = ';;0;0;Trade;0;0\n'
body1_acer = 'J2MA2;laptop-acer-aohappy-lusea0d010038879a01601-00:26:c7:8e:cb:8c;;Trade;foo@foo.com;foo2@foo.com;Supplier;False;;Use;;;0;'
body2_acer = ';;0;0;Trade;0;4692.0\n'
assert body1_lenovo in csv_str
assert body2_lenovo in csv_str
assert body1_acer in csv_str
assert body2_acer in csv_str
# User2 mark this device as Refurbish
action = {'type': ma.Refurbish.t, 'devices': [snap1['device']['id']]}
action_use2, _ = user2.post(action, res=ma.Action)
csv_str, _ = user.get(res=documents.DocumentDef.t,
item='actions/',
accept='text/csv',
query=[('filter', {'type': ['Computer']})])
body2_lenovo = ';Refurbish;0;0;Trade;0;0\n'
body2_acer = ';Refurbish;0;0;Trade;0;4692.0\n'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_metrics_action_status_for_containers(user: UserClient, user2: UserClient):
""" Checks one standard query of metrics for a container."""
# Insert computer
lenovo = yaml2json('desktop-9644w8n-lenovo-0169622.snapshot')
snap, _ = user.post(json_encode(lenovo), res=ma.Snapshot)
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
devices = [('id', snap['device']['id'])]
lot, _ = user.post({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=devices)
request_post = {
'type': 'Trade',
'devices': [snap['device']['id']],
'userFromEmail': user.email,
'userToEmail': user2.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'lot': lot['id'],
'confirms': True,
}
user.post(res=ma.Action, data=request_post)
request_post = {
'filename': 'test.pdf',
'hash': 'bbbbbbbb',
'url': 'http://www.ereuse.org/',
'weight': 150,
'lot': lot['id']
}
tradedocument, _ = user.post(res=TradeDocument, data=request_post)
action = {'type': ma.Recycling.t, 'devices': [], 'documents': [tradedocument['id']]}
action, _ = user.post(action, res=ma.Action)
trade = TradeDocument.query.one()
assert str(trade.actions[-1].id) == action['id']
csv_str, _ = user.get(res=documents.DocumentDef.t,
item='actions/',
accept='text/csv',
query=[('filter', {'type': ['Computer']})])
body1 = ';bbbbbbbb;test.pdf;Trade-Container;foo@foo.com;foo2@foo.com;Supplier;False;;;;;150.0;'
body2 = ';;0;0;Trade-Container;0;0'
assert len(csv_str.split('\n')) == 4
assert body1 in csv_str.split('\n')[-2]
assert body2 in csv_str.split('\n')[-2]
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_visual_metrics_for_old_owners(user: UserClient, user2: UserClient):
""" Checks if one old owner can see the metrics in a trade enviroment."""
# Insert computer
lenovo = yaml2json('desktop-9644w8n-lenovo-0169622.snapshot')
snap1, _ = user.post(json_encode(lenovo), res=ma.Snapshot)
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
devices = [('id', snap1['device']['id'])]
lot, _ = user.post({},
res=Lot,
item='{}/devices'.format(lot['id']),
query=devices)
request_post = {
'type': 'Trade',
'devices': [snap1['device']['id']],
'userFromEmail': user.email,
'userToEmail': user2.email,
'price': 10,
'date': "2020-12-01T02:00:00+00:00",
'lot': lot['id'],
'confirms': True,
}
trade, _ = user.post(res=ma.Action, data=request_post)
request_confirm = {
'type': 'Confirm',
'action': trade['id'],
'devices': [snap1['device']['id']]
}
user2.post(res=ma.Action, data=request_confirm)
action = {'type': ma.Refurbish.t, 'devices': [snap1['device']['id']]}
action_use, _ = user.post(action, res=ma.Action)
csv_supplier, _ = user.get(res=documents.DocumentDef.t,
item='actions/',
accept='text/csv',
query=[('filter', {'type': ['Computer']})])
csv_receiver, _ = user2.get(res=documents.DocumentDef.t,
item='actions/',
accept='text/csv',
query=[('filter', {'type': ['Computer']})])
body = ';;0;0;Trade;0;0\n'
assert body in csv_receiver
assert body in csv_supplier