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

Merge pull request #175 from eReuse/feature/generate-name-tag 

adding tag when there are a new device
This commit is contained in:
cayop 2021-10-23 15:22:39 +02:00 committed by GitHub
parent 9972c1666f 3531f607da
commit 12102d8890
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 93 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
ereuse_devicehub/resources/device/models.py
View File

@ -6,12 +6,11 @@ from fractions import Fraction
from itertools import chain
from operator import attrgetter
from typing import Dict, List, Set
from flask_sqlalchemy import event
from boltons import urlutils
from citext import CIText
from flask_sqlalchemy import event
from ereuse_utils.naming import HID_CONVERSION_DOC, Naming
from flask import g
from more_itertools import unique_everseen
from sqlalchemy import BigInteger, Boolean, Column, Enum as DBEnum, Float, ForeignKey, Integer, \
Sequence, SmallInteger, Unicode, inspect, text
@ -38,8 +37,8 @@ from ereuse_devicehub.resources.device.metrics import Metrics
def create_code(context):
_id = Device.query.order_by(Device.id.desc()).first() or 1
if not _id == 1:
_id = Device.query.order_by(Device.id.desc()).first() or 3
if not _id == 3:
_id = _id.id + 1
return hashcode.encode(_id)
@ -231,7 +230,7 @@ class Device(Thing):
:return a list of actions:
"""
hide_actions = ['Price', 'EreusePrice']
actions = [ac for ac in self.actions if not ac.t in hide_actions]
actions = [ac for ac in self.actions if ac.t not in hide_actions]
actions.reverse()
return actions
@ -288,7 +287,7 @@ class Device(Thing):
status_actions = [ac.t for ac in states.Status.actions()]
history = []
for ac in self.actions:
if not ac.t in status_actions:
if ac.t not in status_actions:
continue
if not history:
history.append(ac)
@ -319,26 +318,26 @@ class Device(Thing):
# 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
# 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:
@ -1167,3 +1166,15 @@ class Manufacturer(db.Model):
listener_reset_field_updated_in_actual_time(Device)
def create_code_tag(mapper, connection, device):
"""
This function create a new tag every time than one device is create.
this tag is the same of devicehub_id.
"""
from ereuse_devicehub.resources.tag.model import Tag
tag = Tag(device_id=device.id, id=device.devicehub_id)
db.session.add(tag)
event.listen(Device, 'after_insert', create_code_tag, propagate=True)

2
tests/files/basic.csv
View File

File diff suppressed because one or more lines are too long

4
tests/files/proposal_extended_csv_report.csv
View File

File diff suppressed because one or more lines are too long

14
tests/test_device.py
View File

@ -309,9 +309,10 @@ def test_sync_execute_register_no_hid_tag_not_linked(tag_id: str):
# they are not the same tags though
# tag is a transient obj and db_tag the one from the db
# they have the same pk though
assert tag != db_tag, 'They are not the same tags though'
assert db_tag.id == tag.id
assert d.Desktop.query.one() == pc, 'd.Desktop had to be set to db'
assert tag != db_tag, 'They are not the same tags though'
for tag in pc.tags:
assert tag.id in ['foo', pc.devicehub_id]
@pytest.mark.mvp
@ -346,8 +347,9 @@ def test_sync_execute_register_tag_linked_same_device():
pc.tags.add(Tag(id='foo'))
db_pc = Sync().execute_register(pc)
assert db_pc.id == orig_pc.id
assert len(db_pc.tags) == 1
assert next(iter(db_pc.tags)).id == 'foo'
assert len(db_pc.tags) == 2
for tag in db_pc.tags:
assert tag.id in ['foo', db_pc.devicehub_id]
@pytest.mark.mvp
@ -399,6 +401,7 @@ def test_sync_execute_register_mismatch_between_tags_and_hid():
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_get_device(user: UserClient):
"""Checks GETting a d.Desktop with its components."""
g.user = User.query.one()
pc = d.Desktop(model='p1mo',
manufacturer='p1ma',
serial_number='p1s',
@ -437,6 +440,7 @@ def test_get_device(user: UserClient):
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_get_devices(app: Devicehub, user: UserClient):
"""Checks GETting multiple devices."""
g.user = User.query.one()
pc = d.Desktop(model='p1mo',
manufacturer='p1ma',
serial_number='p1s',
@ -604,6 +608,7 @@ def test_device_public(user: UserClient, client: Client):
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_computer_accessory_model(user: UserClient):
g.user = User.query.one()
sai = d.SAI(owner_id=user.user['id'])
db.session.add(sai)
keyboard = d.Keyboard(layout=Layouts.ES, owner_id=user.user['id'])
@ -616,6 +621,7 @@ def test_computer_accessory_model(user: UserClient):
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_networking_model(user: UserClient):
g.user = User.query.one()
router = d.Router(speed=1000, wireless=True, owner_id=user.user['id'])
db.session.add(router)
switch = d.Switch(speed=1000, wireless=False, owner_id=user.user['id'])

2
tests/test_device_find.py
View File

@ -183,7 +183,7 @@ def test_device_query(user: UserClient):
pc = next(d for d in i['items'] if d['type'] == 'Desktop')
assert len(pc['actions']) == 4
assert len(pc['components']) == 3
assert not pc['tags']
assert pc['tags'][0]['id'] == pc['devicehubID']
@pytest.mark.mvp

5
tests/test_documents.py
View File

@ -251,7 +251,7 @@ def test_check_insert_hash(app: Devicehub, user: UserClient, client: Client):
assert ReportHash.query.filter_by(hash3=hash3).count() == 1
result, status = client.get(res=documents.DocumentDef.t, item='check/', query=[('hash', hash3)])
assert status.status_code == 200
assert result == True
assert result
ff = open('/tmp/test.csv', 'w')
ff.write(csv_str)
@ -268,10 +268,7 @@ def test_export_extended(app: Devicehub, user: UserClient):
snapshot2, _ = user.post(file('complete.export.snapshot'), res=Snapshot, status=201)
with app.app_context():
# Create a pc with a tag
tag = Tag(id='foo', owner_id=user.user['id'])
# pc = Desktop(serial_number='sn1', chassis=ComputerChassis.Tower, owner_id=user.user['id'])
pc = d.Device.query.filter_by(id=snapshot1['device']['id']).first()
pc.tags.add(tag)
db.session.add(pc)
db.session.commit()

3
tests/test_lot.py
View File

@ -361,6 +361,7 @@ def test_lot_post_add_remove_device_view(app: Devicehub, user: UserClient):
"""
# todo check with components
with app.app_context():
g.user = User.query.one()
device = Desktop(serial_number='foo',
model='bar',
manufacturer='foobar',
@ -391,9 +392,11 @@ def test_lot_post_add_remove_device_view(app: Devicehub, user: UserClient):
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_lot_error_add_device_from_other_user(user: UserClient):
# TODO
"""Tests adding a device to a lot using POST and
removing it with DELETE.
"""
g.user = User.query.one()
user2 = User(email='baz@baz.cxm', password='baz')
user2.individuals.add(Person(name='Tommy'))
db.session.add(user2)

4
tests/test_metrics.py
View File

@ -135,7 +135,7 @@ def test_metrics_action_status(user: UserClient, user2: UserClient):
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;;'
body = 'O48N2;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
@ -176,7 +176,7 @@ def test_complet_metrics_with_trade(user: UserClient, user2: UserClient):
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;'
body1_lenovo = 'O48N2;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;'

5
tests/test_snapshot.py
View File

@ -37,7 +37,6 @@ from tests import conftest
@pytest.mark.mvp
@pytest.mark.usefixtures('auth_app_context')
# cayop
def test_snapshot_model():
"""Tests creating a Snapshot with its relationships ensuring correct
DB mapping.
@ -318,7 +317,7 @@ def test_snapshot_tag_inner_tag(user: UserClient, tag_id: str, app: Devicehub):
snapshot_and_check(user, b,
action_types=(RateComputer.t, BenchmarkProcessor.t, VisualTest.t))
with app.app_context():
tag = Tag.query.one() # type: Tag
tag = Tag.query.all()[0] # type: Tag
assert tag.device_id == 3, 'Tag should be linked to the first device'
@ -358,7 +357,7 @@ def test_snapshot_different_properties_same_tags(user: UserClient, tag_id: str):
def test_snapshot_upload_twice_uuid_error(user: UserClient):
pc1 = file('basic.snapshot')
user.post(pc1, res=Snapshot)
user.post(pc1, res=Snapshot, status=UniqueViolation)
user.post(pc1, res=Snapshot, status=400)
@pytest.mark.mvp

24
tests/test_tag.py
View File

@ -2,6 +2,7 @@ import pathlib
import pytest
import requests_mock
from flask import g
from boltons.urlutils import URL
from ereuse_utils.session import DevicehubClient
from pytest import raises
@ -11,6 +12,7 @@ from teal.marshmallow import ValidationError
from ereuse_devicehub.client import UserClient, Client
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.action.models import Snapshot
from ereuse_devicehub.resources.agent.models import Organization
from ereuse_devicehub.resources.device.models import Desktop, Device
@ -19,7 +21,7 @@ from ereuse_devicehub.resources.tag import Tag
from ereuse_devicehub.resources.tag.view import CannotCreateETag, LinkedToAnotherDevice, \
TagNotLinked
from tests import conftest
from tests.conftest import file, yaml2json, json_encode
from tests.conftest import yaml2json, json_encode
@pytest.mark.mvp
@ -41,6 +43,7 @@ def test_create_tag(user: UserClient):
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_create_tag_with_device(user: UserClient):
"""Creates a tag specifying linked with one device."""
g.user = User.query.one()
pc = Desktop(serial_number='sn1', chassis=ComputerChassis.Tower, owner_id=user.user['id'])
db.session.add(pc)
db.session.commit()
@ -60,13 +63,14 @@ def test_create_tag_with_device(user: UserClient):
def test_delete_tags(user: UserClient, client: Client):
"""Delete a named tag."""
# Delete Tag Named
g.user = User.query.one()
pc = Desktop(serial_number='sn1', chassis=ComputerChassis.Tower, owner_id=user.user['id'])
db.session.add(pc)
db.session.commit()
tag = Tag(id='bar', owner_id=user.user['id'], device_id=pc.id)
db.session.add(tag)
db.session.commit()
tag = Tag.query.one()
tag = Tag.query.all()[-1]
assert tag.id == 'bar'
# Is not possible delete one tag linked to one device
res, _ = user.delete(res=Tag, item=tag.id, status=422)
@ -88,12 +92,12 @@ def test_delete_tags(user: UserClient, client: Client):
tag = Tag(id='bar-1', org=org, provider=URL('http://foo.bar'), owner_id=user.user['id'])
db.session.add(tag)
db.session.commit()
tag = Tag.query.one()
tag = Tag.query.all()[-1]
assert tag.id == 'bar-1'
res, _ = user.delete(res=Tag, item=tag.id, status=422)
msg = 'This tag {} is unnamed tag. It is imposible delete.'.format(tag.id)
assert msg in res['message']
tag = Tag.query.one()
tag = Tag.query.all()[-1]
assert tag.id == 'bar-1'
@ -182,6 +186,7 @@ def test_tag_get_device_from_tag_endpoint(app: Devicehub, user: UserClient):
"""Checks getting a linked device from a tag endpoint"""
with app.app_context():
# Create a pc with a tag
g.user = User.query.one()
tag = Tag(id='foo-bar', owner_id=user.user['id'])
pc = Desktop(serial_number='sn1', chassis=ComputerChassis.Tower, owner_id=user.user['id'])
pc.tags.add(tag)
@ -213,6 +218,7 @@ def test_tag_get_device_from_tag_endpoint_multiple_tags(app: Devicehub, user: Us
system should not return any of both (to be deterministic) so
it should raise an exception.
"""
g.user = User.query.all()[0]
db.session.add(Tag(id='foo', secondary='bar', owner_id=user.user['id']))
db.session.commit()
@ -276,6 +282,7 @@ def test_tag_manual_link_search(app: Devicehub, user: UserClient):
Checks search has the term.
"""
with app.app_context():
g.user = User.query.one()
db.session.add(Tag('foo-bar', secondary='foo-sec', owner_id=user.user['id']))
desktop = Desktop(serial_number='foo', chassis=ComputerChassis.AllInOne, owner_id=user.user['id'])
db.session.add(desktop)
@ -284,7 +291,7 @@ def test_tag_manual_link_search(app: Devicehub, user: UserClient):
devicehub_id = desktop.devicehub_id
user.put({}, res=Tag, item='foo-bar/device/{}'.format(desktop_id), status=204)
device, _ = user.get(res=Device, item=devicehub_id)
assert device['tags'][0]['id'] == 'foo-bar'
assert 'foo-bar' in [x['id'] for x in device['tags']]
# Device already linked
# Just returns an OK to conform to PUT as anything changes
@ -323,8 +330,8 @@ def test_tag_secondary_workbench_link_find(user: UserClient):
s['device']['tags'] = [{'id': 'foo', 'secondary': 'bar', 'type': 'Tag'}]
snapshot, _ = user.post(json_encode(s), res=Snapshot)
device, _ = user.get(res=Device, item=snapshot['device']['devicehubID'])
assert device['tags'][0]['id'] == 'foo'
assert device['tags'][0]['secondary'] == 'bar'
assert 'foo' in [x['id'] for x in device['tags']]
assert 'bar' in [x.get('secondary') for x in device['tags']]
r, _ = user.get(res=Device, query=[('search', 'foo'), ('filter', {'type': ['Computer']})])
assert len(r['items']) == 1
@ -412,6 +419,7 @@ def test_get_tag_permissions(app: Devicehub, user: UserClient, user2: UserClient
"""Creates a tag specifying a custom organization."""
with app.app_context():
# Create a pc with a tag
g.user = User.query.all()[0]
tag = Tag(id='foo-bar', owner_id=user.user['id'])
pc = Desktop(serial_number='sn1', chassis=ComputerChassis.Tower, owner_id=user.user['id'])
pc.tags.add(tag)
@ -424,5 +432,5 @@ def test_get_tag_permissions(app: Devicehub, user: UserClient, user2: UserClient
computer2, res2 = user2.get(url, None)
assert res.status_code == 200
assert res2.status_code == 200
assert len(computer['items']) == 1
assert len(computer['items']) == 2
assert len(computer2['items']) == 0

4
tests/test_workbench.py
View File

@ -67,7 +67,7 @@ def test_workbench_server_condensed(user: UserClient):
assert device['rate']['type'] == RateComputer.t
# TODO JN why haven't same order in actions on each execution?
assert device['actions'][2]['type'] == BenchmarkProcessor.t or device['actions'][2]['type'] == BenchmarkRamSysbench.t
assert device['tags'][0]['id'] == 'tag1'
assert 'tag1' in [x['id'] for x in device['tags']]
@pytest.mark.xfail(reason='Functionality not yet developed.')
@ -184,7 +184,7 @@ def test_snapshot_real_eee_1001pxd_with_rate(user: UserClient):
assert pc['serialNumber'] == 'b8oaas048286'
assert pc['manufacturer'] == 'asustek computer inc.'
assert pc['hid'] == 'laptop-asustek_computer_inc-1001pxd-b8oaas048286-14:da:e9:42:f6:7c'
assert pc['tags'] == []
assert len(pc['tags']) == 1
assert pc['networkSpeeds'] == [100, 0], 'Although it has WiFi we do not know the speed'
assert pc['rate']
rate = pc['rate']