Merge branch 'testing' into feature/#126-named-system-tags
This commit is contained in:
commit
f9281fbdc0
|
@ -13,6 +13,7 @@ ml).
|
|||
|
||||
## [1.0.5-beta]
|
||||
- [addend] #124 adding endpoint for extract the internal stats of use
|
||||
- [addend] #122 system for verify all documents that it's produced from devicehub
|
||||
|
||||
## [1.0.4-beta]
|
||||
- [addend] #95 adding endpoint for check the hash of one report
|
||||
|
|
|
@ -1 +1 @@
|
|||
__version__ = "1.0.4-beta"
|
||||
__version__ = "1.0.5-beta"
|
||||
|
|
|
@ -27,7 +27,7 @@ from ereuse_devicehub.resources.documents.device_row import (DeviceRow, StockRow
|
|||
InternalStatsRow)
|
||||
from ereuse_devicehub.resources.lot import LotView
|
||||
from ereuse_devicehub.resources.lot.models import Lot
|
||||
from ereuse_devicehub.resources.hash_reports import insert_hash, ReportHash
|
||||
from ereuse_devicehub.resources.hash_reports import insert_hash, ReportHash, verify_hash
|
||||
|
||||
|
||||
class Format(enum.Enum):
|
||||
|
@ -80,6 +80,7 @@ class DocumentView(DeviceView):
|
|||
res = flask_weasyprint.render_pdf(
|
||||
flask_weasyprint.HTML(string=template), download_filename='{}.pdf'.format(type)
|
||||
)
|
||||
insert_hash(res.data)
|
||||
else:
|
||||
res = flask.make_response(template)
|
||||
return res
|
||||
|
@ -186,7 +187,9 @@ class LotsDocumentView(LotView):
|
|||
cw.writerow(l.keys())
|
||||
first = False
|
||||
cw.writerow(l.values())
|
||||
output = make_response(data.getvalue())
|
||||
bfile = data.getvalue().encode('utf-8')
|
||||
output = make_response(bfile)
|
||||
insert_hash(bfile)
|
||||
output.headers['Content-Disposition'] = 'attachment; filename=lots-info.csv'
|
||||
output.headers['Content-type'] = 'text/csv'
|
||||
return output
|
||||
|
@ -223,7 +226,9 @@ class StockDocumentView(DeviceView):
|
|||
cw.writerow(d.keys())
|
||||
first = False
|
||||
cw.writerow(d.values())
|
||||
output = make_response(data.getvalue())
|
||||
bfile = data.getvalue().encode('utf-8')
|
||||
output = make_response(bfile)
|
||||
insert_hash(bfile)
|
||||
output.headers['Content-Disposition'] = 'attachment; filename=devices-stock.csv'
|
||||
output.headers['Content-type'] = 'text/csv'
|
||||
return output
|
||||
|
@ -247,12 +252,32 @@ class StampsView(View):
|
|||
This view render one public ans static page for see the links for to do the check
|
||||
of one csv file
|
||||
"""
|
||||
def get(self):
|
||||
def get_url_path(self):
|
||||
url = urlutils.URL(request.url)
|
||||
url.normalize()
|
||||
url.path_parts = url.path_parts[:-2] + ['check', '']
|
||||
url_path = url.to_text()
|
||||
return flask.render_template('documents/stamp.html', rq_url=url_path)
|
||||
return url.to_text()
|
||||
|
||||
def get(self):
|
||||
result = ('', '')
|
||||
return flask.render_template('documents/stamp.html', rq_url=self.get_url_path(),
|
||||
result=result)
|
||||
|
||||
def post(self):
|
||||
result = ('', '')
|
||||
if 'docUpload' in request.files:
|
||||
file_check = request.files['docUpload']
|
||||
bad = 'There are no coincidences. The attached file data does not come \
|
||||
from our backend or it has been subsequently modified.'
|
||||
ok = '100% coincidence. The attached file contains data 100% existing in \
|
||||
to our backend'
|
||||
result = ('Bad', bad)
|
||||
if file_check.mimetype in ['text/csv', 'application/pdf']:
|
||||
if verify_hash(file_check):
|
||||
result = ('Ok', ok)
|
||||
|
||||
return flask.render_template('documents/stamp.html', rq_url=self.get_url_path(),
|
||||
result=result)
|
||||
|
||||
|
||||
class InternalStatsView(DeviceView):
|
||||
|
@ -345,7 +370,7 @@ class DocumentDef(Resource):
|
|||
self.add_url_rule('/check/', defaults={}, view_func=check_view, methods=get)
|
||||
|
||||
stamps_view = StampsView.as_view('StampsView', definition=self, auth=app.auth)
|
||||
self.add_url_rule('/stamps/', defaults={}, view_func=stamps_view, methods=get)
|
||||
self.add_url_rule('/stamps/', defaults={}, view_func=stamps_view, methods={'GET', 'POST'})
|
||||
|
||||
internalstats_view = InternalStatsView.as_view(
|
||||
'InternalStatsView', definition=self, auth=app.auth)
|
||||
|
|
|
@ -38,6 +38,17 @@
|
|||
</nav>
|
||||
<div class="container-fluid">
|
||||
<div class="page-header col-md-6 col-md-offset-3">
|
||||
<div class="row">
|
||||
{% if result.0 == 'Ok' %}
|
||||
<div class="alert alert-info" style="background-color: #3fb618;" role="alert">
|
||||
{{ result.1 }}
|
||||
</div>
|
||||
{% elif result.0 == 'Bad' %}
|
||||
<div class="alert alert-danger" style="background-color: #ff0039" role="alert">
|
||||
{{ result.1 }}
|
||||
</div>
|
||||
{% endif %}
|
||||
</div>
|
||||
<div class="row">
|
||||
<a href="http://dlt.ereuse.org/stamps/create?url={{ rq_url }}" target="_blank">Add one new check in your csv</a>
|
||||
</div>
|
||||
|
@ -45,6 +56,19 @@
|
|||
<a href="http://dlt.ereuse.org/stamps/check?url={{ rq_url }}" target="_blank">Verify a CSV file in here.</a>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="row">
|
||||
<div class="page-header col-md-6 col-md-offset-3">
|
||||
If you want us to verify a document issued by us, upload it using the following form
|
||||
</div>
|
||||
<div class="col-md-6 col-md-offset-3">
|
||||
<form enctype="multipart/form-data" action="" method="post">
|
||||
<label for="name">Document: </label>
|
||||
<input type="file" name="docUpload" accept="*" /><br />
|
||||
<input type="submit" id="send-signup" name="signup" value="Verify" />
|
||||
</form>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
|
|
|
@ -32,3 +32,8 @@ def insert_hash(bfile):
|
|||
db.session.add(db_hash)
|
||||
db.session.commit()
|
||||
db.session.flush()
|
||||
|
||||
|
||||
def verify_hash(bfile):
|
||||
hash3 = hashlib.sha3_256(bfile.read()).hexdigest()
|
||||
return ReportHash.query.filter(ReportHash.hash3 == hash3).count()
|
||||
|
|
|
@ -1,8 +1,9 @@
|
|||
import csv
|
||||
import hashlib
|
||||
from datetime import datetime
|
||||
from io import StringIO
|
||||
from io import StringIO, BytesIO
|
||||
from pathlib import Path
|
||||
from flask import url_for
|
||||
|
||||
import pytest
|
||||
from werkzeug.exceptions import Unauthorized
|
||||
|
@ -463,6 +464,160 @@ def test_get_document_lots(user: UserClient, user2: UserClient):
|
|||
assert export2_csv[1][3] == 'comments,lot3,testcomment-lot3,'
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
def test_verify_stamp(user: UserClient, client: Client):
|
||||
"""Test verify stamp of one export device information in a csv file."""
|
||||
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)
|
||||
csv_str, _ = user.get(res=documents.DocumentDef.t,
|
||||
item='devices/',
|
||||
accept='text/csv',
|
||||
query=[('filter', {'type': ['Computer']})])
|
||||
|
||||
response, _ = client.post(res=documents.DocumentDef.t,
|
||||
item='stamps/',
|
||||
content_type='multipart/form-data',
|
||||
accept='text/html',
|
||||
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')), 'example.csv')]},
|
||||
status=200)
|
||||
assert "alert alert-info" in response
|
||||
assert not "alert alert-danger" in response
|
||||
|
||||
response, _ = client.post(res=documents.DocumentDef.t,
|
||||
item='stamps/',
|
||||
content_type='multipart/form-data',
|
||||
accept='text/html',
|
||||
data={'docUpload': [(BytesIO(b'abc'), 'example.csv')]},
|
||||
status=200)
|
||||
|
||||
assert not "alert alert-info" in response
|
||||
assert "alert alert-danger" in response
|
||||
|
||||
response, _ = client.get(res=documents.DocumentDef.t,
|
||||
item='stamps/',
|
||||
accept='text/html',
|
||||
status=200)
|
||||
|
||||
assert not "alert alert-info" in response
|
||||
assert not "alert alert-danger" in response
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
def test_verify_stamp_log_info(user: UserClient, client: Client):
|
||||
"""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({'name': 'Lot2', 'description': 'comments,lot2,testcomment-lot2,'}, res=Lot)
|
||||
|
||||
csv_str, _ = user.get(res=documents.DocumentDef.t,
|
||||
item='lots/',
|
||||
accept='text/csv')
|
||||
|
||||
response, _ = client.post(res=documents.DocumentDef.t,
|
||||
item='stamps/',
|
||||
content_type='multipart/form-data',
|
||||
accept='text/html',
|
||||
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')),
|
||||
'example.csv')]},
|
||||
status=200)
|
||||
assert "alert alert-info" in response
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
def test_verify_stamp_devices_stock(user: UserClient, client: Client):
|
||||
"""Test verify stamp of one export device information in a csv file."""
|
||||
|
||||
snapshot, _ = user.post(file('basic.snapshot'), res=Snapshot)
|
||||
|
||||
csv_str, _ = user.get(res=documents.DocumentDef.t,
|
||||
item='stock/',
|
||||
accept='text/csv',
|
||||
query=[('filter', {'type': ['Computer']})])
|
||||
|
||||
response, _ = client.post(res=documents.DocumentDef.t,
|
||||
item='stamps/',
|
||||
content_type='multipart/form-data',
|
||||
accept='text/html',
|
||||
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')),
|
||||
'example.csv')]},
|
||||
status=200)
|
||||
assert "alert alert-info" in response
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
def test_verify_stamp_csv_actions(user: UserClient, client: Client):
|
||||
"""Test verify stamp of one export device information in a csv file with others users."""
|
||||
acer = file('acer.happy.battery.snapshot')
|
||||
snapshot, _ = user.post(acer, res=Snapshot)
|
||||
device_id = snapshot['device']['id']
|
||||
post_request = {"transaction": "ccc", "name": "John", "endUsers": 1,
|
||||
"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)
|
||||
hdd = [c for c in acer['components'] if c['type'] == 'HardDrive'][0]
|
||||
hdd_action = [a for a in hdd['actions'] if a['type'] == 'TestDataStorage'][0]
|
||||
hdd_action['lifetime'] += 1000
|
||||
acer.pop('elapsed')
|
||||
acer['licence_version'] = '1.0.0'
|
||||
snapshot, _ = client.post(acer, res=Live)
|
||||
|
||||
csv_str, _ = user.get(res=documents.DocumentDef.t,
|
||||
item='actions/',
|
||||
accept='text/csv',
|
||||
query=[('filter', {'type': ['Computer']})])
|
||||
|
||||
response, _ = client.post(res=documents.DocumentDef.t,
|
||||
item='stamps/',
|
||||
content_type='multipart/form-data',
|
||||
accept='text/html',
|
||||
data={'docUpload': [(BytesIO(bytes(csv_str, 'utf-8')),
|
||||
'example.csv')]},
|
||||
status=200)
|
||||
assert "alert alert-info" in response
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
def test_verify_stamp_erasure_certificate(user: UserClient, client: Client):
|
||||
"""Test verify stamp of one export certificate in PDF."""
|
||||
s = file('erase-sectors.snapshot')
|
||||
snapshot, response = user.post(s, res=Snapshot)
|
||||
# import pdb; pdb.set_trace()
|
||||
|
||||
doc, _ = user.get(res=documents.DocumentDef.t,
|
||||
item='erasures/',
|
||||
query=[('filter', {'id': [snapshot['device']['id']]})],
|
||||
accept=ANY)
|
||||
|
||||
response, _ = client.post(res=documents.DocumentDef.t,
|
||||
item='stamps/',
|
||||
content_type='multipart/form-data',
|
||||
accept='text/html',
|
||||
data={'docUpload': [(BytesIO(bytes(doc, 'utf-8')),
|
||||
'example.csv')]},
|
||||
status=200)
|
||||
assert "alert alert-danger" in response
|
||||
|
||||
doc, _ = user.get(res=documents.DocumentDef.t,
|
||||
item='erasures/',
|
||||
query=[
|
||||
('filter', {'id': [snapshot['device']['id']]}),
|
||||
('format', 'PDF')
|
||||
],
|
||||
accept='application/pdf')
|
||||
|
||||
response, _ = client.post(res=documents.DocumentDef.t,
|
||||
item='stamps/',
|
||||
content_type='multipart/form-data',
|
||||
accept='text/html',
|
||||
data={'docUpload': [(BytesIO(doc),
|
||||
'example.csv')]},
|
||||
status=200)
|
||||
assert "alert alert-info" in response
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
def test_get_document_internal_stats(user: UserClient, user2: UserClient):
|
||||
"""Tests for get teh internal stats."""
|
||||
|
|
Reference in a new issue