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

Link tags to devices through PUT; update client

This commit is contained in:
Xavier Bustamante Talavera 2018-09-20 11:51:25 +02:00
parent 3b0f483a90
commit 32837f5f59
6 changed files with 135 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
ereuse_devicehub/client.py
View File

@ -1,14 +1,14 @@
from inspect import isclass
from typing import Any, Dict, Iterable, Tuple, Type, Union
from typing import Dict, Iterable, Type, Union
from ereuse_utils.test import JSON
from flask import Response
from teal.client import Client as TealClient
from teal.marshmallow import ValidationError
from ereuse_utils.test import JSON, Res
from teal.client import Client as TealClient, Query, Status
from werkzeug.exceptions import HTTPException
from ereuse_devicehub.resources import models, schemas
ResourceLike = Union[Type[Union[models.Thing, schemas.Thing]], str]
class Client(TealClient):
"""A client suited for Devicehub main usage."""
@ -21,15 +21,15 @@ class Client(TealClient):
def open(self,
uri: str,
res: Union[str, Type[Union[models.Thing, schemas.Thing]]] = None,
status: Union[int, Type[HTTPException], Type[ValidationError]] = 200,
query: Iterable[Tuple[str, Any]] = tuple(),
res: ResourceLike = None,
status: Status = 200,
query: Query = tuple(),
accept=JSON,
content_type=JSON,
item=None,
headers: dict = None,
token: str = None,
**kw) -> Tuple[Union[Dict[str, object], str], Response]:
**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,
@ -37,37 +37,79 @@ class Client(TealClient):
def get(self,
uri: str = '',
res: Union[Type[Union[models.Thing, schemas.Thing]], str] = None,
query: Iterable[Tuple[str, Any]] = tuple(),
status: Union[int, Type[HTTPException], Type[ValidationError]] = 200,
res: ResourceLike = None,
query: Query = tuple(),
status: Status = 200,
item: Union[int, str] = None,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw) -> Tuple[Union[Dict[str, object], str], Response]:
**kw) -> Res:
return super().get(uri, res, query, status, item, accept, headers, token, **kw)
def post(self,
data: str or dict,
uri: str = '',
res: Union[Type[Union[models.Thing, schemas.Thing]], str] = None,
query: Iterable[Tuple[str, Any]] = tuple(),
status: Union[int, Type[HTTPException], Type[ValidationError]] = 201,
res: ResourceLike = None,
query: Query = tuple(),
status: Status = 201,
content_type: str = JSON,
accept: str = JSON,
headers: dict = None,
token: str = None,
**kw) -> Tuple[Union[Dict[str, object], str], Response]:
**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 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 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)
def get_many(self,
res: Union[Type[Union[models.Thing, schemas.Thing]], str],
res: ResourceLike,
resources: Iterable[Union[dict, int]],
key: str = None,
**kw) -> Iterable[Union[Dict[str, object], str]]:
@ -98,18 +140,19 @@ class UserClient(Client):
def open(self,
uri: str,
res: Union[str, Type[Union[models.Thing, schemas.Thing]]] = None,
res: ResourceLike = None,
status: int or HTTPException = 200,
query: Iterable[Tuple[str, Any]] = tuple(),
query: Query = tuple(),
accept=JSON,
content_type=JSON,
item=None,
headers: dict = None,
token: str = None,
**kw) -> Tuple[Union[Dict[str, object], str], Response]:
**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]

17
ereuse_devicehub/resources/tag/__init__.py
View File

@ -7,9 +7,10 @@ from teal.resource import Resource
from teal.teal import Teal
from ereuse_devicehub.db import db
from ereuse_devicehub.resources.device import DeviceDef
from ereuse_devicehub.resources.tag import schema
from ereuse_devicehub.resources.tag.model import Tag
from ereuse_devicehub.resources.tag.view import TagView, get_device_from_tag
from ereuse_devicehub.resources.tag.view import TagDeviceView, TagView, get_device_from_tag
class TagDef(Resource):
@ -33,10 +34,18 @@ class TagDef(Resource):
super().__init__(app, import_name, static_folder, static_url_path, template_folder,
url_prefix, subdomain, url_defaults, root_path, cli_commands)
_get_device_from_tag = app.auth.requires_auth(get_device_from_tag)
self.add_url_rule('/<{}:{}>/device'.format(self.ID_CONVERTER.value, self.ID_NAME),
view_func=_get_device_from_tag,
# DeviceTagView URLs
device_view = TagDeviceView.as_view('tag-device-view', definition=self, auth=app.auth)
if self.AUTH:
device_view = app.auth.requires_auth(device_view)
self.add_url_rule('/<{0.ID_CONVERTER.value}:{0.ID_NAME}>/device'.format(self),
view_func=device_view,
methods={'GET'})
self.tag_schema = schema.Tag
self.add_url_rule('/<{0.ID_CONVERTER.value}:tag_id>/'.format(self) +
'device/<{0.ID_CONVERTER.value}:device_id>'.format(DeviceDef),
view_func=device_view,
methods={'PUT'})
@option('-o', '--org', help=ORG_H)
@option('-p', '--provider', help=PROV_H)

32
ereuse_devicehub/resources/tag/view.py
View File

@ -8,7 +8,6 @@ from ereuse_devicehub.resources.tag import Tag
class TagView(View):
def post(self):
"""Creates a tag."""
t = request.get_json()
@ -20,6 +19,31 @@ class TagView(View):
return Response(status=201)
class TagDeviceView(View):
"""Endpoints to work with the device of the tag; /tags/23/device"""
def one(self, id):
"""Gets the device from the tag."""
tag = Tag.from_an_id(id).one() # type: Tag
if not tag.device:
raise TagNotLinked(tag.id)
return app.resources[Device.t].schema.jsonify(tag.device)
# noinspection PyMethodOverriding
def put(self, tag_id: str, device_id: str):
"""Links an existing tag with a device."""
tag = Tag.from_an_id(tag_id).one() # type: Tag
if tag.device_id:
if tag.device_id == device_id:
return Response(status=204)
else:
raise LinkedToAnotherDevice(tag.device_id)
else:
tag.device_id = device_id
db.session.commit()
return Response(status=204)
def get_device_from_tag(id: str):
"""
Gets the device by passing a tag id.
@ -46,3 +70,9 @@ class CannotCreateETag(ValidationError):
def __init__(self, id: str):
message = 'Only sysadmin can create an eReuse.org Tag ({})'.format(id)
super().__init__(message)
class LinkedToAnotherDevice(ValidationError):
def __init__(self, device_id: int):
message = 'The tag is already linked to device {}'.format(device_id)
super().__init__(message)

2
setup.py
View File

@ -34,7 +34,7 @@ setup(
long_description=long_description,
long_description_content_type='text/markdown',
install_requires=[
'teal>=0.2.0a15', # teal always first
'teal>=0.2.0a16', # teal always first
'click',
'click-spinner',
'ereuse-rate==0.0.2',

5
tests/test_basic.py
View File

@ -16,7 +16,7 @@ def test_api_docs(client: Client):
"""Tests /apidocs correct initialization."""
docs, _ = client.get('/apidocs')
assert set(docs['paths'].keys()) == {
'/tags/{id}/device',
# todo this does not appear: '/tags/{id}/device',
'/inventories/',
'/apidocs',
'/users/',
@ -27,7 +27,8 @@ def test_api_docs(client: Client):
'/events/',
'/lots/',
'/lots/{id}/children',
'/lots/{id}/devices'
'/lots/{id}/devices',
'/tags/{tag_id}/device/{device_id}'
}
assert docs['info'] == {'title': 'Devicehub', 'version': '0.2'}
assert docs['components']['securitySchemes']['bearerAuth'] == {

25
tests/test_tag.py
View File

@ -9,10 +9,11 @@ from ereuse_devicehub.client import UserClient
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.resources.agent.models import Organization
from ereuse_devicehub.resources.device.models import Desktop
from ereuse_devicehub.resources.device.models import Desktop, Device
from ereuse_devicehub.resources.enums import ComputerChassis
from ereuse_devicehub.resources.tag import Tag
from ereuse_devicehub.resources.tag.view import CannotCreateETag, TagNotLinked
from ereuse_devicehub.resources.tag.view import CannotCreateETag, LinkedToAnotherDevice, \
TagNotLinked
from tests import conftest
@ -135,6 +136,26 @@ def test_tag_create_tags_cli(app: Devicehub, user: UserClient):
assert tag.org.id == Organization.get_default_org_id()
def test_tag_manual_link(app: Devicehub, user: UserClient):
"""Tests linking manually a tag through PUT /tags/<id>/device/<id>"""
with app.app_context():
db.session.add(Tag('foo-bar', secondary='foo-sec'))
desktop = Desktop(serial_number='foo', chassis=ComputerChassis.AllInOne)
db.session.add(desktop)
db.session.commit()
desktop_id = desktop.id
user.put({}, res=Tag, item='foo-bar/device/{}'.format(desktop_id), status=204)
device, _ = user.get(res=Device, item=1)
assert device['tags'][0]['id'] == 'foo-bar'
# Device already linked
# Just returns an OK to conform to PUT as anything changes
user.put({}, res=Tag, item='foo-sec/device/{}'.format(desktop_id), status=204)
# cannot link to another device when already linked
user.put({}, res=Tag, item='foo-bar/device/99', status=LinkedToAnotherDevice)
@pytest.mark.usefixtures(conftest.app_context.__name__)
def test_tag_secondary():
"""Creates and consumes tags with a secondary id."""