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

resolve conflict

This commit is contained in:
Cayo Puigdefabregas 2022-04-29 19:26:44 +02:00
parent 02bfeea2fc 60a11794ae
commit 67ee1cdde8
8 changed files with 1376 additions and 267 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

353
ereuse_devicehub/client.py
View File

@ -2,7 +2,10 @@ from inspect import isclass
from typing import Dict, Iterable, Type, Union
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 ereuse_devicehub.resources import models, schemas
@ -13,110 +16,156 @@ ResourceLike = Union[Type[Union[models.Thing, schemas.Thing]], str]
class Client(TealClient):
"""A client suited for Devicehub main usage."""
def __init__(self, application,
response_wrapper=None,
use_cookies=False,
allow_subdomain_redirects=False):
super().__init__(application, response_wrapper, use_cookies, allow_subdomain_redirects)
def __init__(
self,
application,
response_wrapper=None,
use_cookies=False,
allow_subdomain_redirects=False,
):
super().__init__(
application, response_wrapper, use_cookies, allow_subdomain_redirects
)
def open(self,
uri: str,
res: ResourceLike = None,
status: Status = 200,
query: Query = tuple(),
accept=JSON,
content_type=JSON,
item=None,
headers: dict = None,
token: str = None,
**kw) -> Res:
def open(
self,
uri: str,
res: ResourceLike = None,
status: Status = 200,
query: Query = tuple(),
accept=JSON,
content_type=JSON,
item=None,
headers: dict = None,
token: str = None,
**kw,
) -> Res:
if isclass(res) and issubclass(res, (models.Thing, schemas.Thing)):
res = res.t
return super().open(uri, res, status, query, accept, content_type, item, headers, token,
**kw)
return super().open(
uri, res, status, query, accept, content_type, item, headers, token, **kw
)
def get(self,
uri: str = '',
res: ResourceLike = None,
query: Query = tuple(),
status: Status = 200,
item: Union[int, str] = None,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw) -> Res:
def get(
self,
uri: str = '',
res: ResourceLike = None,
query: Query = tuple(),
status: Status = 200,
item: Union[int, str] = None,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw,
) -> Res:
return super().get(uri, res, query, status, item, accept, headers, token, **kw)
def post(self,
data: str or dict,
uri: str = '',
res: ResourceLike = None,
query: Query = tuple(),
status: Status = 201,
content_type: str = JSON,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw) -> Res:
return super().post(data, uri, res, query, status, content_type, accept, headers, token,
**kw)
def post(
self,
data: str or dict,
uri: str = '',
res: ResourceLike = None,
query: Query = tuple(),
status: Status = 201,
content_type: str = JSON,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw,
) -> Res:
return super().post(
data, uri, res, query, status, content_type, accept, headers, token, **kw
)
def patch(self,
data: str or dict,
uri: str = '',
res: ResourceLike = None,
query: Query = tuple(),
item: Union[int, str] = None,
status: Status = 200,
content_type: str = JSON,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw) -> Res:
return super().patch(data, uri, res, query, item, status, content_type, accept, token,
headers, **kw)
def patch(
self,
data: str or dict,
uri: str = '',
res: ResourceLike = None,
query: Query = tuple(),
item: Union[int, str] = None,
status: Status = 200,
content_type: str = JSON,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw,
) -> Res:
return super().patch(
data,
uri,
res,
query,
item,
status,
content_type,
accept,
token,
headers,
**kw,
)
def put(self,
data: str or dict,
uri: str = '',
res: ResourceLike = None,
query: Query = tuple(),
item: Union[int, str] = None,
status: Status = 201,
content_type: str = JSON,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw) -> Res:
return super().put(data, uri, res, query, item, status, content_type, accept, token,
headers, **kw)
def put(
self,
data: str or dict,
uri: str = '',
res: ResourceLike = None,
query: Query = tuple(),
item: Union[int, str] = None,
status: Status = 201,
content_type: str = JSON,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw,
) -> Res:
return super().put(
data,
uri,
res,
query,
item,
status,
content_type,
accept,
token,
headers,
**kw,
)
def delete(self,
uri: str = '',
res: ResourceLike = None,
query: Query = tuple(),
status: Status = 204,
item: Union[int, str] = None,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw) -> Res:
return super().delete(uri, res, query, status, item, accept, headers, token, **kw)
def delete(
self,
uri: str = '',
res: ResourceLike = None,
query: Query = tuple(),
status: Status = 204,
item: Union[int, str] = None,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw,
) -> Res:
return super().delete(
uri, res, query, status, item, accept, headers, token, **kw
)
def login(self, email: str, password: str):
assert isinstance(email, 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,
res: ResourceLike,
resources: Iterable[Union[dict, int]],
key: str = None,
**kw) -> Iterable[Union[Dict[str, object], str]]:
def get_many(
self,
res: ResourceLike,
resources: Iterable[Union[dict, int]],
key: str = None,
**kw,
) -> Iterable[Union[Dict[str, object], str]]:
"""Like :meth:`.get` but with many resources."""
return (
self.get(res=res, item=r[key] if key else r, **kw)[0]
for r in resources
self.get(res=res, item=r[key] if key else r, **kw)[0] for r in resources
)
@ -126,33 +175,119 @@ class UserClient(Client):
It will automatically perform login on the first request.
"""
def __init__(self, application,
email: str,
password: str,
response_wrapper=None,
use_cookies=False,
allow_subdomain_redirects=False):
super().__init__(application, response_wrapper, use_cookies, allow_subdomain_redirects)
def __init__(
self,
application,
email: str,
password: str,
response_wrapper=None,
use_cookies=False,
allow_subdomain_redirects=False,
):
super().__init__(
application, response_wrapper, use_cookies, allow_subdomain_redirects
)
self.email = email # type: str
self.password = password # type: str
self.user = None # type: dict
def open(self,
uri: str,
res: ResourceLike = None,
status: int or HTTPException = 200,
query: Query = tuple(),
accept=JSON,
content_type=JSON,
item=None,
headers: dict = None,
token: str = None,
**kw) -> Res:
return super().open(uri, res, status, query, accept, content_type, item, headers,
self.user['token'] if self.user else token, **kw)
def open(
self,
uri: str,
res: ResourceLike = None,
status: int or HTTPException = 200,
query: Query = tuple(),
accept=JSON,
content_type=JSON,
item=None,
headers: dict = None,
token: str = None,
**kw,
) -> Res:
return super().open(
uri,
res,
status,
query,
accept,
content_type,
item,
headers,
self.user['token'] if self.user else token,
**kw,
)
# noinspection PyMethodOverriding
def login(self):
response = super().login(self.email, self.password)
self.user = response[0]
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)

53
ereuse_devicehub/inventory/forms.py
View File

@ -468,7 +468,7 @@ class TagDeviceForm(FlaskForm):
db.session.commit()
class NewActionForm(FlaskForm):
class ActionFormMix(FlaskForm):
name = StringField(
'Name',
[validators.length(max=50)],
@ -500,17 +500,23 @@ class NewActionForm(FlaskForm):
if not is_valid:
return False
self._devices = OrderedSet()
if self.devices.data:
devices = set(self.devices.data.split(","))
self._devices = OrderedSet(
Device.query.filter(Device.id.in_(devices))
.filter(Device.owner_id == g.user.id)
.all()
)
if self.type.data in [None, '']:
return False
if not self._devices:
return False
if not self.devices.data:
return False
self._devices = OrderedSet()
devices = set(self.devices.data.split(","))
self._devices = OrderedSet(
Device.query.filter(Device.id.in_(devices))
.filter(Device.owner_id == g.user.id)
.all()
)
if not self._devices:
return False
return True
@ -543,7 +549,20 @@ class NewActionForm(FlaskForm):
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')
end_time = DateField('End time')
final_user_code = StringField('Final user code', [validators.length(max=50)])
@ -553,6 +572,9 @@ class AllocateForm(NewActionForm):
def validate(self, extra_validators=None):
is_valid = super().validate(extra_validators)
if self.type.data not in ['Allocate', 'Deallocate']:
return False
start_time = self.start_time.data
end_time = self.end_time.data
if start_time and end_time and end_time < start_time:
@ -621,7 +643,7 @@ class DataWipeDocumentForm(Form):
return self._obj
class DataWipeForm(NewActionForm):
class DataWipeForm(ActionFormMix):
document = FormField(DataWipeDocumentForm)
def save(self):
@ -648,7 +670,7 @@ class DataWipeForm(NewActionForm):
return self.instance
class TradeForm(NewActionForm):
class TradeForm(ActionFormMix):
user_from = StringField(
'Supplier',
[validators.Optional()],
@ -695,6 +717,9 @@ class TradeForm(NewActionForm):
email_from = self.user_from.data
email_to = self.user_to.data
if self.type.data != "Trade":
return False
if not self.confirm.data and not self.code.data:
self.code.errors = ["If you don't want to confirm, you need a code"]
is_valid = False

191
ereuse_devicehub/inventory/views.py
View File

@ -7,10 +7,9 @@ import flask_weasyprint
from flask import Blueprint, g, make_response, request, url_for
from flask.views import View
from flask_login import current_user, login_required
from sqlalchemy import or_
from werkzeug.exceptions import NotFound
from ereuse_devicehub import __version__, messages
from ereuse_devicehub import messages
from ereuse_devicehub.db import db
from ereuse_devicehub.inventory.forms import (
AllocateForm,
@ -31,35 +30,21 @@ from ereuse_devicehub.resources.documents.device_row import ActionRow, DeviceRow
from ereuse_devicehub.resources.hash_reports import insert_hash
from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.tag.model import Tag
from ereuse_devicehub.views import GenericMixView
devices = Blueprint('inventory', __name__, url_prefix='/inventory')
logger = logging.getLogger(__name__)
class GenericMixView(View):
def get_lots(self):
return (
Lot.query.outerjoin(Trade)
.filter(
or_(
Trade.user_from == g.user,
Trade.user_to == g.user,
Lot.owner_id == g.user.id,
)
)
.distinct()
)
class DeviceListMix(GenericMixView):
decorators = [login_required]
template_name = 'inventory/device_list.html'
def get_context(self, lot_id):
super().get_context()
lots = self.context['lots']
form_filter = FilterForm()
filter_types = form_filter.search()
lots = self.get_lots()
lot = None
tags = (
Tag.query.filter(Tag.owner_id == current_user.id)
@ -105,21 +90,21 @@ class DeviceListMix(GenericMixView):
if action_devices:
list_devices.extend([int(x) for x in action_devices.split(",")])
self.context = {
'devices': devices,
'lots': lots,
'form_tag_device': TagDeviceForm(),
'form_new_action': form_new_action,
'form_new_allocate': form_new_allocate,
'form_new_datawipe': form_new_datawipe,
'form_new_trade': form_new_trade,
'form_filter': form_filter,
'form_print_labels': PrintLabelsForm(),
'lot': lot,
'tags': tags,
'list_devices': list_devices,
'version': __version__,
}
self.context.update(
{
'devices': devices,
'form_tag_device': TagDeviceForm(),
'form_new_action': form_new_action,
'form_new_allocate': form_new_allocate,
'form_new_datawipe': form_new_datawipe,
'form_new_trade': form_new_trade,
'form_filter': form_filter,
'form_print_labels': PrintLabelsForm(),
'lot': lot,
'tags': tags,
'list_devices': list_devices,
}
)
return self.context
@ -135,20 +120,20 @@ class DeviceDetailView(GenericMixView):
template_name = 'inventory/device_detail.html'
def dispatch_request(self, id):
lots = self.get_lots()
self.get_context()
device = (
Device.query.filter(Device.owner_id == current_user.id)
.filter(Device.devicehub_id == id)
.one()
)
context = {
'device': device,
'lots': lots,
'page_title': 'Device {}'.format(device.devicehub_id),
'version': __version__,
}
return flask.render_template(self.template_name, **context)
self.context.update(
{
'device': device,
'page_title': 'Device {}'.format(device.devicehub_id),
}
)
return flask.render_template(self.template_name, **self.context)
class LotCreateView(GenericMixView):
@ -164,14 +149,14 @@ class LotCreateView(GenericMixView):
next_url = url_for('inventory.lotdevicelist', lot_id=form.id)
return flask.redirect(next_url)
lots = self.get_lots()
context = {
'form': form,
'title': self.title,
'lots': lots,
'version': __version__,
}
return flask.render_template(self.template_name, **context)
self.get_context()
self.context.update(
{
'form': form,
'title': self.title,
}
)
return flask.render_template(self.template_name, **self.context)
class LotUpdateView(View):
@ -187,14 +172,14 @@ class LotUpdateView(View):
next_url = url_for('inventory.lotdevicelist', lot_id=id)
return flask.redirect(next_url)
lots = Lot.query.filter(Lot.owner_id == current_user.id)
context = {
'form': form,
'title': self.title,
'lots': lots,
'version': __version__,
}
return flask.render_template(self.template_name, **context)
self.get_context()
self.context.update(
{
'form': form,
'title': self.title,
}
)
return flask.render_template(self.template_name, **self.context)
class LotDeleteView(View):
@ -221,25 +206,26 @@ class UploadSnapshotView(GenericMixView):
template_name = 'inventory/upload_snapshot.html'
def dispatch_request(self, lot_id=None):
lots = self.get_lots()
self.get_context()
form = UploadSnapshotForm()
context = {
'page_title': 'Upload Snapshot',
'lots': lots,
'form': form,
'lot_id': lot_id,
'version': __version__,
}
self.context.update(
{
'page_title': 'Upload Snapshot',
'form': form,
'lot_id': lot_id,
}
)
if form.validate_on_submit():
snapshot, devices = form.save(commit=False)
if lot_id:
lots = self.context['lots']
lot = lots.filter(Lot.id == lot_id).one()
for dev in devices:
lot.devices.add(dev)
db.session.add(lot)
db.session.commit()
return flask.render_template(self.template_name, **context)
return flask.render_template(self.template_name, **self.context)
class DeviceCreateView(GenericMixView):
@ -248,20 +234,21 @@ class DeviceCreateView(GenericMixView):
template_name = 'inventory/device_create.html'
def dispatch_request(self, lot_id=None):
lots = self.get_lots()
self.get_context()
form = NewDeviceForm()
context = {
'page_title': 'New Device',
'lots': lots,
'form': form,
'lot_id': lot_id,
'version': __version__,
}
self.context.update(
{
'page_title': 'New Device',
'form': form,
'lot_id': lot_id,
}
)
if form.validate_on_submit():
snapshot = form.save(commit=False)
next_url = url_for('inventory.devicelist')
if lot_id:
next_url = url_for('inventory.lotdevicelist', lot_id=lot_id)
lots = self.context['lots']
lot = lots.filter(Lot.id == lot_id).one()
lot.devices.add(snapshot.device)
db.session.add(lot)
@ -270,7 +257,7 @@ class DeviceCreateView(GenericMixView):
messages.success('Device "{}" created successfully!'.format(form.type.data))
return flask.redirect(next_url)
return flask.render_template(self.template_name, **context)
return flask.render_template(self.template_name, **self.context)
class TagLinkDeviceView(View):
@ -286,13 +273,13 @@ class TagLinkDeviceView(View):
return flask.redirect(request.referrer)
class TagUnlinkDeviceView(View):
class TagUnlinkDeviceView(GenericMixView):
methods = ['POST', 'GET']
decorators = [login_required]
template_name = 'inventory/tag_unlink_device.html'
def dispatch_request(self, id):
lots = Lot.query.filter(Lot.owner_id == current_user.id)
self.get_context()
form = TagDeviceForm(delete=True, device=id)
if form.validate_on_submit():
form.remove()
@ -300,14 +287,15 @@ class TagUnlinkDeviceView(View):
next_url = url_for('inventory.devicelist')
return flask.redirect(next_url)
return flask.render_template(
self.template_name,
form=form,
lots=lots,
referrer=request.referrer,
version=__version__,
self.context.update(
{
'form': form,
'referrer': request.referrer,
}
)
return flask.render_template(self.template_name, **self.context)
class NewActionView(View):
methods = ['POST']
@ -316,16 +304,19 @@ class NewActionView(View):
def dispatch_request(self):
self.form = self.form_class()
next_url = self.get_next_url()
if self.form.validate_on_submit():
self.form.save()
messages.success(
'Action "{}" created successfully!'.format(self.form.type.data)
)
next_url = self.get_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):
lot_id = self.form.lot.data
@ -351,10 +342,12 @@ class NewAllocateView(NewActionView, DeviceListMix):
next_url = self.get_next_url()
return flask.redirect(next_url)
lot_id = self.form.lot.data
self.get_context(lot_id)
self.context['form_new_allocate'] = self.form
return flask.render_template(self.template_name, **self.context)
messages.error('Action {} error!'.format(self.form.type.data))
for k, v in self.form.errors.items():
value = ';'.join(v)
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):
@ -373,10 +366,9 @@ class NewDataWipeView(NewActionView, DeviceListMix):
next_url = self.get_next_url()
return flask.redirect(next_url)
lot_id = self.form.lot.data
self.get_context(lot_id)
self.context['form_new_datawipe'] = self.form
return flask.render_template(self.template_name, **self.context)
messages.error('Action {} error!'.format(self.form.type.data))
next_url = self.get_next_url()
return flask.redirect(next_url)
class NewTradeView(NewActionView, DeviceListMix):
@ -395,10 +387,9 @@ class NewTradeView(NewActionView, DeviceListMix):
next_url = self.get_next_url()
return flask.redirect(next_url)
lot_id = self.form.lot.data
self.get_context(lot_id)
self.context['form_new_trade'] = self.form
return flask.render_template(self.template_name, **self.context)
messages.error('Action {} error!'.format(self.form.type.data))
next_url = self.get_next_url()
return flask.redirect(next_url)
class NewTradeDocumentView(View):
@ -410,6 +401,7 @@ class NewTradeDocumentView(View):
def dispatch_request(self, lot_id):
self.form = self.form_class(lot=lot_id)
self.get_context()
if self.form.validate_on_submit():
self.form.save()
@ -417,9 +409,8 @@ class NewTradeDocumentView(View):
next_url = url_for('inventory.lotdevicelist', lot_id=lot_id)
return flask.redirect(next_url)
return flask.render_template(
self.template_name, form=self.form, title=self.title, version=__version__
)
self.context.update({'form': self.form, 'title': self.title})
return flask.render_template(self.template_name, **self.context)
class ExportsView(View):

46
ereuse_devicehub/views.py
View File

@ -1,11 +1,14 @@
import flask
from flask import Blueprint
from flask import Blueprint, g
from flask.views import View
from flask_login import current_user, login_required, login_user, logout_user
from sqlalchemy import or_
from ereuse_devicehub import __version__, messages
from ereuse_devicehub.db import db
from ereuse_devicehub.forms import LoginForm, PasswordForm
from ereuse_devicehub.resources.action.models import Trade
from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.user.models import User
from ereuse_devicehub.utils import is_safe_url
@ -46,18 +49,45 @@ class LogoutView(View):
return flask.redirect(flask.url_for('core.login'))
class UserProfileView(View):
class GenericMixView(View):
decorators = [login_required]
def get_lots(self):
return (
Lot.query.outerjoin(Trade)
.filter(
or_(
Trade.user_from == g.user,
Trade.user_to == g.user,
Lot.owner_id == g.user.id,
)
)
.distinct()
)
def get_context(self):
self.context = {
'lots': self.get_lots(),
'version': __version__,
}
return self.context
class UserProfileView(GenericMixView):
decorators = [login_required]
template_name = 'ereuse_devicehub/user_profile.html'
def dispatch_request(self):
context = {
'current_user': current_user,
'version': __version__,
'password_form': PasswordForm(),
}
self.get_context()
self.context.update(
{
'current_user': current_user,
'password_form': PasswordForm(),
}
)
return flask.render_template(self.template_name, **context)
return flask.render_template(self.template_name, **self.context)
class UserPasswordView(View):

90
tests/conftest.py
View File

@ -1,29 +1,31 @@
import io
import json
import uuid
import jwt
import ereuse_utils
from contextlib import redirect_stdout
from datetime import datetime
from pathlib import Path
from decouple import config
import boltons.urlutils
import ereuse_utils
import jwt
import pytest
import yaml
from decouple import config
from psycopg2 import IntegrityError
from sqlalchemy.exc import ProgrammingError
from ereuse_devicehub.client import Client, UserClient
from ereuse_devicehub.api.views import api
from ereuse_devicehub.client import Client, UserClient, UserClientFlask
from ereuse_devicehub.config import DevicehubConfig
from ereuse_devicehub.db import db
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.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.api.views import api
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)
"""A dummy starting time to use in tests."""
@ -52,6 +54,21 @@ def config():
@pytest.fixture(scope='session')
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.register_blueprint(api)
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)
@ -63,14 +80,15 @@ def app(request, _app: Devicehub) -> Devicehub:
db.drop_all()
def _init():
_app.init_db(name='Test Inventory',
org_name='FooOrg',
org_id='foo-org-id',
tag_url=boltons.urlutils.URL('https://example.com'),
tag_token=uuid.UUID('52dacef0-6bcb-4919-bfed-f10d2c96ecee'),
erase=False,
common=True)
_app.register_blueprint(api)
_app.init_db(
name='Test Inventory',
org_name='FooOrg',
org_id='foo-org-id',
tag_url=boltons.urlutils.URL('https://example.com'),
tag_token=uuid.UUID('52dacef0-6bcb-4919-bfed-f10d2c96ecee'),
erase=False,
common=True,
)
with _app.app_context():
try:
@ -102,7 +120,9 @@ def user(app: Devicehub) -> UserClient:
with app.app_context():
password = 'foo'
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()
return client
@ -114,11 +134,34 @@ def user2(app: Devicehub) -> UserClient:
password = 'foo'
email = 'foo2@foo.com'
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()
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:
user = User(email=email, password=password)
user.individuals.add(Person(name='Timmy'))
@ -148,16 +191,13 @@ def auth_app_context(app: Devicehub):
def json_encode(dev: str) -> dict:
"""Encode json."""
data = {"type": "Snapshot"}
data['data'] = jwt.encode(dev,
P,
algorithm="HS256",
json_encoder=ereuse_utils.JSONEncoder
data['data'] = jwt.encode(
dev, P, algorithm="HS256", json_encoder=ereuse_utils.JSONEncoder
)
return data
def yaml2json(name: str) -> dict:
"""Opens and parses a YAML file from the ``files`` subdir."""
with Path(__file__).parent.joinpath('files').joinpath(name + '.yaml').open() as f:
@ -176,7 +216,9 @@ def file_json(name):
def file_workbench(name: str) -> dict:
"""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)

2
tests/files/export_devices.csv Normal file
View File

File diff suppressed because one or more lines are too long

52
tests/test_basic.py
View File

@ -1,7 +1,7 @@
import pytest
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.client import Client
from ereuse_devicehub.devicehub import Devicehub
@pytest.mark.mvp
@ -28,38 +28,66 @@ def test_api_docs(client: Client):
"""Tests /apidocs correct initialization."""
docs, _ = client.get('/apidocs')
assert set(docs['paths'].keys()) == {
'/',
'/actions/',
'/allocates/',
'/apidocs',
'/api/inventory/',
'/allocates/',
'/deallocates/',
'/deliverynotes/',
'/devices/',
'/devices/static/{filename}',
'/documents/static/{filename}',
'/documents/actions/',
'/documents/erasures/',
'/documents/devices/',
'/documents/stamps/',
'/documents/wbconf/{wbtype}',
'/documents/internalstats/',
'/documents/stock/',
'/documents/check/',
'/documents/devices/',
'/documents/erasures/',
'/documents/internalstats/',
'/documents/lots/',
'/versions/',
'/manufacturers/',
'/documents/stamps/',
'/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/',
'/lives/',
'/login/',
'/logout/',
'/lots/',
'/lots/{id}/children',
'/lots/{id}/devices',
'/manufacturers/',
'/metrics/',
'/profile/',
'/set_password/',
'/tags/',
'/tags/{tag_id}/device/{device_id}',
'/trade-documents/',
'/users/',
'/users/login/',
'/users/logout/',
'/versions/',
}
assert docs['info'] == {'title': 'Devicehub', 'version': '0.2'}
assert docs['components']['securitySchemes']['bearerAuth'] == {
@ -68,6 +96,6 @@ def test_api_docs(client: Client):
'description:': 'HTTP Basic scheme',
'type': 'http',
'scheme': 'basic',
'name': 'Authorization'
'name': 'Authorization',
}
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