Merge pull request #229 from eReuse/feature/server-side-render-testing

Feature/server side render testing
This commit is contained in:
cayop 2022-04-29 18:39:37 +02:00 committed by GitHub
commit 861515c46d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 1261 additions and 170 deletions

View file

@ -2,7 +2,10 @@ from inspect import isclass
from typing import Dict, Iterable, Type, Union from typing import Dict, Iterable, Type, Union
from ereuse_utils.test import JSON, Res from ereuse_utils.test import JSON, Res
from teal.client import Client as TealClient, Query, Status from flask.testing import FlaskClient
from flask_wtf.csrf import generate_csrf
from teal.client import Client as TealClient
from teal.client import Query, Status
from werkzeug.exceptions import HTTPException from werkzeug.exceptions import HTTPException
from ereuse_devicehub.resources import models, schemas from ereuse_devicehub.resources import models, schemas
@ -13,13 +16,19 @@ ResourceLike = Union[Type[Union[models.Thing, schemas.Thing]], str]
class Client(TealClient): class Client(TealClient):
"""A client suited for Devicehub main usage.""" """A client suited for Devicehub main usage."""
def __init__(self, application, def __init__(
self,
application,
response_wrapper=None, response_wrapper=None,
use_cookies=False, use_cookies=False,
allow_subdomain_redirects=False): allow_subdomain_redirects=False,
super().__init__(application, response_wrapper, use_cookies, allow_subdomain_redirects) ):
super().__init__(
application, response_wrapper, use_cookies, allow_subdomain_redirects
)
def open(self, def open(
self,
uri: str, uri: str,
res: ResourceLike = None, res: ResourceLike = None,
status: Status = 200, status: Status = 200,
@ -29,13 +38,16 @@ class Client(TealClient):
item=None, item=None,
headers: dict = None, headers: dict = None,
token: str = None, token: str = None,
**kw) -> Res: **kw,
) -> Res:
if isclass(res) and issubclass(res, (models.Thing, schemas.Thing)): if isclass(res) and issubclass(res, (models.Thing, schemas.Thing)):
res = res.t res = res.t
return super().open(uri, res, status, query, accept, content_type, item, headers, token, return super().open(
**kw) uri, res, status, query, accept, content_type, item, headers, token, **kw
)
def get(self, def get(
self,
uri: str = '', uri: str = '',
res: ResourceLike = None, res: ResourceLike = None,
query: Query = tuple(), query: Query = tuple(),
@ -44,10 +56,12 @@ class Client(TealClient):
accept: str = JSON, accept: str = JSON,
headers: dict = None, headers: dict = None,
token: str = None, token: str = None,
**kw) -> Res: **kw,
) -> Res:
return super().get(uri, res, query, status, item, accept, headers, token, **kw) return super().get(uri, res, query, status, item, accept, headers, token, **kw)
def post(self, def post(
self,
data: str or dict, data: str or dict,
uri: str = '', uri: str = '',
res: ResourceLike = None, res: ResourceLike = None,
@ -57,11 +71,14 @@ class Client(TealClient):
accept: str = JSON, accept: str = JSON,
headers: dict = None, headers: dict = None,
token: str = None, token: str = None,
**kw) -> Res: **kw,
return super().post(data, uri, res, query, status, content_type, accept, headers, token, ) -> Res:
**kw) return super().post(
data, uri, res, query, status, content_type, accept, headers, token, **kw
)
def patch(self, def patch(
self,
data: str or dict, data: str or dict,
uri: str = '', uri: str = '',
res: ResourceLike = None, res: ResourceLike = None,
@ -72,11 +89,24 @@ class Client(TealClient):
accept: str = JSON, accept: str = JSON,
headers: dict = None, headers: dict = None,
token: str = None, token: str = None,
**kw) -> Res: **kw,
return super().patch(data, uri, res, query, item, status, content_type, accept, token, ) -> Res:
headers, **kw) return super().patch(
data,
uri,
res,
query,
item,
status,
content_type,
accept,
token,
headers,
**kw,
)
def put(self, def put(
self,
data: str or dict, data: str or dict,
uri: str = '', uri: str = '',
res: ResourceLike = None, res: ResourceLike = None,
@ -87,11 +117,24 @@ class Client(TealClient):
accept: str = JSON, accept: str = JSON,
headers: dict = None, headers: dict = None,
token: str = None, token: str = None,
**kw) -> Res: **kw,
return super().put(data, uri, res, query, item, status, content_type, accept, token, ) -> Res:
headers, **kw) return super().put(
data,
uri,
res,
query,
item,
status,
content_type,
accept,
token,
headers,
**kw,
)
def delete(self, def delete(
self,
uri: str = '', uri: str = '',
res: ResourceLike = None, res: ResourceLike = None,
query: Query = tuple(), query: Query = tuple(),
@ -100,23 +143,29 @@ class Client(TealClient):
accept: str = JSON, accept: str = JSON,
headers: dict = None, headers: dict = None,
token: str = None, token: str = None,
**kw) -> Res: **kw,
return super().delete(uri, res, query, status, item, accept, headers, token, **kw) ) -> Res:
return super().delete(
uri, res, query, status, item, accept, headers, token, **kw
)
def login(self, email: str, password: str): def login(self, email: str, password: str):
assert isinstance(email, str) assert isinstance(email, str)
assert isinstance(password, str) assert isinstance(password, str)
return self.post({'email': email, 'password': password}, '/users/login/', status=200) return self.post(
{'email': email, 'password': password}, '/users/login/', status=200
)
def get_many(self, def get_many(
self,
res: ResourceLike, res: ResourceLike,
resources: Iterable[Union[dict, int]], resources: Iterable[Union[dict, int]],
key: str = None, key: str = None,
**kw) -> Iterable[Union[Dict[str, object], str]]: **kw,
) -> Iterable[Union[Dict[str, object], str]]:
"""Like :meth:`.get` but with many resources.""" """Like :meth:`.get` but with many resources."""
return ( return (
self.get(res=res, item=r[key] if key else r, **kw)[0] self.get(res=res, item=r[key] if key else r, **kw)[0] for r in resources
for r in resources
) )
@ -126,18 +175,24 @@ class UserClient(Client):
It will automatically perform login on the first request. It will automatically perform login on the first request.
""" """
def __init__(self, application, def __init__(
self,
application,
email: str, email: str,
password: str, password: str,
response_wrapper=None, response_wrapper=None,
use_cookies=False, use_cookies=False,
allow_subdomain_redirects=False): allow_subdomain_redirects=False,
super().__init__(application, response_wrapper, use_cookies, allow_subdomain_redirects) ):
super().__init__(
application, response_wrapper, use_cookies, allow_subdomain_redirects
)
self.email = email # type: str self.email = email # type: str
self.password = password # type: str self.password = password # type: str
self.user = None # type: dict self.user = None # type: dict
def open(self, def open(
self,
uri: str, uri: str,
res: ResourceLike = None, res: ResourceLike = None,
status: int or HTTPException = 200, status: int or HTTPException = 200,
@ -147,12 +202,92 @@ class UserClient(Client):
item=None, item=None,
headers: dict = None, headers: dict = None,
token: str = None, token: str = None,
**kw) -> Res: **kw,
return super().open(uri, res, status, query, accept, content_type, item, headers, ) -> Res:
self.user['token'] if self.user else token, **kw) return super().open(
uri,
res,
status,
query,
accept,
content_type,
item,
headers,
self.user['token'] if self.user else token,
**kw,
)
# noinspection PyMethodOverriding # noinspection PyMethodOverriding
def login(self): def login(self):
response = super().login(self.email, self.password) response = super().login(self.email, self.password)
self.user = response[0] self.user = response[0]
return response return response
class UserClientFlask:
def __init__(
self,
application,
email: str,
password: str,
response_wrapper=None,
use_cookies=True,
follow_redirects=True,
):
self.email = email
self.password = password
self.follow_redirects = follow_redirects
self.user = None
self.client = FlaskClient(application, use_cookies=use_cookies)
self.client.get('/login/')
data = {
'email': email,
'password': password,
'csrf_token': generate_csrf(),
}
body, status, headers = self.client.post(
'/login/', data=data, follow_redirects=True
)
self.headers = headers
body = next(body).decode("utf-8")
assert "Unassgined" in body
def get(
self,
uri='',
data=None,
follow_redirects=True,
content_type='text/html; charset=utf-8',
decode=True,
**kw,
):
body, status, headers = self.client.get(
uri, data=data, follow_redirects=follow_redirects, headers=self.headers
)
if decode:
body = next(body).decode("utf-8")
return (body, status)
def post(
self,
uri='',
data=None,
follow_redirects=True,
content_type='application/x-www-form-urlencoded',
decode=True,
**kw,
):
body, status, headers = self.client.post(
uri,
data=data,
follow_redirects=follow_redirects,
headers=self.headers,
content_type=content_type,
)
if decode:
body = next(body).decode("utf-8")
return (body, status)

View file

@ -497,7 +497,7 @@ class TagDeviceForm(FlaskForm):
db.session.commit() db.session.commit()
class NewActionForm(FlaskForm): class ActionFormMix(FlaskForm):
name = StringField( name = StringField(
'Name', 'Name',
[validators.length(max=50)], [validators.length(max=50)],
@ -529,8 +529,14 @@ class NewActionForm(FlaskForm):
if not is_valid: if not is_valid:
return False return False
if self.type.data in [None, '']:
return False
if not self.devices.data:
return False
self._devices = OrderedSet() self._devices = OrderedSet()
if self.devices.data:
devices = set(self.devices.data.split(",")) devices = set(self.devices.data.split(","))
self._devices = OrderedSet( self._devices = OrderedSet(
Device.query.filter(Device.id.in_(devices)) Device.query.filter(Device.id.in_(devices))
@ -572,7 +578,20 @@ class NewActionForm(FlaskForm):
return self.type.data return self.type.data
class AllocateForm(NewActionForm): class NewActionForm(ActionFormMix):
def validate(self, extra_validators=None):
is_valid = super().validate(extra_validators)
if not is_valid:
return False
if self.type.data in ['Allocate', 'Deallocate', 'Trade', 'DataWipe']:
return False
return True
class AllocateForm(ActionFormMix):
start_time = DateField('Start time') start_time = DateField('Start time')
end_time = DateField('End time') end_time = DateField('End time')
final_user_code = StringField('Final user code', [validators.length(max=50)]) final_user_code = StringField('Final user code', [validators.length(max=50)])
@ -582,6 +601,9 @@ class AllocateForm(NewActionForm):
def validate(self, extra_validators=None): def validate(self, extra_validators=None):
is_valid = super().validate(extra_validators) is_valid = super().validate(extra_validators)
if self.type.data not in ['Allocate', 'Deallocate']:
return False
start_time = self.start_time.data start_time = self.start_time.data
end_time = self.end_time.data end_time = self.end_time.data
if start_time and end_time and end_time < start_time: if start_time and end_time and end_time < start_time:
@ -650,7 +672,7 @@ class DataWipeDocumentForm(Form):
return self._obj return self._obj
class DataWipeForm(NewActionForm): class DataWipeForm(ActionFormMix):
document = FormField(DataWipeDocumentForm) document = FormField(DataWipeDocumentForm)
def save(self): def save(self):
@ -677,7 +699,7 @@ class DataWipeForm(NewActionForm):
return self.instance return self.instance
class TradeForm(NewActionForm): class TradeForm(ActionFormMix):
user_from = StringField( user_from = StringField(
'Supplier', 'Supplier',
[validators.Optional()], [validators.Optional()],
@ -724,6 +746,9 @@ class TradeForm(NewActionForm):
email_from = self.user_from.data email_from = self.user_from.data
email_to = self.user_to.data email_to = self.user_to.data
if self.type.data != "Trade":
return False
if not self.confirm.data and not self.code.data: if not self.confirm.data and not self.code.data:
self.code.errors = ["If you don't want to confirm, you need a code"] self.code.errors = ["If you don't want to confirm, you need a code"]
is_valid = False is_valid = False

View file

@ -315,16 +315,19 @@ class NewActionView(View):
def dispatch_request(self): def dispatch_request(self):
self.form = self.form_class() self.form = self.form_class()
next_url = self.get_next_url()
if self.form.validate_on_submit(): if self.form.validate_on_submit():
self.form.save() self.form.save()
messages.success( messages.success(
'Action "{}" created successfully!'.format(self.form.type.data) 'Action "{}" created successfully!'.format(self.form.type.data)
) )
next_url = self.get_next_url() next_url = self.get_next_url()
return flask.redirect(next_url) return flask.redirect(next_url)
messages.error('Action {} error!'.format(self.form.type.data))
return flask.redirect(next_url)
def get_next_url(self): def get_next_url(self):
lot_id = self.form.lot.data lot_id = self.form.lot.data
@ -350,10 +353,12 @@ class NewAllocateView(NewActionView, DeviceListMix):
next_url = self.get_next_url() next_url = self.get_next_url()
return flask.redirect(next_url) return flask.redirect(next_url)
lot_id = self.form.lot.data messages.error('Action {} error!'.format(self.form.type.data))
self.get_context(lot_id) for k, v in self.form.errors.items():
self.context['form_new_allocate'] = self.form value = ';'.join(v)
return flask.render_template(self.template_name, **self.context) messages.error('Action Error {key}: {value}!'.format(key=k, value=value))
next_url = self.get_next_url()
return flask.redirect(next_url)
class NewDataWipeView(NewActionView, DeviceListMix): class NewDataWipeView(NewActionView, DeviceListMix):
@ -372,10 +377,9 @@ class NewDataWipeView(NewActionView, DeviceListMix):
next_url = self.get_next_url() next_url = self.get_next_url()
return flask.redirect(next_url) return flask.redirect(next_url)
lot_id = self.form.lot.data messages.error('Action {} error!'.format(self.form.type.data))
self.get_context(lot_id) next_url = self.get_next_url()
self.context['form_new_datawipe'] = self.form return flask.redirect(next_url)
return flask.render_template(self.template_name, **self.context)
class NewTradeView(NewActionView, DeviceListMix): class NewTradeView(NewActionView, DeviceListMix):
@ -394,10 +398,9 @@ class NewTradeView(NewActionView, DeviceListMix):
next_url = self.get_next_url() next_url = self.get_next_url()
return flask.redirect(next_url) return flask.redirect(next_url)
lot_id = self.form.lot.data messages.error('Action {} error!'.format(self.form.type.data))
self.get_context(lot_id) next_url = self.get_next_url()
self.context['form_new_trade'] = self.form return flask.redirect(next_url)
return flask.render_template(self.template_name, **self.context)
class NewTradeDocumentView(View): class NewTradeDocumentView(View):

View file

@ -1,27 +1,29 @@
import io import io
import uuid import uuid
import jwt
import ereuse_utils
from contextlib import redirect_stdout from contextlib import redirect_stdout
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from decouple import config
import boltons.urlutils import boltons.urlutils
import ereuse_utils
import jwt
import pytest import pytest
import yaml import yaml
from decouple import config
from psycopg2 import IntegrityError from psycopg2 import IntegrityError
from sqlalchemy.exc import ProgrammingError from sqlalchemy.exc import ProgrammingError
from ereuse_devicehub.client import Client, UserClient from ereuse_devicehub.client import Client, UserClient, UserClientFlask
from ereuse_devicehub.config import DevicehubConfig from ereuse_devicehub.config import DevicehubConfig
from ereuse_devicehub.db import db from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.inventory.views import devices
from ereuse_devicehub.labels.views import labels
from ereuse_devicehub.resources.agent.models import Person from ereuse_devicehub.resources.agent.models import Person
from ereuse_devicehub.resources.tag import Tag
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.resources.user.models import Session
from ereuse_devicehub.resources.enums import SessionType from ereuse_devicehub.resources.enums import SessionType
from ereuse_devicehub.resources.tag import Tag
from ereuse_devicehub.resources.user.models import Session, User
from ereuse_devicehub.views import core
STARTT = datetime(year=2000, month=1, day=1, hour=1) STARTT = datetime(year=2000, month=1, day=1, hour=1)
"""A dummy starting time to use in tests.""" """A dummy starting time to use in tests."""
@ -50,6 +52,20 @@ def config():
@pytest.fixture(scope='session') @pytest.fixture(scope='session')
def _app(config: TestConfig) -> Devicehub: def _app(config: TestConfig) -> Devicehub:
# dh_config = DevicehubConfig()
# config = TestConfig(dh_config)
app = Devicehub(inventory='test', config=config, db=db)
app.register_blueprint(core)
app.register_blueprint(devices)
app.register_blueprint(labels)
app.config["SQLALCHEMY_RECORD_QUERIES"] = True
app.config['PROFILE'] = True
# app.wsgi_app = ProfilerMiddleware(app.wsgi_app, restrictions=[30])
return app
@pytest.fixture(scope='session')
def _app2(config: TestConfig) -> Devicehub:
return Devicehub(inventory='test', config=config, db=db) return Devicehub(inventory='test', config=config, db=db)
@ -61,13 +77,15 @@ def app(request, _app: Devicehub) -> Devicehub:
db.drop_all() db.drop_all()
def _init(): def _init():
_app.init_db(name='Test Inventory', _app.init_db(
name='Test Inventory',
org_name='FooOrg', org_name='FooOrg',
org_id='foo-org-id', org_id='foo-org-id',
tag_url=boltons.urlutils.URL('https://example.com'), tag_url=boltons.urlutils.URL('https://example.com'),
tag_token=uuid.UUID('52dacef0-6bcb-4919-bfed-f10d2c96ecee'), tag_token=uuid.UUID('52dacef0-6bcb-4919-bfed-f10d2c96ecee'),
erase=False, erase=False,
common=True) common=True,
)
with _app.app_context(): with _app.app_context():
try: try:
@ -99,7 +117,9 @@ def user(app: Devicehub) -> UserClient:
with app.app_context(): with app.app_context():
password = 'foo' password = 'foo'
user = create_user(password=password) user = create_user(password=password)
client = UserClient(app, user.email, password, response_wrapper=app.response_class) client = UserClient(
app, user.email, password, response_wrapper=app.response_class
)
client.login() client.login()
return client return client
@ -111,11 +131,34 @@ def user2(app: Devicehub) -> UserClient:
password = 'foo' password = 'foo'
email = 'foo2@foo.com' email = 'foo2@foo.com'
user = create_user(email=email, password=password) user = create_user(email=email, password=password)
client = UserClient(app, user.email, password, response_wrapper=app.response_class) client = UserClient(
app, user.email, password, response_wrapper=app.response_class
)
client.login() client.login()
return client return client
@pytest.fixture()
def user3(app: Devicehub) -> UserClientFlask:
"""Gets a client with a logged-in dummy user."""
with app.app_context():
password = 'foo'
user = create_user(password=password)
client = UserClientFlask(app, user.email, password)
return client
@pytest.fixture()
def user4(app: Devicehub) -> UserClient:
"""Gets a client with a logged-in dummy user."""
with app.app_context():
password = 'foo'
email = 'foo2@foo.com'
user = create_user(email=email, password=password)
client = UserClientFlask(app, user.email, password)
return client
def create_user(email='foo@foo.com', password='foo') -> User: def create_user(email='foo@foo.com', password='foo') -> User:
user = User(email=email, password=password) user = User(email=email, password=password)
user.individuals.add(Person(name='Timmy')) user.individuals.add(Person(name='Timmy'))
@ -145,16 +188,13 @@ def auth_app_context(app: Devicehub):
def json_encode(dev: str) -> dict: def json_encode(dev: str) -> dict:
"""Encode json.""" """Encode json."""
data = {"type": "Snapshot"} data = {"type": "Snapshot"}
data['data'] = jwt.encode(dev, data['data'] = jwt.encode(
P, dev, P, algorithm="HS256", json_encoder=ereuse_utils.JSONEncoder
algorithm="HS256",
json_encoder=ereuse_utils.JSONEncoder
) )
return data return data
def yaml2json(name: str) -> dict: def yaml2json(name: str) -> dict:
"""Opens and parses a YAML file from the ``files`` subdir.""" """Opens and parses a YAML file from the ``files`` subdir."""
with Path(__file__).parent.joinpath('files').joinpath(name + '.yaml').open() as f: with Path(__file__).parent.joinpath('files').joinpath(name + '.yaml').open() as f:
@ -168,7 +208,9 @@ def file(name: str) -> dict:
def file_workbench(name: str) -> dict: def file_workbench(name: str) -> dict:
"""Opens and parses a YAML file from the ``files`` subdir.""" """Opens and parses a YAML file from the ``files`` subdir."""
with Path(__file__).parent.joinpath('workbench_files').joinpath(name + '.json').open() as f: with Path(__file__).parent.joinpath('workbench_files').joinpath(
name + '.json'
).open() as f:
return yaml.load(f) return yaml.load(f)

File diff suppressed because one or more lines are too long

View file

@ -1,7 +1,7 @@
import pytest import pytest
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.client import Client from ereuse_devicehub.client import Client
from ereuse_devicehub.devicehub import Devicehub
@pytest.mark.mvp @pytest.mark.mvp
@ -28,37 +28,65 @@ def test_api_docs(client: Client):
"""Tests /apidocs correct initialization.""" """Tests /apidocs correct initialization."""
docs, _ = client.get('/apidocs') docs, _ = client.get('/apidocs')
assert set(docs['paths'].keys()) == { assert set(docs['paths'].keys()) == {
'/',
'/actions/', '/actions/',
'/apidocs',
'/allocates/', '/allocates/',
'/apidocs',
'/deallocates/', '/deallocates/',
'/deliverynotes/', '/deliverynotes/',
'/devices/', '/devices/',
'/devices/static/{filename}', '/devices/static/{filename}',
'/documents/static/{filename}',
'/documents/actions/', '/documents/actions/',
'/documents/erasures/',
'/documents/devices/',
'/documents/stamps/',
'/documents/wbconf/{wbtype}',
'/documents/internalstats/',
'/documents/stock/',
'/documents/check/', '/documents/check/',
'/documents/devices/',
'/documents/erasures/',
'/documents/internalstats/',
'/documents/lots/', '/documents/lots/',
'/versions/', '/documents/stamps/',
'/manufacturers/', '/documents/static/{filename}',
'/documents/stock/',
'/documents/wbconf/{wbtype}',
'/inventory/action/add/',
'/inventory/action/allocate/add/',
'/inventory/action/datawipe/add/',
'/inventory/action/trade/add/',
'/inventory/device/',
'/inventory/device/add/',
'/inventory/device/{id}/',
'/inventory/export/{export_id}/',
'/inventory/lot/add/',
'/inventory/lot/{id}/',
'/inventory/lot/{id}/del/',
'/inventory/lot/{lot_id}/device/',
'/inventory/lot/{lot_id}/device/add/',
'/inventory/lot/{lot_id}/trade-document/add/',
'/inventory/lot/{lot_id}/upload-snapshot/',
'/inventory/tag/devices/add/',
'/inventory/tag/devices/{id}/del/',
'/inventory/upload-snapshot/',
'/labels/',
'/labels/add/',
'/labels/print',
'/labels/unnamed/add/',
'/labels/{id}/',
'/licences/', '/licences/',
'/lives/', '/lives/',
'/login/',
'/logout/',
'/lots/', '/lots/',
'/lots/{id}/children', '/lots/{id}/children',
'/lots/{id}/devices', '/lots/{id}/devices',
'/manufacturers/',
'/metrics/', '/metrics/',
'/profile/',
'/set_password/',
'/tags/', '/tags/',
'/tags/{tag_id}/device/{device_id}', '/tags/{tag_id}/device/{device_id}',
'/trade-documents/', '/trade-documents/',
'/users/', '/users/',
'/users/login/', '/users/login/',
'/users/logout/', '/users/logout/',
'/versions/',
} }
assert docs['info'] == {'title': 'Devicehub', 'version': '0.2'} assert docs['info'] == {'title': 'Devicehub', 'version': '0.2'}
assert docs['components']['securitySchemes']['bearerAuth'] == { assert docs['components']['securitySchemes']['bearerAuth'] == {
@ -67,6 +95,6 @@ def test_api_docs(client: Client):
'description:': 'HTTP Basic scheme', 'description:': 'HTTP Basic scheme',
'type': 'http', 'type': 'http',
'scheme': 'basic', 'scheme': 'basic',
'name': 'Authorization' 'name': 'Authorization',
} }
assert len(docs['definitions']) == 132 assert len(docs['definitions']) == 132

856
tests/test_render_2_0.py Normal file
View file

@ -0,0 +1,856 @@
import csv
import json
from io import BytesIO
from pathlib import Path
import pytest
from flask.testing import FlaskClient
from flask_wtf.csrf import generate_csrf
from ereuse_devicehub.client import UserClient, UserClientFlask
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.action.models import Snapshot
from ereuse_devicehub.resources.device.models import Device
from ereuse_devicehub.resources.lot.models import Lot
from tests import conftest
def create_device(user, file_name):
uri = '/inventory/upload-snapshot/'
snapshot = conftest.yaml2json(file_name.split(".json")[0])
b_snapshot = bytes(json.dumps(snapshot), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
user.get(uri)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
user.post(uri, data=data, content_type="multipart/form-data")
return Snapshot.query.one()
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_login(user: UserClient, app: Devicehub):
"""Checks a simple login"""
client = FlaskClient(app, use_cookies=True)
body, status, headers = client.get('/login/')
body = next(body).decode("utf-8")
assert status == '200 OK'
assert "Login to Your Account" in body
data = {
'email': user.email,
'password': 'foo',
'remember': False,
'csrf_token': generate_csrf(),
}
body, status, headers = client.post('/login/', data=data, follow_redirects=True)
body = next(body).decode("utf-8")
assert status == '200 OK'
assert "Login to Your Account" not in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_profile(user3: UserClientFlask):
body, status = user3.get('/profile/')
assert status == '200 OK'
assert "Profile" in body
assert user3.email in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_inventory(user3: UserClientFlask):
body, status = user3.get('/inventory/device/')
assert status == '200 OK'
assert "Unassgined" in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_add_lot(user3: UserClientFlask):
body, status = user3.get('/inventory/lot/add/')
lot_name = "lot1"
assert status == '200 OK'
assert "Add a new lot" in body
assert lot_name not in body
data = {
'name': lot_name,
'csrf_token': generate_csrf(),
}
body, status = user3.post('/inventory/lot/add/', data=data)
assert status == '200 OK'
assert lot_name in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_del_lot(user3: UserClientFlask):
body, status = user3.get('/inventory/lot/add/')
lot_name = "lot1"
assert status == '200 OK'
assert "Add a new lot" in body
assert lot_name not in body
data = {
'name': lot_name,
'csrf_token': generate_csrf(),
}
body, status = user3.post('/inventory/lot/add/', data=data)
assert status == '200 OK'
assert lot_name in body
lot = Lot.query.filter_by(name=lot_name).one()
uri = '/inventory/lot/{id}/del/'.format(id=lot.id)
body, status = user3.get(uri)
assert lot_name not in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_update_lot(user3: UserClientFlask):
user3.get('/inventory/lot/add/')
# Add lot
data = {
'name': "lot1",
'csrf_token': generate_csrf(),
}
user3.post('/inventory/lot/add/', data=data)
data = {
'name': "lot2",
'csrf_token': generate_csrf(),
}
lot = Lot.query.one()
uri = '/inventory/lot/{uuid}/'.format(uuid=lot.id)
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert "lot2" in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_upload_snapshot(user3: UserClientFlask):
uri = '/inventory/upload-snapshot/'
file_name = 'real-eee-1001pxd.snapshot.12.json'
body, status = user3.get(uri)
assert status == '200 OK'
assert "Select a Snapshot file" in body
snapshot = conftest.yaml2json(file_name.split(".json")[0])
b_snapshot = bytes(json.dumps(snapshot), 'utf-8')
file_snap = (BytesIO(b_snapshot), file_name)
data = {
'snapshot': file_snap,
'csrf_token': generate_csrf(),
}
body, status = user3.post(uri, data=data, content_type="multipart/form-data")
txt = f"{file_name}: Ok"
assert status == '200 OK'
assert txt in body
db_snapthot = Snapshot.query.one()
dev = db_snapthot.device
assert str(db_snapthot.uuid) == snapshot['uuid']
assert dev.type == 'Laptop'
assert dev.serial_number == 'b8oaas048285'
assert len(dev.actions) == 12
assert len(dev.components) == 9
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_inventory_with_device(user3: UserClientFlask):
db_snapthot = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
body, status = user3.get('/inventory/device/')
assert status == '200 OK'
assert "Unassgined" in body
assert db_snapthot.device.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_inventory_filter(user3: UserClientFlask):
db_snapthot = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
csrf = generate_csrf()
body, status = user3.get(f'/inventory/device/?filter=Laptop&csrf_token={csrf}')
assert status == '200 OK'
assert "Unassgined" in body
assert db_snapthot.device.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_export_devices(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
uri = "/inventory/export/devices/?ids={id}".format(id=snap.device.devicehub_id)
body, status = user3.get(uri)
assert status == '200 OK'
export_csv = [line.split(";") for line in body.split("\n")]
with Path(__file__).parent.joinpath('files').joinpath(
'export_devices.csv'
).open() as csv_file:
obj_csv = csv.reader(csv_file, delimiter=';', quotechar='"')
fixture_csv = list(obj_csv)
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][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 fixture_csv[1][83] == export_csv[1][83], 'Computer information are not equal'
assert (
fixture_csv[1][86:] == export_csv[1][86:]
), 'Computer information are not equal'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_export_metrics(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
uri = "/inventory/export/metrics/?ids={id}".format(id=snap.device.devicehub_id)
body, status = user3.get(uri)
assert status == '200 OK'
assert body == ''
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_export_links(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
uri = "/inventory/export/links/?ids={id}".format(id=snap.device.devicehub_id)
body, status = user3.get(uri)
assert status == '200 OK'
body = body.split("\n")
assert ['links', 'http://localhost/devices/O48N2', ''] == body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_export_certificates(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
uri = "/inventory/export/certificates/?ids={id}".format(id=snap.device.devicehub_id)
body, status = user3.get(uri, decode=False)
body = str(next(body))
assert status == '200 OK'
assert "PDF-1.5" in body
assert 'hts54322' in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_labels(user3: UserClientFlask):
body, status = user3.get('/labels/')
assert status == '200 OK'
assert "Unique Identifiers Management" in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_add_tag(user3: UserClientFlask):
uri = '/labels/add/'
body, status = user3.get(uri)
assert status == '200 OK'
assert "Add a new Unique Identifier" in body
data = {
'code': "tag1",
'csrf_token': generate_csrf(),
}
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert "tag1" in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_label_details(user3: UserClientFlask):
uri = '/labels/add/'
user3.get(uri)
data = {
'code': "tag1",
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data)
body, status = user3.get('/labels/tag1/')
assert "tag1" in body
assert "Print Label" in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_link_tag_to_device(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/labels/add/'
user3.get(uri)
data = {
'code': "tag1",
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data)
body, status = user3.get('/inventory/device/')
assert "tag1" in body
data = {
'tag': "tag1",
'device': dev.id,
'csrf_token': generate_csrf(),
}
uri = '/inventory/tag/devices/add/'
user3.post(uri, data=data)
assert len(list(dev.tags)) == 2
tags = [tag.id for tag in dev.tags]
assert "tag1" in tags
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_unlink_tag_to_device(user3: UserClientFlask):
# create device
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
# create tag
uri = '/labels/add/'
user3.get(uri)
data = {
'code': "tag1",
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data)
# link tag to device
data = {
'tag': "tag1",
'device': dev.id,
'csrf_token': generate_csrf(),
}
uri = '/inventory/tag/devices/add/'
user3.post(uri, data=data)
# unlink tag to device
uri = '/inventory/tag/devices/{id}/del/'.format(id=dev.id)
user3.get(uri)
data = {
'code': "tag1",
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data)
data = {
'tag': "tag1",
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data)
assert len(list(dev.tags)) == 1
tag = list(dev.tags)[0]
assert not tag.id == "tag1"
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_print_labels(user3: UserClientFlask):
# create device
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
# create tag
uri = '/labels/add/'
user3.get(uri)
data = {
'code': "tag1",
'csrf_token': generate_csrf(),
}
user3.post(uri, data=data)
# link tag to device
data = {
'tag': "tag1",
'device': dev.id,
'csrf_token': generate_csrf(),
}
uri = '/inventory/tag/devices/add/'
user3.post(uri, data=data)
assert len(list(dev.tags)) == 2
uri = '/labels/print'
data = {
'devices': "{}".format(dev.id),
'csrf_token': generate_csrf(),
}
body, status = user3.post(uri, data=data)
assert status == '200 OK'
path = "/inventory/device/{}/".format(dev.devicehub_id)
assert path in body
assert "tag1" not in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_add_monitor(user3: UserClientFlask):
uri = '/inventory/device/add/'
body, status = user3.get(uri)
assert status == '200 OK'
assert "New Device" in body
data = {
'csrf_token': generate_csrf(),
'type': "Monitor",
'serial_number': "AAAAB",
'model': "LC27T55",
'manufacturer': "Samsung",
'generation': 1,
'weight': 0.1,
'height': 0.1,
'depth': 0.1,
}
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert 'Device &#34;Monitor&#34; created successfully!' in body
dev = Device.query.one()
assert dev.type == 'Monitor'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_filter_monitor(user3: UserClientFlask):
uri = '/inventory/device/add/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "Monitor",
'serial_number': "AAAAB",
'model': "LC27T55",
'manufacturer': "Samsung",
'generation': 1,
'weight': 0.1,
'height': 0.1,
'depth': 0.1,
}
user3.post(uri, data=data)
csrf = generate_csrf()
uri = f'/inventory/device/?filter=Monitor&csrf_token={csrf}'
body, status = user3.get(uri)
assert status == '200 OK'
dev = Device.query.one()
assert dev.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_recycling(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
# fail request
data = {
'csrf_token': generate_csrf(),
'type': "Allocate",
'severity': "Info",
'devices': "{}".format(dev.id),
}
uri = '/inventory/action/add/'
body, status = user3.post(uri, data=data)
assert dev.actions[-1].type == 'EreusePrice'
assert 'Action Allocate error!' in body
# good request
data = {
'csrf_token': generate_csrf(),
'type': "Recycling",
'severity': "Info",
'devices': "{}".format(dev.id),
}
uri = '/inventory/action/add/'
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert dev.actions[-1].type == 'Recycling'
assert 'Action &#34;Recycling&#34; created successfully!' in body
assert dev.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_error_without_devices(user3: UserClientFlask):
uri = '/inventory/device/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "Recycling",
'severity': "Info",
'devices': "",
}
uri = '/inventory/action/add/'
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert 'Action Recycling error!' in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_use(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "Use",
'severity': "Info",
'devices': "{}".format(dev.id),
}
uri = '/inventory/action/add/'
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert dev.actions[-1].type == 'Use'
assert 'Action &#34;Use&#34; created successfully!' in body
assert dev.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_refurbish(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "Refurbish",
'severity': "Info",
'devices': "{}".format(dev.id),
}
uri = '/inventory/action/add/'
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert dev.actions[-1].type == 'Refurbish'
assert 'Action &#34;Refurbish&#34; created successfully!' in body
assert dev.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_management(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "Management",
'severity': "Info",
'devices': "{}".format(dev.id),
}
uri = '/inventory/action/add/'
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert dev.actions[-1].type == 'Management'
assert 'Action &#34;Management&#34; created successfully!' in body
assert dev.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_allocate(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "Allocate",
'severity': "Info",
'devices': "{}".format(dev.id),
'start_time': '2000-01-01',
'end_time': '2000-06-01',
'end_users': 2,
}
uri = '/inventory/action/allocate/add/'
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert dev.actions[-1].type == 'Allocate'
assert 'Action &#34;Allocate&#34; created successfully!' in body
assert dev.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_allocate_error_required(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "Trade",
'severity': "Info",
'devices': "{}".format(dev.id),
}
uri = '/inventory/action/allocate/add/'
body, status = user3.post(uri, data=data)
assert dev.actions[-1].type != 'Allocate'
data = {
'csrf_token': generate_csrf(),
'type': "Allocate",
'severity': "Info",
'devices': "{}".format(dev.id),
}
uri = '/inventory/action/allocate/add/'
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert 'Action Allocate error' in body
assert 'You need to specify a number of users!' in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_allocate_error_dates(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "Allocate",
'severity': "Info",
'devices': "{}".format(dev.id),
'start_time': '2000-06-01',
'end_time': '2000-01-01',
'end_users': 2,
}
uri = '/inventory/action/allocate/add/'
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert 'Action Allocate error' in body
assert 'The action cannot finish before it starts.' in body
assert dev.actions[-1].type != 'Allocate'
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_deallocate(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "Allocate",
'severity': "Info",
'devices': "{}".format(dev.id),
'start_time': '2000-01-01',
'end_time': '2000-06-01',
'end_users': 2,
}
uri = '/inventory/action/allocate/add/'
user3.post(uri, data=data)
assert dev.actions[-1].type == 'Allocate'
data = {
'csrf_token': generate_csrf(),
'type': "Deallocate",
'severity': "Info",
'devices': "{}".format(dev.id),
'start_time': '2000-01-01',
'end_time': '2000-06-01',
'end_users': 2,
}
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert dev.actions[-1].type == 'Deallocate'
assert 'Action &#34;Deallocate&#34; created successfully!' in body
assert dev.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_toprepare(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "ToPrepare",
'severity': "Info",
'devices': "{}".format(dev.id),
}
uri = '/inventory/action/add/'
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert dev.actions[-1].type == 'ToPrepare'
assert 'Action &#34;ToPrepare&#34; created successfully!' in body
assert dev.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_prepare(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "Prepare",
'severity': "Info",
'devices': "{}".format(dev.id),
}
uri = '/inventory/action/add/'
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert dev.actions[-1].type == 'Prepare'
assert 'Action &#34;Prepare&#34; created successfully!' in body
assert dev.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_torepair(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "ToRepair",
'severity': "Info",
'devices': "{}".format(dev.id),
}
uri = '/inventory/action/add/'
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert dev.actions[-1].type == 'ToRepair'
assert 'Action &#34;ToRepair&#34; created successfully!' in body
assert dev.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_ready(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
data = {
'csrf_token': generate_csrf(),
'type': "Ready",
'severity': "Info",
'devices': "{}".format(dev.id),
}
uri = '/inventory/action/add/'
body, status = user3.post(uri, data=data)
assert status == '200 OK'
assert dev.actions[-1].type == 'Ready'
assert 'Action &#34;Ready&#34; created successfully!' in body
assert dev.devicehub_id in body
@pytest.mark.mvp
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_action_datawipe(user3: UserClientFlask):
snap = create_device(user3, 'real-eee-1001pxd.snapshot.12.json')
dev = snap.device
uri = '/inventory/device/'
user3.get(uri)
b_file = b'1234567890'
file_name = "my_file.doc"
file_upload = (BytesIO(b_file), file_name)
data = {
'csrf_token': generate_csrf(),
'type': "DataWipe",
'severity': "Info",
'devices': "{}".format(dev.id),
'document-file_name': file_upload,
}
uri = '/inventory/action/datawipe/add/'
body, status = user3.post(uri, data=data, content_type="multipart/form-data")
assert status == '200 OK'
assert dev.actions[-1].type == 'DataWipe'
assert 'Action &#34;DataWipe&#34; created successfully!' in body
assert dev.devicehub_id in body