Merge pull request #274 from eReuse/feature/#3396-add-columns-status

Feature/#3396 add columns status
This commit is contained in:
cayop 2022-05-16 10:50:48 +02:00 committed by GitHub
commit ac0d6c7a22
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 1191 additions and 873 deletions

View File

@ -332,6 +332,35 @@ class Device(Thing):
with suppress(LookupError, ValueError): with suppress(LookupError, ValueError):
return self.last_action_of(*states.Trading.actions()) return self.last_action_of(*states.Trading.actions())
@property
def allocated_status(self):
"""Show the actual status of device.
The status depend of one of this 3 actions:
- Allocate
- Deallocate
- InUse (Live register)
"""
from ereuse_devicehub.resources.device import states
with suppress(LookupError, ValueError):
return self.last_action_of(*states.Usage.actions())
@property
def physical_status(self):
"""Show the actual status of device for this owner.
The status depend of one of this 4 actions:
- ToPrepare
- Prepare
- DataWipe
- ToRepair
- Ready
"""
from ereuse_devicehub.resources.device import states
with suppress(LookupError, ValueError):
# import pdb; pdb.set_trace()
return self.last_action_of(*states.Physical.actions())
@property @property
def status(self): def status(self):
"""Show the actual status of device for this owner. """Show the actual status of device for this owner.
@ -560,6 +589,20 @@ class Device(Thing):
args[POLYMORPHIC_ON] = cls.type args[POLYMORPHIC_ON] = cls.type
return args return args
def is_status(self, action):
from ereuse_devicehub.resources.device import states
if action.type in states.Usage.__members__:
return "Allocate State: "
if action.type in states.Status.__members__:
return "Lifecycle State: "
if action.type in states.Physical.__members__:
return "Physical State: "
return ""
def set_hid(self): def set_hid(self):
with suppress(TypeError): with suppress(TypeError):
self.hid = Naming.hid( self.hid = Naming.hid(

View File

@ -33,6 +33,7 @@ class Trading(State):
:cvar ProductDisposed: The device has been removed :cvar ProductDisposed: The device has been removed
from the facility. It does not mean end-of-life. from the facility. It does not mean end-of-life.
""" """
Reserved = e.Reserve Reserved = e.Reserve
Trade = e.Trade Trade = e.Trade
Confirm = e.Confirm Confirm = e.Confirm
@ -50,16 +51,17 @@ class Trading(State):
class Physical(State): class Physical(State):
"""Physical states. """Physical states.
:cvar ToBeRepaired: The device has been selected for reparation. :cvar Repair: The device has been repaired.
:cvar Repaired: The device has been repaired. :cvar ToPrepare: The device is going to be or being prepared.
:cvar Preparing: The device is going to be or being prepared. :cvar Prepare: The device has been prepared.
:cvar Prepared: The device has been prepared.
:cvar Ready: The device is in working conditions. :cvar Ready: The device is in working conditions.
:cvar DataWipe: Do DataWipe over the device.
""" """
ToBeRepaired = e.ToRepair
Repaired = e.Repair ToPrepare = e.ToPrepare
Preparing = e.ToPrepare Prepare = e.Prepare
Prepared = e.Prepare DataWipe = e.DataWipe
ToRepair = e.ToRepair
Ready = e.Ready Ready = e.Ready
@ -68,22 +70,24 @@ class Traking(State):
:cvar Receive: The device changes hands :cvar Receive: The device changes hands
""" """
# Receive = e.Receive # Receive = e.Receive
pass pass
class Usage(State): class Usage(State):
"""Usage states. """Usage states.
:cvar Allocate: The device is allocate in other Agent (organization, person ...) :cvar Allocate: The device is allocate in other Agent (organization, person ...)
:cvar Deallocate: The device is deallocate and return to the owner :cvar Deallocate: The device is deallocate and return to the owner
:cvar InUse: The device is being reported to be in active use. :cvar InUse: The device is being reported to be in active use.
""" """
Allocate = e.Allocate Allocate = e.Allocate
Deallocate = e.Deallocate Deallocate = e.Deallocate
InUse = e.Live InUse = e.Live
class Status(State): class Status(State):
"""Define status of device for one user. """Define status of device for one user.
:cvar Use: The device is in use for one final user. :cvar Use: The device is in use for one final user.
@ -91,6 +95,7 @@ class Status(State):
:cvar Recycling: The device is sended to recycling. :cvar Recycling: The device is sended to recycling.
:cvar Management: The device is owned by one Manager. :cvar Management: The device is owned by one Manager.
""" """
Use = e.Use Use = e.Use
Refurbish = e.Refurbish Refurbish = e.Refurbish
Recycling = e.Recycling Recycling = e.Recycling

View File

@ -154,6 +154,37 @@
</tbody> </tbody>
</table> </table>
</div> </div>
<h4>Actual Status</h4>
<ol>
<li>
<strong>
Lifecycle Status
</strong>
{% if device.status %}
{{ device.status.type }}
{% endif %}
</li>
<li>
<strong>
Allocate Status
</strong>
{% if device.allocated_status %}
{{ device.allocated_status.type }}
{% endif %}
</li>
<li>
<strong>
Physical Status
</strong>
{% if device.physical_status %}
{{ device.physical_status.type }}
{% endif %}
</li>
</ol>
<h4>Public traceability log of the device</h4> <h4>Public traceability log of the device</h4>
<div class="text-right"> <div class="text-right">
<small>Latest one.</small> <small>Latest one.</small>
@ -162,10 +193,17 @@
{% for action in device.public_actions %} {% for action in device.public_actions %}
<li> <li>
<strong> <strong>
{{ action.type }} {{ device.is_status(action) }}
{% if not device.is_status(action) %}
{{ action.type }}
{% endif %}
</strong> </strong>
{{ action }} {% if device.is_status(action) %}
{{ action }} {{ action.type }}
{% else %}
{{ action }}
{% endif %}
<br> <br>
<div class="text-muted"> <div class="text-muted">
<small> <small>

View File

@ -70,16 +70,19 @@ class DeviceRow(OrderedDict):
self['Updated in (software)'] = device.updated self['Updated in (software)'] = device.updated
self['Updated in (web)'] = '' self['Updated in (web)'] = ''
self['Physical state'] = ''
if device.physical_status:
self['Physical state'] = device.physical_status.type
self['Allocate state'] = ''
if device.allocated_status:
self['Allocate state'] = device.allocated_status.type
try: try:
self['Physical state'] = device.last_action_of( self['Lifecycle state'] = device.last_action_of(
*states.Physical.actions()).t
except LookupError:
self['Physical state'] = ''
try:
self['Trading state'] = device.last_action_of(
*states.Trading.actions()).t *states.Trading.actions()).t
except LookupError: except LookupError:
self['Trading state'] = '' self['Lifecycle state'] = ''
if isinstance(device, d.Computer): if isinstance(device, d.Computer):
self['Processor'] = none2str(device.processor_model) self['Processor'] = none2str(device.processor_model)
self['RAM (MB)'] = none2str(device.ram_size) self['RAM (MB)'] = none2str(device.ram_size)
@ -367,15 +370,18 @@ class StockRow(OrderedDict):
self['Model'] = none2str(device.model) self['Model'] = none2str(device.model)
self['Manufacturer'] = none2str(device.manufacturer) self['Manufacturer'] = none2str(device.manufacturer)
self['Registered in'] = format(device.created, '%c') self['Registered in'] = format(device.created, '%c')
self['Physical state'] = ''
if device.physical_status:
self['Physical state'] = device.physical_status.type
self['Allocate state'] = ''
if device.allocated_status:
self['Allocate state'] = device.allocated_status.type
try: try:
self['Physical state'] = device.last_action_of( self['Lifecycle state'] = device.last_action_of(*states.Trading.actions()).t
*states.Physical.actions()).t
except LookupError: except LookupError:
self['Physical state'] = '' self['Lifecycle state'] = ''
try:
self['Trading state'] = device.last_action_of(*states.Trading.actions()).t
except LookupError:
self['Trading state'] = ''
self['Price'] = none2str(device.price) self['Price'] = none2str(device.price)
self['Processor'] = none2str(device.processor_model) self['Processor'] = none2str(device.processor_model)
self['RAM (MB)'] = none2str(device.ram_size) self['RAM (MB)'] = none2str(device.ram_size)

View File

@ -124,16 +124,28 @@
<div class="tab-pane fade profile-overview" id="status"> <div class="tab-pane fade profile-overview" id="status">
<h5 class="card-title">Status Details</h5> <h5 class="card-title">Status Details</h5>
<div class="row"> <div class="row">
<div class="col-lg-3 col-md-4 label">Physical States</div> <div class="col-lg-3 col-md-4 label">Physical State</div>
<div class="col-lg-9 col-md-8">{{ device.physical or '' }}</div> <div class="col-lg-9 col-md-8">
{% if device.physical_status %}
{{ device.physical_status.type }}
{% endif %}
</div>
</div> </div>
<div class="row"> <div class="row">
<div class="col-lg-3 col-md-4 label">Trading States</div> <div class="col-lg-3 col-md-4 label">Lifecycle State</div>
<div class="col-lg-9 col-md-8">{{ device.last_action_trading or ''}}</div> <div class="col-lg-9 col-md-8">
{% if device.status %}
{{ device.status.type }}
{% endif %}
</div>
</div> </div>
<div class="row"> <div class="row">
<div class="col-lg-3 col-md-4 label">Usage States</div> <div class="col-lg-3 col-md-4 label">Allocated State</div>
<div class="col-lg-9 col-md-8">{{ device.usage or '' }}</div> <div class="col-lg-9 col-md-8">
{% if device.allocated_status %}
{{ device.allocated_status.type }}
{% endif %}
</div>
</div> </div>
</div> </div>

View File

@ -338,7 +338,9 @@
<th scope="col">Title</th> <th scope="col">Title</th>
<th scope="col">DHID</th> <th scope="col">DHID</th>
<th scope="col">Unique Identifiers</th> <th scope="col">Unique Identifiers</th>
<th scope="col">Status</th> <th scope="col">Lifecycle Status</th>
<th scope="col">Allocated Status</th>
<th scope="col">Physical Status</th>
<th scope="col" data-type="date" data-format="DD-MM-YYYY">Update</th> <th scope="col" data-type="date" data-format="DD-MM-YYYY">Update</th>
</tr> </tr>
</thead> </thead>
@ -381,6 +383,8 @@
{% endfor %} {% endfor %}
</td> </td>
<td>{% if dev.status %}{{ dev.status.type }}{% endif %}</td> <td>{% if dev.status %}{{ dev.status.type }}{% endif %}</td>
<td>{% if dev.allocated_status %}{{ dev.allocated_status.type }}{% endif %}</td>
<td>{% if dev.physical_status %}{{ dev.physical_status.type }}{% endif %}</td>
<td>{{ dev.updated.strftime('%H:%M %d-%m-%Y') }}</td> <td>{{ dev.updated.strftime('%H:%M %d-%m-%Y') }}</td>
</tr> </tr>
{% endfor %} {% endfor %}

View File

@ -1,2 +1,2 @@
Type,Chassis,Serial Number,Model,Manufacturer,Registered in,Physical state,Trading state,Price,Processor,RAM (MB),Data Storage Size (MB),Rate,Range,Processor Rate,Processor Range,RAM Rate,RAM Range,Data Storage Rate,Data Storage Range Type;Chassis;Serial Number;Model;Manufacturer;Registered in;Physical state;Allocate state;Lifecycle state;Price;Processor;RAM (MB);Data Storage Size (MB);Rate;Range;Processor Rate;Processor Range;RAM Rate;RAM Range;Data Storage Rate;Data Storage Range
Desktop,Microtower,d1s,d1ml,d1mr,Tue Jul 2 10:35:10 2019,,,,p1ml,0,0,1.0,Very low,1.0,Very low,1.0,Very low,1.0,Very low Desktop;Microtower;d1s;d1ml;d1mr;Mon May 16 09:34:22 2022;;;;;p1ml;0;0;1.0;Very low;1.0;Very low;1.0;Very low;1.0;Very low

1 Type Chassis Serial Number Model Manufacturer Registered in Physical state Trading state Allocate state Lifecycle state Price Processor RAM (MB) Data Storage Size (MB) Rate Range Processor Rate Processor Range RAM Rate RAM Range Data Storage Rate Data Storage Range
2 Desktop Microtower d1s d1ml d1mr Tue Jul 2 10:35:10 2019 Mon May 16 09:34:22 2022 p1ml 0 0 1.0 Very low 1.0 Very low 1.0 Very low 1.0 Very low

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because it is too large Load Diff

View File

@ -1,30 +1,30 @@
import csv import csv
import hashlib import hashlib
from datetime import datetime from datetime import datetime
from io import StringIO, BytesIO from io import BytesIO, StringIO
from pathlib import Path from pathlib import Path
from flask import url_for
import pytest import pytest
from werkzeug.exceptions import Unauthorized
import teal.marshmallow import teal.marshmallow
from ereuse_utils.test import ANY from ereuse_utils.test import ANY
from flask import url_for
from werkzeug.exceptions import Unauthorized
from ereuse_devicehub import auth from ereuse_devicehub import auth
from ereuse_devicehub.client import Client, UserClient from ereuse_devicehub.client import Client, UserClient
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.user.models import Session from ereuse_devicehub.resources.action.models import Allocate, Live, Snapshot
from ereuse_devicehub.resources.action.models import Snapshot, Allocate, Live
from ereuse_devicehub.resources.documents import documents
from ereuse_devicehub.resources.tradedocument.models import TradeDocument
from ereuse_devicehub.resources.device import models as d from ereuse_devicehub.resources.device import models as d
from ereuse_devicehub.resources.documents import documents
from ereuse_devicehub.resources.enums import SessionType
from ereuse_devicehub.resources.hash_reports import ReportHash
from ereuse_devicehub.resources.lot.models import Lot from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.tag.model import Tag from ereuse_devicehub.resources.tag.model import Tag
from ereuse_devicehub.resources.hash_reports import ReportHash from ereuse_devicehub.resources.tradedocument.models import TradeDocument
from ereuse_devicehub.resources.enums import SessionType from ereuse_devicehub.resources.user.models import Session
from ereuse_devicehub.db import db
from tests import conftest from tests import conftest
from tests.conftest import file, yaml2json, json_encode from tests.conftest import file, json_encode, yaml2json
@pytest.mark.mvp @pytest.mark.mvp
@ -33,27 +33,30 @@ def test_erasure_certificate_public_one(user: UserClient, client: Client):
s = file('erase-sectors.snapshot') s = file('erase-sectors.snapshot')
snapshot, _ = user.post(s, res=Snapshot) snapshot, _ = user.post(s, res=Snapshot)
doc, response = user.get(res=documents.DocumentDef.t, doc, response = user.get(
item='erasures/{}'.format( res=documents.DocumentDef.t,
snapshot['device']['id']), item='erasures/{}'.format(snapshot['device']['id']),
accept=ANY) accept=ANY,
)
assert 'html' in response.content_type assert 'html' in response.content_type
assert '<html' in doc assert '<html' in doc
assert '2018' in doc assert '2018' in doc
doc, response = client.get(res=documents.DocumentDef.t, doc, response = client.get(
item='erasures/{}'.format( res=documents.DocumentDef.t,
snapshot['device']['id']), item='erasures/{}'.format(snapshot['device']['id']),
query=[('format', 'PDF')], query=[('format', 'PDF')],
accept='application/pdf') accept='application/pdf',
)
assert 'application/pdf' == response.content_type assert 'application/pdf' == response.content_type
erasure = next(e for e in snapshot['actions'] erasure = next(e for e in snapshot['actions'] if e['type'] == 'EraseSectors')
if e['type'] == 'EraseSectors')
doc, response = client.get(res=documents.DocumentDef.t, doc, response = client.get(
item='erasures/{}'.format(erasure['id']), res=documents.DocumentDef.t,
accept=ANY) item='erasures/{}'.format(erasure['id']),
accept=ANY,
)
assert 'html' in response.content_type assert 'html' in response.content_type
assert '<html' in doc assert '<html' in doc
assert '2018' in doc assert '2018' in doc
@ -67,30 +70,32 @@ def test_erasure_certificate_private_query(user: UserClient):
s = file('erase-sectors.snapshot') s = file('erase-sectors.snapshot')
snapshot, response = user.post(s, res=Snapshot) snapshot, response = user.post(s, res=Snapshot)
doc, response = user.get(res=documents.DocumentDef.t, doc, response = user.get(
item='erasures/', res=documents.DocumentDef.t,
query=[ item='erasures/',
('filter', {'ids': [snapshot['device']['id']]})], query=[('filter', {'ids': [snapshot['device']['id']]})],
accept=ANY) accept=ANY,
)
assert 'html' in response.content_type assert 'html' in response.content_type
assert '<html' in doc assert '<html' in doc
assert '2018' in doc assert '2018' in doc
doc, response = user.get(res=documents.DocumentDef.t, doc, response = user.get(
item='erasures/', res=documents.DocumentDef.t,
query=[ item='erasures/',
('filter', { query=[('filter', {'ids': [snapshot['device']['id']]}), ('format', 'PDF')],
'ids': [snapshot['device']['id']]}), accept='application/pdf',
('format', 'PDF') )
],
accept='application/pdf')
assert 'application/pdf' == response.content_type assert 'application/pdf' == response.content_type
@pytest.mark.mvp @pytest.mark.mvp
def test_erasure_certificate_wrong_id(client: Client): def test_erasure_certificate_wrong_id(client: Client):
client.get(res=documents.DocumentDef.t, item='erasures/this-is-not-an-id', client.get(
status=teal.marshmallow.ValidationError) res=documents.DocumentDef.t,
item='erasures/this-is-not-an-id',
status=teal.marshmallow.ValidationError,
)
@pytest.mark.mvp @pytest.mark.mvp
@ -98,20 +103,27 @@ def test_export_csv_permitions(user: UserClient, user2: UserClient, client: Clie
"""test export device information in a csv file with others users.""" """test export device information in a csv file with others users."""
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot) snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)
dev_id = snapshot['device']['id'] dev_id = snapshot['device']['id']
csv_user, _ = user.get(res=documents.DocumentDef.t, csv_user, _ = user.get(
item='devices/', res=documents.DocumentDef.t,
accept='text/csv', item='devices/',
query=[('filter', {'type': ['Computer'], 'ids': [dev_id]})]) accept='text/csv',
query=[('filter', {'type': ['Computer'], 'ids': [dev_id]})],
)
csv_user2, _ = user2.get(res=documents.DocumentDef.t, csv_user2, _ = user2.get(
item='devices/', res=documents.DocumentDef.t,
accept='text/csv', item='devices/',
query=[('filter', {'type': ['Computer'], 'ids': [dev_id]})]) accept='text/csv',
query=[('filter', {'type': ['Computer'], 'ids': [dev_id]})],
)
_, res = client.get(res=documents.DocumentDef.t, _, res = client.get(
item='devices/', res=documents.DocumentDef.t,
accept='text/csv', item='devices/',
query=[('filter', {'type': ['Computer'], 'ids': [dev_id]})], status=401) accept='text/csv',
query=[('filter', {'type': ['Computer'], 'ids': [dev_id]})],
status=401,
)
assert res.status_code == 401 assert res.status_code == 401
assert len(csv_user) > 0 assert len(csv_user) > 0
@ -124,36 +136,46 @@ def test_export_csv_actions(user: UserClient, user2: UserClient, client: Client)
acer = yaml2json('acer.happy.battery.snapshot') acer = yaml2json('acer.happy.battery.snapshot')
snapshot, _ = user.post(json_encode(acer), res=Snapshot) snapshot, _ = user.post(json_encode(acer), res=Snapshot)
device_id = snapshot['device']['id'] device_id = snapshot['device']['id']
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1, post_request = {
"devices": [device_id], "description": "aaa", "transaction": "ccc",
"finalUserCode": "abcdefjhi", "name": "John",
"startTime": "2020-11-01T02:00:00+00:00", "endUsers": 1,
"endTime": "2020-12-01T02:00:00+00:00" "devices": [device_id],
} "description": "aaa",
"finalUserCode": "abcdefjhi",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00",
}
user.post(res=Allocate, data=post_request) user.post(res=Allocate, data=post_request)
hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0] hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0]
hdd_action = [a for a in hdd['actions'] hdd_action = [a for a in hdd['actions'] if a['type'] == 'TestDataStorage'][0]
if a['type'] == 'TestDataStorage'][0]
hdd_action['lifetime'] += 1000 hdd_action['lifetime'] += 1000
acer.pop('elapsed') acer.pop('elapsed')
acer['licence_version'] = '1.0.0' acer['licence_version'] = '1.0.0'
snapshot, _ = client.post(acer, res=Live) snapshot, _ = client.post(acer, res=Live)
csv_user, _ = user.get(res=documents.DocumentDef.t, csv_user, _ = user.get(
item='actions/', res=documents.DocumentDef.t,
accept='text/csv', item='actions/',
query=[('filter', {'type': ['Computer'], 'ids': [device_id]})]) accept='text/csv',
query=[('filter', {'type': ['Computer'], 'ids': [device_id]})],
)
csv_user2, _ = user2.get(res=documents.DocumentDef.t, csv_user2, _ = user2.get(
item='actions/', res=documents.DocumentDef.t,
accept='text/csv', item='actions/',
query=[('filter', {'type': ['Computer']})]) accept='text/csv',
query=[('filter', {'type': ['Computer']})],
)
_, res = client.get(res=documents.DocumentDef.t, _, res = client.get(
item='actions/', res=documents.DocumentDef.t,
accept='text/csv', item='actions/',
query=[('filter', {'type': ['Computer']})], status=401) accept='text/csv',
query=[('filter', {'type': ['Computer']})],
status=401,
)
assert res.status_code == 401 assert res.status_code == 401
assert len(csv_user) > 0 assert len(csv_user) > 0
@ -167,21 +189,27 @@ def test_live_export_csv2(user: UserClient, client: Client, app: Devicehub):
acer = yaml2json('acer-happy.snapshot-test1') acer = yaml2json('acer-happy.snapshot-test1')
snapshot, _ = user.post(json_encode(acer), res=Snapshot) snapshot, _ = user.post(json_encode(acer), res=Snapshot)
device_id = snapshot['device']['id'] device_id = snapshot['device']['id']
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1, post_request = {
"devices": [device_id], "description": "aaa", "transaction": "ccc",
"finalUserCode": "abcdefjhi", "name": "John",
"startTime": "2020-11-01T02:00:00+00:00", "endUsers": 1,
"endTime": "2020-12-01T02:00:00+00:00" "devices": [device_id],
} "description": "aaa",
"finalUserCode": "abcdefjhi",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00",
}
user.post(res=Allocate, data=post_request) user.post(res=Allocate, data=post_request)
acer = yaml2json('acer-happy.live-test1') acer = yaml2json('acer-happy.live-test1')
live, _ = client.post(acer, res=Live) live, _ = client.post(acer, res=Live)
csv_user, _ = user.get(res=documents.DocumentDef.t, csv_user, _ = user.get(
item='actions/', res=documents.DocumentDef.t,
accept='text/csv', item='actions/',
query=[('filter', {'type': ['Computer'], 'ids': [device_id]})]) accept='text/csv',
query=[('filter', {'type': ['Computer'], 'ids': [device_id]})],
)
assert "4692" in csv_user assert "4692" in csv_user
assert "8692" in csv_user assert "8692" in csv_user
@ -195,11 +223,15 @@ def test_live_example2(user: UserClient, client: Client, app: Devicehub):
acer = yaml2json('acer-happy.snapshot-test1') acer = yaml2json('acer-happy.snapshot-test1')
snapshot, _ = user.post(json_encode(acer), res=Snapshot) snapshot, _ = user.post(json_encode(acer), res=Snapshot)
device_id = snapshot['device']['id'] device_id = snapshot['device']['id']
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1, post_request = {
"devices": [device_id], "description": "aaa", "transaction": "ccc",
"finalUserCode": "abcdefjhi", "name": "John",
"startTime": "2020-11-01T02:00:00+00:00", "endUsers": 1,
"endTime": "2020-12-01T02:00:00+00:00" "devices": [device_id],
"description": "aaa",
"finalUserCode": "abcdefjhi",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00",
} }
user.post(res=Allocate, data=post_request) user.post(res=Allocate, data=post_request)
@ -217,27 +249,36 @@ def test_export_basic_snapshot(user: UserClient):
"""Test export device information in a csv file.""" """Test export device information in a csv file."""
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot) snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)
dev_id = snapshot['device']['id'] dev_id = snapshot['device']['id']
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(
item='devices/', res=documents.DocumentDef.t,
accept='text/csv', item='devices/',
query=[('filter', {'type': ['Computer'], 'ids': [dev_id]})]) accept='text/csv',
query=[('filter', {'type': ['Computer'], 'ids': [dev_id]})],
)
f = StringIO(csv_str) f = StringIO(csv_str)
obj_csv = csv.reader(f, f, delimiter=';', quotechar='"') obj_csv = csv.reader(f, f, delimiter=';', quotechar='"')
export_csv = list(obj_csv) export_csv = list(obj_csv)
# Open fixture csv and transform to list # Open fixture csv and transform to list
with Path(__file__).parent.joinpath('files').joinpath('basic.csv').open() as csv_file: with Path(__file__).parent.joinpath('files').joinpath(
'basic.csv'
).open() as csv_file:
obj_csv = csv.reader(csv_file, delimiter=';', quotechar='"') obj_csv = csv.reader(csv_file, delimiter=';', quotechar='"')
fixture_csv = list(obj_csv) fixture_csv = list(obj_csv)
assert isinstance(datetime.strptime(export_csv[1][19], '%c'), datetime), \ assert isinstance(
'Register in field is not a datetime' datetime.strptime(export_csv[1][19], '%c'), datetime
), 'Register in field is not a datetime'
assert fixture_csv[0] == export_csv[0], 'Headers are not equal' assert fixture_csv[0] == export_csv[0], 'Headers are not equal'
assert fixture_csv[1][:19] == export_csv[1][:19], 'Computer information are not equal' assert (
fixture_csv[1][:19] == export_csv[1][:19]
), 'Computer information are not equal'
assert fixture_csv[1][20] == export_csv[1][20], 'Computer information are not equal' assert fixture_csv[1][20] == export_csv[1][20], 'Computer information are not equal'
assert fixture_csv[1][22:] == export_csv[1][22:], 'Computer information are not equal' assert (
fixture_csv[1][22:] == export_csv[1][22:]
), 'Computer information are not equal'
@pytest.mark.mvp @pytest.mark.mvp
@ -245,13 +286,17 @@ def test_export_basic_snapshot(user: UserClient):
def test_check_insert_hash(app: Devicehub, user: UserClient, client: Client): def test_check_insert_hash(app: Devicehub, user: UserClient, client: Client):
"""Test export device information in a csv file.""" """Test export device information in a csv file."""
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot) snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(
item='devices/', res=documents.DocumentDef.t,
accept='text/csv', item='devices/',
query=[('filter', {'type': ['Computer']})]) accept='text/csv',
query=[('filter', {'type': ['Computer']})],
)
hash3 = hashlib.sha3_256(csv_str.encode('utf-8')).hexdigest() hash3 = hashlib.sha3_256(csv_str.encode('utf-8')).hexdigest()
assert ReportHash.query.filter_by(hash3=hash3).count() == 1 assert ReportHash.query.filter_by(hash3=hash3).count() == 1
result, status = client.get(res=documents.DocumentDef.t, item='check/', query=[('hash', hash3)]) result, status = client.get(
res=documents.DocumentDef.t, item='check/', query=[('hash', hash3)]
)
assert status.status_code == 200 assert status.status_code == 200
assert result assert result
@ -266,7 +311,9 @@ def test_check_insert_hash(app: Devicehub, user: UserClient, client: Client):
@pytest.mark.mvp @pytest.mark.mvp
def test_export_extended(app: Devicehub, user: UserClient): def test_export_extended(app: Devicehub, user: UserClient):
"""Test a export device with all information and a lot of components.""" """Test a export device with all information and a lot of components."""
snapshot1, _ = user.post(file('real-eee-1001pxd.snapshot.12'), res=Snapshot, status=201) snapshot1, _ = user.post(
file('real-eee-1001pxd.snapshot.12'), res=Snapshot, status=201
)
snapshot2, _ = user.post(file('complete.export.snapshot'), res=Snapshot, status=201) snapshot2, _ = user.post(file('complete.export.snapshot'), res=Snapshot, status=201)
dev1_id = snapshot1['device']['id'] dev1_id = snapshot1['device']['id']
dev2_id = snapshot2['device']['id'] dev2_id = snapshot2['device']['id']
@ -276,10 +323,12 @@ def test_export_extended(app: Devicehub, user: UserClient):
db.session.add(pc) db.session.add(pc)
db.session.commit() db.session.commit()
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(
item='devices/', res=documents.DocumentDef.t,
accept='text/csv', item='devices/',
query=[('filter', {'type': ['Computer'], 'ids': [dev1_id, dev2_id]})]) accept='text/csv',
query=[('filter', {'type': ['Computer'], 'ids': [dev1_id, dev2_id]})],
)
f = StringIO(csv_str) f = StringIO(csv_str)
obj_csv = csv.reader(f, f, delimiter=';', quotechar='"') obj_csv = csv.reader(f, f, delimiter=';', quotechar='"')
@ -287,28 +336,50 @@ def test_export_extended(app: Devicehub, user: UserClient):
# Open fixture csv and transform to list # Open fixture csv and transform to list
with Path(__file__).parent.joinpath('files').joinpath( with Path(__file__).parent.joinpath('files').joinpath(
'proposal_extended_csv_report.csv').open() as csv_file: 'proposal_extended_csv_report.csv'
).open() as csv_file:
obj_csv = csv.reader(csv_file, delimiter=';', quotechar='"') obj_csv = csv.reader(csv_file, delimiter=';', quotechar='"')
fixture_csv = list(obj_csv) fixture_csv = list(obj_csv)
assert isinstance(datetime.strptime(export_csv[1][19], '%c'), datetime), \ assert isinstance(
'Register in field is not a datetime' datetime.strptime(export_csv[1][19], '%c'), datetime
), 'Register in field is not a datetime'
assert fixture_csv[0] == export_csv[0], 'Headers are not equal' assert fixture_csv[0] == export_csv[0], 'Headers are not equal'
assert fixture_csv[1][:19] == export_csv[1][:19], 'Computer information are not equal' assert (
fixture_csv[1][:19] == export_csv[1][:19]
), 'Computer information are not equal'
assert fixture_csv[1][20] == export_csv[1][20], 'Computer information are not equal' assert fixture_csv[1][20] == export_csv[1][20], 'Computer information are not equal'
assert fixture_csv[1][22:82] == export_csv[1][22:82], 'Computer information are not equal' assert (
assert fixture_csv[1][83] == export_csv[1][83], 'Computer information are not equal' fixture_csv[1][22:83] == export_csv[1][22:83]
assert fixture_csv[1][86:] == export_csv[1][86:], 'Computer information are not equal' ), 'Computer information are not equal'
assert fixture_csv[2][:19] == export_csv[2][:19], 'Computer information are not equal' assert fixture_csv[1][84] == export_csv[1][84], 'Computer information are not equal'
assert (
fixture_csv[1][87:] == export_csv[1][87:]
), 'Computer information are not equal'
assert (
fixture_csv[2][:19] == export_csv[2][:19]
), 'Computer information are not equal'
assert fixture_csv[2][20] == export_csv[2][20], 'Computer information are not equal' assert fixture_csv[2][20] == export_csv[2][20], 'Computer information are not equal'
assert fixture_csv[2][22:82] == export_csv[2][22:82], 'Computer information are not equal' assert (
assert fixture_csv[2][83] == export_csv[2][83], 'Computer information are not equal' fixture_csv[2][22:83] == export_csv[2][22:83]
assert fixture_csv[2][86:106] == export_csv[2][86:106], 'Computer information are not equal' ), 'Computer information are not equal'
assert fixture_csv[2][109] == export_csv[2][109], 'Computer information are not equal' assert fixture_csv[2][84] == export_csv[2][84], 'Computer information are not equal'
assert fixture_csv[2][112:133] == export_csv[2][112:133], 'Computer information are not equal' assert (
assert fixture_csv[2][135] == export_csv[2][135], 'Computer information are not equal' fixture_csv[2][87:107] == export_csv[2][87:107]
assert fixture_csv[2][138:] == export_csv[2][138:], 'Computer information are not equal' ), 'Computer information are not equal'
assert (
fixture_csv[2][110] == export_csv[2][110]
), 'Computer information are not equal'
assert (
fixture_csv[2][113:134] == export_csv[2][113:134]
), 'Computer information are not equal'
assert (
fixture_csv[2][136] == export_csv[2][136]
), 'Computer information are not equal'
assert (
fixture_csv[2][139:] == export_csv[2][139:]
), 'Computer information are not equal'
@pytest.mark.mvp @pytest.mark.mvp
@ -316,9 +387,9 @@ def test_export_empty(user: UserClient):
"""Test to check works correctly exporting csv without any information, """Test to check works correctly exporting csv without any information,
export a placeholder device. export a placeholder device.
""" """
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(
accept='text/csv', res=documents.DocumentDef.t, accept='text/csv', item='devices/'
item='devices/') )
f = StringIO(csv_str) f = StringIO(csv_str)
obj_csv = csv.reader(f, f) obj_csv = csv.reader(f, f)
export_csv = list(obj_csv) export_csv = list(obj_csv)
@ -330,16 +401,19 @@ def test_export_empty(user: UserClient):
def test_export_computer_monitor(user: UserClient): def test_export_computer_monitor(user: UserClient):
"""Test a export device type computer monitor.""" """Test a export device type computer monitor."""
snapshot, _ = user.post(file('computer-monitor.snapshot'), res=Snapshot) snapshot, _ = user.post(file('computer-monitor.snapshot'), res=Snapshot)
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(
item='devices/', res=documents.DocumentDef.t,
accept='text/csv', item='devices/',
query=[('filter', {'type': ['ComputerMonitor']})]) accept='text/csv',
query=[('filter', {'type': ['ComputerMonitor']})],
)
f = StringIO(csv_str) f = StringIO(csv_str)
obj_csv = csv.reader(f, f) obj_csv = csv.reader(f, f)
export_csv = list(obj_csv) export_csv = list(obj_csv)
# Open fixture csv and transform to list # Open fixture csv and transform to list
with Path(__file__).parent.joinpath('files').joinpath('computer-monitor.csv').open() \ with Path(__file__).parent.joinpath('files').joinpath(
as csv_file: 'computer-monitor.csv'
).open() as csv_file:
obj_csv = csv.reader(csv_file) obj_csv = csv.reader(csv_file)
fixture_csv = list(obj_csv) fixture_csv = list(obj_csv)
@ -355,15 +429,19 @@ def test_export_computer_monitor(user: UserClient):
def test_export_keyboard(user: UserClient): def test_export_keyboard(user: UserClient):
"""Test a export device type keyboard.""" """Test a export device type keyboard."""
snapshot, _ = user.post(file('keyboard.snapshot'), res=Snapshot) snapshot, _ = user.post(file('keyboard.snapshot'), res=Snapshot)
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(
item='devices/', res=documents.DocumentDef.t,
accept='text/csv', item='devices/',
query=[('filter', {'type': ['Keyboard']})]) accept='text/csv',
query=[('filter', {'type': ['Keyboard']})],
)
f = StringIO(csv_str) f = StringIO(csv_str)
obj_csv = csv.reader(f, f) obj_csv = csv.reader(f, f)
export_csv = list(obj_csv) export_csv = list(obj_csv)
# Open fixture csv and transform to list # Open fixture csv and transform to list
with Path(__file__).parent.joinpath('files').joinpath('keyboard.csv').open() as csv_file: with Path(__file__).parent.joinpath('files').joinpath(
'keyboard.csv'
).open() as csv_file:
obj_csv = csv.reader(csv_file) obj_csv = csv.reader(csv_file)
fixture_csv = list(obj_csv) fixture_csv = list(obj_csv)
@ -381,8 +459,9 @@ def test_export_multiple_different_devices(user: UserClient):
computers, keyboards, monitors, etc..) computers, keyboards, monitors, etc..)
""" """
# Open fixture csv and transform to list # Open fixture csv and transform to list
with Path(__file__).parent.joinpath('files').joinpath('multiples_devices.csv').open() \ with Path(__file__).parent.joinpath('files').joinpath(
as csv_file: 'multiples_devices.csv'
).open() as csv_file:
fixture_csv = list(csv.reader(csv_file)) fixture_csv = list(csv.reader(csv_file))
for row in fixture_csv: for row in fixture_csv:
del row[8] # We remove the 'Registered in' column del row[8] # We remove the 'Registered in' column
@ -393,10 +472,12 @@ def test_export_multiple_different_devices(user: UserClient):
snapshot_keyboard, _ = user.post(file('keyboard.snapshot'), res=Snapshot) snapshot_keyboard, _ = user.post(file('keyboard.snapshot'), res=Snapshot)
snapshot_monitor, _ = user.post(file('computer-monitor.snapshot'), res=Snapshot) snapshot_monitor, _ = user.post(file('computer-monitor.snapshot'), res=Snapshot)
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(
item='devices/', res=documents.DocumentDef.t,
query=[('filter', {'type': ['Computer', 'Keyboard', 'Monitor']})], item='devices/',
accept='text/csv') query=[('filter', {'type': ['Computer', 'Keyboard', 'Monitor']})],
accept='text/csv',
)
f = StringIO(csv_str) f = StringIO(csv_str)
obj_csv = csv.reader(f, f) obj_csv = csv.reader(f, f)
export_csv = list(obj_csv) export_csv = list(obj_csv)
@ -409,20 +490,25 @@ def test_export_multiple_different_devices(user: UserClient):
@pytest.mark.mvp @pytest.mark.mvp
def test_report_devices_stock_control(user: UserClient, user2: UserClient): def test_report_devices_stock_control(user: UserClient, user2: UserClient):
# TODO
"""Test export device information in a csv file.""" """Test export device information in a csv file."""
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot) snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)
snapshot2, _ = user2.post(file('basic.snapshot2'), res=Snapshot) snapshot2, _ = user2.post(file('basic.snapshot2'), res=Snapshot)
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(
item='stock/', res=documents.DocumentDef.t,
accept='text/csv', item='stock/',
query=[('filter', {'type': ['Computer']})]) accept='text/csv',
query=[('filter', {'type': ['Computer']})],
)
f = StringIO(csv_str) f = StringIO(csv_str)
obj_csv = csv.reader(f, f) obj_csv = csv.reader(f, f)
export_csv = list(obj_csv) export_csv = list(obj_csv)
# Open fixture csv and transform to list # Open fixture csv and transform to list
with Path(__file__).parent.joinpath('files').joinpath('basic-stock.csv').open() as csv_file: with Path(__file__).parent.joinpath('files').joinpath(
'basic-stock.csv'
).open() as csv_file:
obj_csv = csv.reader(csv_file) obj_csv = csv.reader(csv_file)
fixture_csv = list(obj_csv) fixture_csv = list(obj_csv)
@ -431,8 +517,12 @@ def test_report_devices_stock_control(user: UserClient, user2: UserClient):
export_csv[0] = export_csv[0][0].split(';') export_csv[0] = export_csv[0][0].split(';')
export_csv[1] = export_csv[1][0].split(';') export_csv[1] = export_csv[1][0].split(';')
assert isinstance(datetime.strptime(export_csv[1][5], '%c'), datetime), \ fixture_csv[0] = fixture_csv[0][0].split(';')
'Register in field is not a datetime' fixture_csv[1] = fixture_csv[1][0].split(';')
assert isinstance(
datetime.strptime(export_csv[1][5], '%c'), datetime
), 'Register in field is not a datetime'
# Pop dates fields from csv lists to compare them # Pop dates fields from csv lists to compare them
fixture_csv[1] = fixture_csv[1][:5] + fixture_csv[1][6:] fixture_csv[1] = fixture_csv[1][:5] + fixture_csv[1][6:]
@ -447,17 +537,22 @@ def test_report_devices_stock_control(user: UserClient, user2: UserClient):
def test_get_document_lots(user: UserClient, user2: UserClient): def test_get_document_lots(user: UserClient, user2: UserClient):
"""Tests submitting and retreiving all lots.""" """Tests submitting and retreiving all lots."""
l, _ = user.post({'name': 'Lot1', 'description': 'comments,lot1,testcomment-lot1,'}, res=Lot) l, _ = user.post(
l, _ = user.post({'name': 'Lot2', 'description': 'comments,lot2,testcomment-lot2,'}, res=Lot) {'name': 'Lot1', 'description': 'comments,lot1,testcomment-lot1,'}, res=Lot
l, _ = user2.post({'name': 'Lot3-User2', 'description': 'comments,lot3,testcomment-lot3,'}, res=Lot) )
l, _ = user.post(
{'name': 'Lot2', 'description': 'comments,lot2,testcomment-lot2,'}, res=Lot
)
l, _ = user2.post(
{'name': 'Lot3-User2', 'description': 'comments,lot3,testcomment-lot3,'},
res=Lot,
)
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(res=documents.DocumentDef.t, item='lots/', accept='text/csv')
item='lots/',
accept='text/csv')
csv2_str, _ = user2.get(res=documents.DocumentDef.t, csv2_str, _ = user2.get(
item='lots/', res=documents.DocumentDef.t, item='lots/', accept='text/csv'
accept='text/csv') )
f = StringIO(csv_str) f = StringIO(csv_str)
obj_csv = csv.reader(f, f) obj_csv = csv.reader(f, f)
@ -470,10 +565,17 @@ def test_get_document_lots(user: UserClient, user2: UserClient):
assert len(export_csv) == 3 assert len(export_csv) == 3
assert len(export2_csv) == 2 assert len(export2_csv) == 2
assert export_csv[0] == export2_csv[0] == ['Id', 'Name', 'Registered in', 'Description'] assert (
export_csv[0]
== export2_csv[0]
== ['Id', 'Name', 'Registered in', 'Description']
)
assert export_csv[1][1] == 'Lot1' or 'Lot2' assert export_csv[1][1] == 'Lot1' or 'Lot2'
assert export_csv[1][3] == 'comments,lot1,testcomment-lot1,' or 'comments,lot2,testcomment-lot2,' assert (
export_csv[1][3] == 'comments,lot1,testcomment-lot1,'
or 'comments,lot2,testcomment-lot2,'
)
assert export2_csv[1][1] == 'Lot3-User2' assert export2_csv[1][1] == 'Lot3-User2'
assert export2_csv[1][3] == 'comments,lot3,testcomment-lot3,' assert export2_csv[1][3] == 'comments,lot3,testcomment-lot3,'
@ -482,34 +584,39 @@ def test_get_document_lots(user: UserClient, user2: UserClient):
def test_verify_stamp(user: UserClient, client: Client): def test_verify_stamp(user: UserClient, client: Client):
"""Test verify stamp of one export device information in a csv file.""" """Test verify stamp of one export device information in a csv file."""
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot) snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(
item='devices/', res=documents.DocumentDef.t,
accept='text/csv', item='devices/',
query=[('filter', {'type': ['Computer']})]) accept='text/csv',
query=[('filter', {'type': ['Computer']})],
)
response, _ = client.post(res=documents.DocumentDef.t, response, _ = client.post(
item='stamps/', res=documents.DocumentDef.t,
content_type='multipart/form-data', item='stamps/',
accept='text/html', content_type='multipart/form-data',
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), 'example.csv')]}, accept='text/html',
status=200) data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), 'example.csv')]},
status=200,
)
assert "alert alert-info" in response assert "alert alert-info" in response
assert not "alert alert-danger" in response assert not "alert alert-danger" in response
response, _ = client.post(res=documents.DocumentDef.t, response, _ = client.post(
item='stamps/', res=documents.DocumentDef.t,
content_type='multipart/form-data', item='stamps/',
accept='text/html', content_type='multipart/form-data',
data={'docUpload': [(BytesIO(b'abc'), 'example.csv')]}, accept='text/html',
status=200) data={'docUpload': [(BytesIO(b'abc'), 'example.csv')]},
status=200,
)
assert not "alert alert-info" in response assert not "alert alert-info" in response
assert "alert alert-danger" in response assert "alert alert-danger" in response
response, _ = client.get(res=documents.DocumentDef.t, response, _ = client.get(
item='stamps/', res=documents.DocumentDef.t, item='stamps/', accept='text/html', status=200
accept='text/html', )
status=200)
assert not "alert alert-info" in response assert not "alert alert-info" in response
assert not "alert alert-danger" in response assert not "alert alert-danger" in response
@ -519,20 +626,23 @@ def test_verify_stamp(user: UserClient, client: Client):
def test_verify_stamp_log_info(user: UserClient, client: Client): def test_verify_stamp_log_info(user: UserClient, client: Client):
"""Test verify stamp of one export lots-info in a csv file.""" """Test verify stamp of one export lots-info in a csv file."""
l, _ = user.post({'name': 'Lot1', 'description': 'comments,lot1,testcomment-lot1,'}, res=Lot) l, _ = user.post(
l, _ = user.post({'name': 'Lot2', 'description': 'comments,lot2,testcomment-lot2,'}, res=Lot) {'name': 'Lot1', 'description': 'comments,lot1,testcomment-lot1,'}, res=Lot
)
l, _ = user.post(
{'name': 'Lot2', 'description': 'comments,lot2,testcomment-lot2,'}, res=Lot
)
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(res=documents.DocumentDef.t, item='lots/', accept='text/csv')
item='lots/',
accept='text/csv')
response, _ = client.post(res=documents.DocumentDef.t, response, _ = client.post(
item='stamps/', res=documents.DocumentDef.t,
content_type='multipart/form-data', item='stamps/',
accept='text/html', content_type='multipart/form-data',
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), accept='text/html',
'example.csv')]}, data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), 'example.csv')]},
status=200) status=200,
)
assert "alert alert-info" in response assert "alert alert-info" in response
@ -542,18 +652,21 @@ def test_verify_stamp_devices_stock(user: UserClient, client: Client):
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot) snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(
item='stock/', res=documents.DocumentDef.t,
accept='text/csv', item='stock/',
query=[('filter', {'type': ['Computer']})]) accept='text/csv',
query=[('filter', {'type': ['Computer']})],
)
response, _ = client.post(res=documents.DocumentDef.t, response, _ = client.post(
item='stamps/', res=documents.DocumentDef.t,
content_type='multipart/form-data', item='stamps/',
accept='text/html', content_type='multipart/form-data',
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), accept='text/html',
'example.csv')]}, data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), 'example.csv')]},
status=200) status=200,
)
assert "alert alert-info" in response assert "alert alert-info" in response
@ -563,11 +676,15 @@ def test_verify_stamp_csv_actions(user: UserClient, client: Client):
acer = yaml2json('acer.happy.battery.snapshot') acer = yaml2json('acer.happy.battery.snapshot')
snapshot, _ = user.post(json_encode(acer), res=Snapshot) snapshot, _ = user.post(json_encode(acer), res=Snapshot)
device_id = snapshot['device']['id'] device_id = snapshot['device']['id']
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1, post_request = {
"devices": [device_id], "description": "aaa", "transaction": "ccc",
"finalUserCode": "abcdefjhi", "name": "John",
"startTime": "2020-11-01T02:00:00+00:00", "endUsers": 1,
"endTime": "2020-12-01T02:00:00+00:00" "devices": [device_id],
"description": "aaa",
"finalUserCode": "abcdefjhi",
"startTime": "2020-11-01T02:00:00+00:00",
"endTime": "2020-12-01T02:00:00+00:00",
} }
user.post(res=Allocate, data=post_request) user.post(res=Allocate, data=post_request)
@ -578,18 +695,21 @@ def test_verify_stamp_csv_actions(user: UserClient, client: Client):
acer['licence_version'] = '1.0.0' acer['licence_version'] = '1.0.0'
snapshot, _ = client.post(acer, res=Live) snapshot, _ = client.post(acer, res=Live)
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(
item='actions/', res=documents.DocumentDef.t,
accept='text/csv', item='actions/',
query=[('filter', {'type': ['Computer']})]) accept='text/csv',
query=[('filter', {'type': ['Computer']})],
)
response, _ = client.post(res=documents.DocumentDef.t, response, _ = client.post(
item='stamps/', res=documents.DocumentDef.t,
content_type='multipart/form-data', item='stamps/',
accept='text/html', content_type='multipart/form-data',
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), accept='text/html',
'example.csv')]}, data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), 'example.csv')]},
status=200) status=200,
)
assert "alert alert-info" in response assert "alert alert-info" in response
@ -599,35 +719,38 @@ def test_verify_stamp_erasure_certificate(user: UserClient, client: Client):
s = file('erase-sectors.snapshot') s = file('erase-sectors.snapshot')
snapshot, response = user.post(s, res=Snapshot) snapshot, response = user.post(s, res=Snapshot)
doc, _ = user.get(res=documents.DocumentDef.t, doc, _ = user.get(
item='erasures/', res=documents.DocumentDef.t,
query=[('filter', {'ids': [snapshot['device']['id']]})], item='erasures/',
accept=ANY) query=[('filter', {'ids': [snapshot['device']['id']]})],
accept=ANY,
)
response, _ = client.post(res=documents.DocumentDef.t, response, _ = client.post(
item='stamps/', res=documents.DocumentDef.t,
content_type='multipart/form-data', item='stamps/',
accept='text/html', content_type='multipart/form-data',
data={'docUpload': [(BytesIO(bytes(doc, 'utf-8')), accept='text/html',
'example.csv')]}, data={'docUpload': [(BytesIO(bytes(doc, 'utf-8')), 'example.csv')]},
status=200) status=200,
)
assert "alert alert-danger" in response assert "alert alert-danger" in response
doc, _ = user.get(res=documents.DocumentDef.t, doc, _ = user.get(
item='erasures/', res=documents.DocumentDef.t,
query=[ item='erasures/',
('filter', {'ids': [snapshot['device']['id']]}), query=[('filter', {'ids': [snapshot['device']['id']]}), ('format', 'PDF')],
('format', 'PDF') accept='application/pdf',
], )
accept='application/pdf')
response, _ = client.post(res=documents.DocumentDef.t, response, _ = client.post(
item='stamps/', res=documents.DocumentDef.t,
content_type='multipart/form-data', item='stamps/',
accept='text/html', content_type='multipart/form-data',
data={'docUpload': [(BytesIO(doc), accept='text/html',
'example.csv')]}, data={'docUpload': [(BytesIO(doc), 'example.csv')]},
status=200) status=200,
)
assert "alert alert-info" in response assert "alert alert-info" in response
@ -636,11 +759,10 @@ def test_get_document_internal_stats(user: UserClient, user2: UserClient):
"""Tests for get the internal stats.""" """Tests for get the internal stats."""
# csv_str, _ = user.get(res=documents.DocumentDef.t, # csv_str, _ = user.get(res=documents.DocumentDef.t,
# item='internalstats/') # item='internalstats/')
csv_str, _ = user.get(res=documents.DocumentDef.t, csv_str, _ = user.get(
item='internalstats/', res=documents.DocumentDef.t, item='internalstats/', accept='text/csv', query=[]
accept='text/csv', )
query=[])
f = StringIO(csv_str) f = StringIO(csv_str)
obj_csv = csv.reader(f, f) obj_csv = csv.reader(f, f)
@ -648,10 +770,9 @@ def test_get_document_internal_stats(user: UserClient, user2: UserClient):
assert len(export_csv) == 1 assert len(export_csv) == 1
csv_str, _ = user2.get(res=documents.DocumentDef.t, csv_str, _ = user2.get(
item='internalstats/', res=documents.DocumentDef.t, item='internalstats/', accept='text/csv', query=[]
accept='text/csv', )
query=[])
f = StringIO(csv_str) f = StringIO(csv_str)
obj_csv = csv.reader(f, f) obj_csv = csv.reader(f, f)
@ -672,8 +793,9 @@ def test_get_wbconf(user: UserClient):
assert 'WB_ERASE =' in env assert 'WB_ERASE =' in env
# assert 'WB_ERASE = True' in env # assert 'WB_ERASE = True' in env
session = Session.query.filter_by(user_id=user.user['id'], session = Session.query.filter_by(
type=SessionType.Internal).first() user_id=user.user['id'], type=SessionType.Internal
).first()
token = session.token token = session.token
token = auth.Auth.encode(session.token) token = auth.Auth.encode(session.token)
assert token in env assert token in env
@ -691,7 +813,7 @@ def test_trade_documents(user: UserClient):
'filename': 'test.pdf', 'filename': 'test.pdf',
'hash': 'bbbbbbbb', 'hash': 'bbbbbbbb',
'url': 'http://www.ereuse.org/', 'url': 'http://www.ereuse.org/',
'lot': lot['id'] 'lot': lot['id'],
} }
doc, _ = user.post(res=TradeDocument, data=request_post) doc, _ = user.post(res=TradeDocument, data=request_post)
assert doc['filename'] == request_post['filename'] assert doc['filename'] == request_post['filename']
@ -707,13 +829,15 @@ def test_trade_documents_with_weight(user: UserClient):
lot, _ = user.post({'name': 'MyLot'}, res=Lot) lot, _ = user.post({'name': 'MyLot'}, res=Lot)
# long url # long url
url = 'http://www.ereuse.org/apapaapaapaapaapaapaapaapaapaapapaapaapaapaapaapaapaapaapaapapaapaapaapaapaapaapaapaapaapaaaa', url = (
'http://www.ereuse.org/apapaapaapaapaapaapaapaapaapaapapaapaapaapaapaapaapaapaapaapapaapaapaapaapaapaapaapaapaapaaaa',
)
request_post = { request_post = {
'filename': 'test.pdf', 'filename': 'test.pdf',
'hash': 'bbbbbbbb', 'hash': 'bbbbbbbb',
'url': url, 'url': url,
'weight': 15, 'weight': 15,
'lot': lot['id'] 'lot': lot['id'],
} }
doc, _ = user.post(res=TradeDocument, data=request_post) doc, _ = user.post(res=TradeDocument, data=request_post)
assert doc['weight'] == request_post['weight'] assert doc['weight'] == request_post['weight']

View File

@ -224,11 +224,12 @@ def test_export_devices(user3: UserClientFlask):
), 'Computer information are not equal' ), 'Computer information are not equal'
assert fixture_csv[1][20] == export_csv[1][20], 'Computer information are not equal' assert fixture_csv[1][20] == export_csv[1][20], 'Computer information are not equal'
assert ( assert (
fixture_csv[1][22:82] == export_csv[1][22:82] fixture_csv[1][22:83] == export_csv[1][22:83]
), 'Computer information are not equal' ), 'Computer information are not equal'
assert fixture_csv[1][83] == export_csv[1][83], 'Computer information are not equal'
assert fixture_csv[1][84] == export_csv[1][84], 'Computer information are not equal'
assert ( assert (
fixture_csv[1][86:] == export_csv[1][86:] fixture_csv[1][88:] == export_csv[1][88:]
), 'Computer information are not equal' ), 'Computer information are not equal'