Merge branch 'testing' into bugfix/#3288-common-lots
This commit is contained in:
commit
98ffe1092c
|
@ -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)
|
||||||
|
|
|
@ -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
|
||||||
|
|
|
@ -303,16 +303,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
|
||||||
|
|
||||||
|
@ -338,10 +341,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):
|
||||||
|
@ -360,10 +365,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):
|
||||||
|
@ -382,10 +386,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):
|
||||||
|
|
|
@ -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
|
@ -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
|
||||||
|
|
|
@ -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 "Monitor" 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 "Recycling" 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 "Use" 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 "Refurbish" 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 "Management" 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 "Allocate" 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 "Deallocate" 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 "ToPrepare" 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 "Prepare" 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 "ToRepair" 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 "Ready" 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 "DataWipe" created successfully!' in body
|
||||||
|
assert dev.devicehub_id in body
|
Reference in New Issue