Merge pull request #127 from eReuse/feature/#126-named-system-tags
Feature/#126 named system tags
This commit is contained in:
commit
a2da7f37d9
|
@ -0,0 +1,62 @@
|
||||||
|
"""drop unique org for tag
|
||||||
|
|
||||||
|
Revision ID: 6a2a939d5668
|
||||||
|
Revises: eca457d8b2a4
|
||||||
|
Create Date: 2021-02-25 18:47:47.441195
|
||||||
|
|
||||||
|
"""
|
||||||
|
from alembic import op
|
||||||
|
import sqlalchemy as sa
|
||||||
|
from alembic import context
|
||||||
|
|
||||||
|
|
||||||
|
# revision identifiers, used by Alembic.
|
||||||
|
revision = '6a2a939d5668'
|
||||||
|
down_revision = 'eca457d8b2a4'
|
||||||
|
branch_labels = None
|
||||||
|
depends_on = None
|
||||||
|
|
||||||
|
|
||||||
|
def get_inv():
|
||||||
|
INV = context.get_x_argument(as_dictionary=True).get('inventory')
|
||||||
|
if not INV:
|
||||||
|
raise ValueError("Inventory value is not specified")
|
||||||
|
return INV
|
||||||
|
|
||||||
|
|
||||||
|
def upgrade_data():
|
||||||
|
con = op.get_bind()
|
||||||
|
tags = con.execute(f"select id from {get_inv()}.tag")
|
||||||
|
i = 1
|
||||||
|
for c in tags:
|
||||||
|
id_tag = c.id
|
||||||
|
internal_id = i
|
||||||
|
i += 1
|
||||||
|
sql = f"update {get_inv()}.tag set internal_id='{internal_id}' where id='{id_tag}';"
|
||||||
|
con.execute(sql)
|
||||||
|
|
||||||
|
sql = f"CREATE SEQUENCE {get_inv()}.tag_internal_id_seq START {i};"
|
||||||
|
con.execute(sql)
|
||||||
|
|
||||||
|
|
||||||
|
def upgrade():
|
||||||
|
op.drop_constraint('one tag id per organization', 'tag', schema=f'{get_inv()}')
|
||||||
|
op.drop_constraint('one secondary tag per organization', 'tag', schema=f'{get_inv()}')
|
||||||
|
op.create_primary_key('one tag id per owner', 'tag', ['id', 'owner_id'], schema=f'{get_inv()}'),
|
||||||
|
op.create_unique_constraint('one secondary tag per owner', 'tag', ['secondary', 'owner_id'], schema=f'{get_inv()}'),
|
||||||
|
op.add_column('tag', sa.Column('internal_id', sa.BigInteger(), nullable=True,
|
||||||
|
comment='The identifier of the tag for this database. Used only\n internally for software; users should not use this.\n'), schema=f'{get_inv()}')
|
||||||
|
|
||||||
|
upgrade_data()
|
||||||
|
|
||||||
|
op.alter_column('tag', sa.Column('internal_id', sa.BigInteger(), nullable=False,
|
||||||
|
comment='The identifier of the tag for this database. Used only\n internally for software; users should not use this.\n'), schema=f'{get_inv()}')
|
||||||
|
|
||||||
|
|
||||||
|
def downgrade():
|
||||||
|
op.drop_constraint('one tag id per owner', 'tag', schema=f'{get_inv()}')
|
||||||
|
op.drop_constraint('one secondary tag per owner', 'tag', schema=f'{get_inv()}')
|
||||||
|
op.create_primary_key('one tag id per organization', 'tag', ['id', 'org_id'], schema=f'{get_inv()}'),
|
||||||
|
op.create_unique_constraint('one secondary tag per organization', 'tag', ['secondary', 'org_id'], schema=f'{get_inv()}'),
|
||||||
|
op.drop_column('tag', 'internal_id', schema=f'{get_inv()}')
|
||||||
|
op.execute(f"DROP SEQUENCE {get_inv()}.tag_internal_id_seq;")
|
|
@ -250,6 +250,11 @@ class MakeAvailable(ActionDef):
|
||||||
SCHEMA = schemas.MakeAvailable
|
SCHEMA = schemas.MakeAvailable
|
||||||
|
|
||||||
|
|
||||||
|
class TradeDef(ActionDef):
|
||||||
|
VIEW = None
|
||||||
|
SCHEMA = schemas.Trade
|
||||||
|
|
||||||
|
|
||||||
class CancelTradeDef(ActionDef):
|
class CancelTradeDef(ActionDef):
|
||||||
VIEW = None
|
VIEW = None
|
||||||
SCHEMA = schemas.CancelTrade
|
SCHEMA = schemas.CancelTrade
|
||||||
|
|
|
@ -50,8 +50,7 @@ class DeviceRow(OrderedDict):
|
||||||
self['Tag 2 Type'] = self['Tag 2 ID'] = self['Tag 2 Organization'] = ''
|
self['Tag 2 Type'] = self['Tag 2 ID'] = self['Tag 2 Organization'] = ''
|
||||||
self['Tag 3 Type'] = self['Tag 3 ID'] = self['Tag 3 Organization'] = ''
|
self['Tag 3 Type'] = self['Tag 3 ID'] = self['Tag 3 Organization'] = ''
|
||||||
for i, tag in zip(range(1, 3), device.tags):
|
for i, tag in zip(range(1, 3), device.tags):
|
||||||
# TODO @cayop we need redefined how save the Tag Type info
|
self['Tag {} Type'.format(i)] = 'unamed' if tag.provider else 'named'
|
||||||
self['Tag {} Type'.format(i)] = 'unamed'
|
|
||||||
self['Tag {} ID'.format(i)] = tag.id
|
self['Tag {} ID'.format(i)] = tag.id
|
||||||
self['Tag {} Organization'.format(i)] = tag.org.name
|
self['Tag {} Organization'.format(i)] = tag.org.name
|
||||||
|
|
||||||
|
|
|
@ -48,6 +48,10 @@ class TagDef(Resource):
|
||||||
'device/<{0.ID_CONVERTER.value}:device_id>'.format(DeviceDef),
|
'device/<{0.ID_CONVERTER.value}:device_id>'.format(DeviceDef),
|
||||||
view_func=device_view,
|
view_func=device_view,
|
||||||
methods={'PUT'})
|
methods={'PUT'})
|
||||||
|
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={'DELETE'})
|
||||||
|
|
||||||
@option('-u', '--owner', help=OWNER_H)
|
@option('-u', '--owner', help=OWNER_H)
|
||||||
@option('-o', '--org', help=ORG_H)
|
@option('-o', '--org', help=ORG_H)
|
||||||
|
|
|
@ -3,7 +3,7 @@ from typing import Set
|
||||||
|
|
||||||
from boltons import urlutils
|
from boltons import urlutils
|
||||||
from flask import g
|
from flask import g
|
||||||
from sqlalchemy import BigInteger, Column, ForeignKey, UniqueConstraint
|
from sqlalchemy import BigInteger, Column, ForeignKey, UniqueConstraint, Sequence
|
||||||
from sqlalchemy.dialects.postgresql import UUID
|
from sqlalchemy.dialects.postgresql import UUID
|
||||||
from sqlalchemy.orm import backref, relationship, validates
|
from sqlalchemy.orm import backref, relationship, validates
|
||||||
from teal.db import DB_CASCADE_SET_NULL, Query, URL
|
from teal.db import DB_CASCADE_SET_NULL, Query, URL
|
||||||
|
@ -15,6 +15,7 @@ from ereuse_devicehub.resources.agent.models import Organization
|
||||||
from ereuse_devicehub.resources.device.models import Device
|
from ereuse_devicehub.resources.device.models import Device
|
||||||
from ereuse_devicehub.resources.models import Thing
|
from ereuse_devicehub.resources.models import Thing
|
||||||
from ereuse_devicehub.resources.user.models import User
|
from ereuse_devicehub.resources.user.models import User
|
||||||
|
from ereuse_devicehub.resources.utils import hashcode
|
||||||
|
|
||||||
|
|
||||||
class Tags(Set['Tag']):
|
class Tags(Set['Tag']):
|
||||||
|
@ -25,17 +26,23 @@ class Tags(Set['Tag']):
|
||||||
return ', '.join(format(tag, format_spec) for tag in self).strip()
|
return ', '.join(format(tag, format_spec) for tag in self).strip()
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
class Tag(Thing):
|
class Tag(Thing):
|
||||||
|
internal_id = Column(BigInteger, Sequence('tag_internal_id_seq'), unique=True, nullable=False)
|
||||||
|
internal_id.comment = """The identifier of the tag for this database. Used only
|
||||||
|
internally for software; users should not use this.
|
||||||
|
"""
|
||||||
id = Column(db.CIText(), primary_key=True)
|
id = Column(db.CIText(), primary_key=True)
|
||||||
id.comment = """The ID of the tag."""
|
id.comment = """The ID of the tag."""
|
||||||
owner_id = Column(UUID(as_uuid=True),
|
owner_id = Column(UUID(as_uuid=True),
|
||||||
ForeignKey(User.id),
|
ForeignKey(User.id),
|
||||||
|
primary_key=True,
|
||||||
nullable=False,
|
nullable=False,
|
||||||
default=lambda: g.user.id)
|
default=lambda: g.user.id)
|
||||||
owner = relationship(User, primaryjoin=owner_id == User.id)
|
owner = relationship(User, primaryjoin=owner_id == User.id)
|
||||||
org_id = Column(UUID(as_uuid=True),
|
org_id = Column(UUID(as_uuid=True),
|
||||||
ForeignKey(Organization.id),
|
ForeignKey(Organization.id),
|
||||||
primary_key=True,
|
|
||||||
# If we link with the Organization object this instance
|
# If we link with the Organization object this instance
|
||||||
# will be set as persistent and added to session
|
# will be set as persistent and added to session
|
||||||
# which is something we don't want to enforce by default
|
# which is something we don't want to enforce by default
|
||||||
|
@ -97,8 +104,8 @@ class Tag(Thing):
|
||||||
return url
|
return url
|
||||||
|
|
||||||
__table_args__ = (
|
__table_args__ = (
|
||||||
UniqueConstraint(id, org_id, name='one tag id per organization'),
|
UniqueConstraint(id, owner_id, name='one tag id per owner'),
|
||||||
UniqueConstraint(secondary, org_id, name='one secondary tag per organization')
|
UniqueConstraint(secondary, owner_id, name='one secondary tag per organization')
|
||||||
)
|
)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
|
@ -109,7 +116,7 @@ class Tag(Thing):
|
||||||
def url(self) -> urlutils.URL:
|
def url(self) -> urlutils.URL:
|
||||||
"""The URL where to GET this device."""
|
"""The URL where to GET this device."""
|
||||||
# todo this url only works for printable internal tags
|
# todo this url only works for printable internal tags
|
||||||
return urlutils.URL(url_for_resource(Tag, item_id=self.id))
|
return urlutils.URL(url_for_resource(Tag, item_id=self.code))
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def printable(self) -> bool:
|
def printable(self) -> bool:
|
||||||
|
@ -125,6 +132,23 @@ class Tag(Thing):
|
||||||
"""Return a SQLAlchemy filter expression for printable queries."""
|
"""Return a SQLAlchemy filter expression for printable queries."""
|
||||||
return cls.org_id == Organization.get_default_org_id()
|
return cls.org_id == Organization.get_default_org_id()
|
||||||
|
|
||||||
|
@property
|
||||||
|
def code(self) -> str:
|
||||||
|
return hashcode.encode(self.internal_id)
|
||||||
|
|
||||||
|
def delete(self):
|
||||||
|
"""Deletes the tag.
|
||||||
|
|
||||||
|
This method removes the tag if is named tag and don't have any linked device.
|
||||||
|
"""
|
||||||
|
if self.device:
|
||||||
|
raise TagLinked(self)
|
||||||
|
if self.provider:
|
||||||
|
# if is an unnamed tag not delete
|
||||||
|
raise TagUnnamed(self.id)
|
||||||
|
|
||||||
|
db.session.delete(self)
|
||||||
|
|
||||||
def __repr__(self) -> str:
|
def __repr__(self) -> str:
|
||||||
return '<Tag {0.id} org:{0.org_id} device:{0.device_id}>'.format(self)
|
return '<Tag {0.id} org:{0.org_id} device:{0.device_id}>'.format(self)
|
||||||
|
|
||||||
|
@ -133,3 +157,15 @@ class Tag(Thing):
|
||||||
|
|
||||||
def __format__(self, format_spec: str) -> str:
|
def __format__(self, format_spec: str) -> str:
|
||||||
return '{0.org.name} {0.id}'.format(self)
|
return '{0.org.name} {0.id}'.format(self)
|
||||||
|
|
||||||
|
|
||||||
|
class TagLinked(ValidationError):
|
||||||
|
def __init__(self, tag):
|
||||||
|
message = 'The tag {} is linked to device {}.'.format(tag.id, tag.device.id)
|
||||||
|
super().__init__(message, field_names=['device'])
|
||||||
|
|
||||||
|
|
||||||
|
class TagUnnamed(ValidationError):
|
||||||
|
def __init__(self, id):
|
||||||
|
message = 'This tag {} is unnamed tag. It is imposible delete.'.format(id)
|
||||||
|
super().__init__(message, field_names=['device'])
|
||||||
|
|
|
@ -28,3 +28,4 @@ class Tag(Thing):
|
||||||
secondary = SanitizedStr(lower=True, description=m.Tag.secondary.comment)
|
secondary = SanitizedStr(lower=True, description=m.Tag.secondary.comment)
|
||||||
printable = Boolean(dump_only=True, decsription=m.Tag.printable.__doc__)
|
printable = Boolean(dump_only=True, decsription=m.Tag.printable.__doc__)
|
||||||
url = URL(dump_only=True, description=m.Tag.url.__doc__)
|
url = URL(dump_only=True, description=m.Tag.url.__doc__)
|
||||||
|
code = SanitizedStr(dump_only=True, description=m.Tag.internal_id.comment)
|
||||||
|
|
|
@ -6,18 +6,29 @@ from teal.resource import View, url_for_resource
|
||||||
from ereuse_devicehub import auth
|
from ereuse_devicehub import auth
|
||||||
from ereuse_devicehub.db import db
|
from ereuse_devicehub.db import db
|
||||||
from ereuse_devicehub.query import things_response
|
from ereuse_devicehub.query import things_response
|
||||||
|
from ereuse_devicehub.resources.utils import hashcode
|
||||||
from ereuse_devicehub.resources.device.models import Device
|
from ereuse_devicehub.resources.device.models import Device
|
||||||
from ereuse_devicehub.resources.tag import Tag
|
from ereuse_devicehub.resources.tag import Tag
|
||||||
|
|
||||||
|
|
||||||
class TagView(View):
|
class TagView(View):
|
||||||
|
def one(self, code):
|
||||||
|
"""Gets the device from the named tag, /tags/namedtag."""
|
||||||
|
internal_id = hashcode.decode(code.upper()) or -1
|
||||||
|
tag = Tag.query.filter_by(internal_id=internal_id).one() # type: Tag
|
||||||
|
if not tag.device:
|
||||||
|
raise TagNotLinked(tag.id)
|
||||||
|
return redirect(location=url_for_resource(Device, tag.device.id))
|
||||||
|
|
||||||
@auth.Auth.requires_auth
|
@auth.Auth.requires_auth
|
||||||
def post(self):
|
def post(self):
|
||||||
"""Creates a tag."""
|
"""Creates a tag."""
|
||||||
num = request.args.get('num', type=int)
|
num = request.args.get('num', type=int)
|
||||||
if num:
|
if num:
|
||||||
|
# create unnamed tag
|
||||||
res = self._create_many_regular_tags(num)
|
res = self._create_many_regular_tags(num)
|
||||||
else:
|
else:
|
||||||
|
# create named tag
|
||||||
res = self._post_one()
|
res = self._post_one()
|
||||||
return res
|
return res
|
||||||
|
|
||||||
|
@ -42,7 +53,6 @@ class TagView(View):
|
||||||
return response
|
return response
|
||||||
|
|
||||||
def _post_one(self):
|
def _post_one(self):
|
||||||
# todo do we use this?
|
|
||||||
t = request.get_json()
|
t = request.get_json()
|
||||||
tag = Tag(**t)
|
tag = Tag(**t)
|
||||||
if tag.like_etag():
|
if tag.like_etag():
|
||||||
|
@ -52,34 +62,69 @@ class TagView(View):
|
||||||
db.session.commit()
|
db.session.commit()
|
||||||
return Response(status=201)
|
return Response(status=201)
|
||||||
|
|
||||||
|
@auth.Auth.requires_auth
|
||||||
|
def delete(self, id):
|
||||||
|
tag = Tag.from_an_id(id).filter_by(owner=g.user).one()
|
||||||
|
tag.delete()
|
||||||
|
db.session().final_flush()
|
||||||
|
db.session.commit()
|
||||||
|
return Response(status=204)
|
||||||
|
|
||||||
|
|
||||||
class TagDeviceView(View):
|
class TagDeviceView(View):
|
||||||
"""Endpoints to work with the device of the tag; /tags/23/device."""
|
"""Endpoints to work with the device of the tag; /tags/23/device."""
|
||||||
|
|
||||||
def one(self, id):
|
def one(self, id):
|
||||||
"""Gets the device from the tag."""
|
"""Gets the device from the tag."""
|
||||||
|
if request.authorization:
|
||||||
|
return self.one_authorization(id)
|
||||||
|
|
||||||
tag = Tag.from_an_id(id).one() # type: Tag
|
tag = Tag.from_an_id(id).one() # type: Tag
|
||||||
if not tag.device:
|
if not tag.device:
|
||||||
raise TagNotLinked(tag.id)
|
raise TagNotLinked(tag.id)
|
||||||
if not request.authorization:
|
return redirect(location=url_for_resource(Device, tag.device.id))
|
||||||
return redirect(location=url_for_resource(Device, tag.device.id))
|
|
||||||
|
@auth.Auth.requires_auth
|
||||||
|
def one_authorization(self, id):
|
||||||
|
tag = Tag.from_an_id(id).filter_by(owner=g.user).one() # type: Tag
|
||||||
|
if not tag.device:
|
||||||
|
raise TagNotLinked(tag.id)
|
||||||
return app.resources[Device.t].schema.jsonify(tag.device)
|
return app.resources[Device.t].schema.jsonify(tag.device)
|
||||||
|
|
||||||
# noinspection PyMethodOverriding
|
# noinspection PyMethodOverriding
|
||||||
|
@auth.Auth.requires_auth
|
||||||
def put(self, tag_id: str, device_id: str):
|
def put(self, tag_id: str, device_id: str):
|
||||||
"""Links an existing tag with a device."""
|
"""Links an existing tag with a device."""
|
||||||
tag = Tag.from_an_id(tag_id).one() # type: Tag
|
# tag = Tag.from_an_id(tag_id).one() # type: Tag
|
||||||
|
tag = Tag.from_an_id(tag_id).filter_by(owner=g.user).one() # type: Tag
|
||||||
if tag.device_id:
|
if tag.device_id:
|
||||||
if tag.device_id == device_id:
|
if tag.device_id == device_id:
|
||||||
return Response(status=204)
|
return Response(status=204)
|
||||||
else:
|
else:
|
||||||
raise LinkedToAnotherDevice(tag.device_id)
|
raise LinkedToAnotherDevice(tag.device_id)
|
||||||
else:
|
else:
|
||||||
|
# Check if this device exist for this owner
|
||||||
|
Device.query.filter_by(owner=g.user).filter_by(id=device_id).one()
|
||||||
tag.device_id = device_id
|
tag.device_id = device_id
|
||||||
|
|
||||||
db.session().final_flush()
|
db.session().final_flush()
|
||||||
db.session.commit()
|
db.session.commit()
|
||||||
return Response(status=204)
|
return Response(status=204)
|
||||||
|
|
||||||
|
@auth.Auth.requires_auth
|
||||||
|
def delete(self, tag_id: str, device_id: str):
|
||||||
|
tag = Tag.from_an_id(tag_id).filter_by(owner=g.user).one() # type: Tag
|
||||||
|
device = Device.query.filter_by(owner=g.user).filter_by(id=device_id).one()
|
||||||
|
if tag.provider:
|
||||||
|
# if is an unamed tag not do nothing
|
||||||
|
return Response(status=204)
|
||||||
|
|
||||||
|
if tag.device == device:
|
||||||
|
tag.device_id = None
|
||||||
|
db.session().final_flush()
|
||||||
|
db.session.commit()
|
||||||
|
return Response(status=204)
|
||||||
|
|
||||||
|
|
||||||
def get_device_from_tag(id: str):
|
def get_device_from_tag(id: str):
|
||||||
"""Gets the device by passing a tag id.
|
"""Gets the device by passing a tag id.
|
||||||
|
|
6
ereuse_devicehub/resources/utils.py
Normal file
6
ereuse_devicehub/resources/utils.py
Normal file
|
@ -0,0 +1,6 @@
|
||||||
|
from hashids import Hashids
|
||||||
|
from decouple import config
|
||||||
|
|
||||||
|
ALPHABET = '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ'
|
||||||
|
SECRET = config('TAG_HASH', '')
|
||||||
|
hashcode = Hashids(SECRET, min_length=5, alphabet=ALPHABET)
|
File diff suppressed because one or more lines are too long
|
@ -750,6 +750,8 @@ def test_deallocate_bad_dates(user: UserClient):
|
||||||
def test_trade(action_model_state: Tuple[Type[models.Action], states.Trading], user: UserClient):
|
def test_trade(action_model_state: Tuple[Type[models.Action], states.Trading], user: UserClient):
|
||||||
"""Tests POSTing all Trade actions."""
|
"""Tests POSTing all Trade actions."""
|
||||||
# todo missing None states.Trading for after cancelling renting, for example
|
# todo missing None states.Trading for after cancelling renting, for example
|
||||||
|
# import pdb; pdb.set_trace()
|
||||||
|
# Remove this test
|
||||||
action_model, state = action_model_state
|
action_model, state = action_model_state
|
||||||
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
|
snapshot, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
|
||||||
action = {
|
action = {
|
||||||
|
|
|
@ -119,4 +119,4 @@ def test_api_docs(client: Client):
|
||||||
'scheme': 'basic',
|
'scheme': 'basic',
|
||||||
'name': 'Authorization'
|
'name': 'Authorization'
|
||||||
}
|
}
|
||||||
assert len(docs['definitions']) == 117
|
assert len(docs['definitions']) == 118
|
||||||
|
|
|
@ -584,7 +584,6 @@ def test_verify_stamp_erasure_certificate(user: UserClient, client: Client):
|
||||||
"""Test verify stamp of one export certificate in PDF."""
|
"""Test verify stamp of one export certificate in PDF."""
|
||||||
s = file('erase-sectors.snapshot')
|
s = file('erase-sectors.snapshot')
|
||||||
snapshot, response = user.post(s, res=Snapshot)
|
snapshot, response = user.post(s, res=Snapshot)
|
||||||
# import pdb; pdb.set_trace()
|
|
||||||
|
|
||||||
doc, _ = user.get(res=documents.DocumentDef.t,
|
doc, _ = user.get(res=documents.DocumentDef.t,
|
||||||
item='erasures/',
|
item='erasures/',
|
||||||
|
|
|
@ -8,7 +8,7 @@ from pytest import raises
|
||||||
from teal.db import MultipleResourcesFound, ResourceNotFound, UniqueViolation, DBError
|
from teal.db import MultipleResourcesFound, ResourceNotFound, UniqueViolation, DBError
|
||||||
from teal.marshmallow import ValidationError
|
from teal.marshmallow import ValidationError
|
||||||
|
|
||||||
from ereuse_devicehub.client import UserClient
|
from ereuse_devicehub.client import UserClient, Client
|
||||||
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.resources.action.models import Snapshot
|
from ereuse_devicehub.resources.action.models import Snapshot
|
||||||
|
@ -33,6 +33,68 @@ def test_create_tag(user: UserClient):
|
||||||
tag = Tag.query.one()
|
tag = Tag.query.one()
|
||||||
assert tag.id == 'bar-1'
|
assert tag.id == 'bar-1'
|
||||||
assert tag.provider == URL('http://foo.bar')
|
assert tag.provider == URL('http://foo.bar')
|
||||||
|
res, _ = user.get(res=Tag, item=tag.code, status=422)
|
||||||
|
assert res['type'] == 'TagNotLinked'
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.mvp
|
||||||
|
@pytest.mark.usefixtures(conftest.app_context.__name__)
|
||||||
|
def test_create_tag_with_device(user: UserClient):
|
||||||
|
"""Creates a tag specifying linked with one device."""
|
||||||
|
pc = Desktop(serial_number='sn1', chassis=ComputerChassis.Tower, owner_id=user.user['id'])
|
||||||
|
db.session.add(pc)
|
||||||
|
db.session.commit()
|
||||||
|
tag = Tag(id='bar', owner_id=user.user['id'])
|
||||||
|
db.session.add(tag)
|
||||||
|
db.session.commit()
|
||||||
|
data = '{tag_id}/device/{device_id}'.format(tag_id=tag.id, device_id=pc.id)
|
||||||
|
user.put({}, res=Tag, item=data, status=204)
|
||||||
|
user.get(res=Tag, item='{}/device'.format(tag.id))
|
||||||
|
user.delete({}, res=Tag, item=data, status=204)
|
||||||
|
res, _ = user.get(res=Tag, item='{}/device'.format(tag.id), status=422)
|
||||||
|
assert res['type'] == 'TagNotLinked'
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.mvp
|
||||||
|
@pytest.mark.usefixtures(conftest.app_context.__name__)
|
||||||
|
def test_delete_tags(user: UserClient, client: Client):
|
||||||
|
"""Delete a named tag."""
|
||||||
|
# Delete Tag Named
|
||||||
|
pc = Desktop(serial_number='sn1', chassis=ComputerChassis.Tower, owner_id=user.user['id'])
|
||||||
|
db.session.add(pc)
|
||||||
|
db.session.commit()
|
||||||
|
tag = Tag(id='bar', owner_id=user.user['id'], device_id=pc.id)
|
||||||
|
db.session.add(tag)
|
||||||
|
db.session.commit()
|
||||||
|
tag = Tag.query.one()
|
||||||
|
assert tag.id == 'bar'
|
||||||
|
# Is not possible delete one tag linked to one device
|
||||||
|
res, _ = user.delete(res=Tag, item=tag.id, status=422)
|
||||||
|
msg = 'The tag bar is linked to device'
|
||||||
|
assert msg in res['message'][0]
|
||||||
|
|
||||||
|
tag.device_id = None
|
||||||
|
db.session.add(tag)
|
||||||
|
db.session.commit()
|
||||||
|
# Is not possible delete one tag from an anonymous user
|
||||||
|
client.delete(res=Tag, item=tag.id, status=401)
|
||||||
|
|
||||||
|
# Is possible delete one normal tag
|
||||||
|
user.delete(res=Tag, item=tag.id)
|
||||||
|
user.get(res=Tag, item=tag.id, status=404)
|
||||||
|
|
||||||
|
# Delete Tag UnNamed
|
||||||
|
org = Organization(name='bar', tax_id='bartax')
|
||||||
|
tag = Tag(id='bar-1', org=org, provider=URL('http://foo.bar'), owner_id=user.user['id'])
|
||||||
|
db.session.add(tag)
|
||||||
|
db.session.commit()
|
||||||
|
tag = Tag.query.one()
|
||||||
|
assert tag.id == 'bar-1'
|
||||||
|
res, _ = user.delete(res=Tag, item=tag.id, status=422)
|
||||||
|
msg = 'This tag {} is unnamed tag. It is imposible delete.'.format(tag.id)
|
||||||
|
assert msg in res['message']
|
||||||
|
tag = Tag.query.one()
|
||||||
|
assert tag.id == 'bar-1'
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.mvp
|
@pytest.mark.mvp
|
||||||
|
@ -51,13 +113,15 @@ def test_create_tag_default_org(user: UserClient):
|
||||||
|
|
||||||
@pytest.mark.mvp
|
@pytest.mark.mvp
|
||||||
@pytest.mark.usefixtures(conftest.app_context.__name__)
|
@pytest.mark.usefixtures(conftest.app_context.__name__)
|
||||||
def test_create_tag_no_slash():
|
def test_create_same_tag_default_org_two_users(user: UserClient, user2: UserClient):
|
||||||
"""Checks that no tags can be created that contain a slash."""
|
"""Creates a tag using the default organization."""
|
||||||
with raises(ValidationError):
|
tag = Tag(id='foo-1', owner_id=user.user['id'])
|
||||||
Tag('/')
|
tag2 = Tag(id='foo-1', owner_id=user2.user['id'])
|
||||||
|
db.session.add(tag)
|
||||||
with raises(ValidationError):
|
db.session.add(tag2)
|
||||||
Tag('bar', secondary='/')
|
db.session.commit()
|
||||||
|
assert tag.org.name == 'FooOrg' # as defined in the settings
|
||||||
|
assert tag2.org.name == 'FooOrg' # as defined in the settings
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.mvp
|
@pytest.mark.mvp
|
||||||
|
@ -75,7 +139,19 @@ def test_create_two_same_tags(user: UserClient):
|
||||||
db.session.add(Tag(id='foo-bar', owner_id=user.user['id']))
|
db.session.add(Tag(id='foo-bar', owner_id=user.user['id']))
|
||||||
org2 = Organization(name='org 2', tax_id='tax id org 2')
|
org2 = Organization(name='org 2', tax_id='tax id org 2')
|
||||||
db.session.add(Tag(id='foo-bar', org=org2, owner_id=user.user['id']))
|
db.session.add(Tag(id='foo-bar', org=org2, owner_id=user.user['id']))
|
||||||
db.session.commit()
|
with raises(DBError):
|
||||||
|
db.session.commit()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.mvp
|
||||||
|
@pytest.mark.usefixtures(conftest.app_context.__name__)
|
||||||
|
def test_create_tag_no_slash():
|
||||||
|
"""Checks that no tags can be created that contain a slash."""
|
||||||
|
with raises(ValidationError):
|
||||||
|
Tag('/')
|
||||||
|
|
||||||
|
with raises(ValidationError):
|
||||||
|
Tag('bar', secondary='/')
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.mvp
|
@pytest.mark.mvp
|
||||||
|
@ -131,17 +207,39 @@ def test_tag_get_device_from_tag_endpoint_no_tag(user: UserClient):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.mvp
|
@pytest.mark.mvp
|
||||||
def test_tag_get_device_from_tag_endpoint_multiple_tags(app: Devicehub, user: UserClient):
|
@pytest.mark.usefixtures(conftest.app_context.__name__)
|
||||||
"""As above, but when there are two tags with the same ID, the
|
def test_tag_get_device_from_tag_endpoint_multiple_tags(app: Devicehub, user: UserClient, user2: UserClient, client: Client):
|
||||||
|
"""As above, but when there are two tags with the secondary ID, the
|
||||||
system should not return any of both (to be deterministic) so
|
system should not return any of both (to be deterministic) so
|
||||||
it should raise an exception.
|
it should raise an exception.
|
||||||
"""
|
"""
|
||||||
with app.app_context():
|
db.session.add(Tag(id='foo', secondary='bar', owner_id=user.user['id']))
|
||||||
db.session.add(Tag(id='foo-bar', owner_id=user.user['id']))
|
db.session.commit()
|
||||||
org2 = Organization(name='org 2', tax_id='tax id org 2')
|
|
||||||
db.session.add(Tag(id='foo-bar', org=org2, owner_id=user.user['id']))
|
db.session.add(Tag(id='foo', secondary='bar', owner_id=user2.user['id']))
|
||||||
|
db.session.commit()
|
||||||
|
|
||||||
|
db.session.add(Tag(id='foo2', secondary='bar', owner_id=user.user['id']))
|
||||||
|
with raises(DBError):
|
||||||
db.session.commit()
|
db.session.commit()
|
||||||
user.get(res=Tag, item='foo-bar/device', status=MultipleResourcesFound)
|
db.session.rollback()
|
||||||
|
|
||||||
|
tag1 = Tag.from_an_id('foo').filter_by(owner_id=user.user['id']).one()
|
||||||
|
tag2 = Tag.from_an_id('foo').filter_by(owner_id=user2.user['id']).one()
|
||||||
|
pc1 = Desktop(serial_number='sn1', chassis=ComputerChassis.Tower, owner_id=user.user['id'])
|
||||||
|
pc2 = Desktop(serial_number='sn2', chassis=ComputerChassis.Tower, owner_id=user2.user['id'])
|
||||||
|
pc1.tags.add(tag1)
|
||||||
|
pc2.tags.add(tag2)
|
||||||
|
db.session.add(pc1)
|
||||||
|
db.session.add(pc2)
|
||||||
|
db.session.commit()
|
||||||
|
computer, _ = user.get(res=Tag, item='foo/device')
|
||||||
|
assert computer['serialNumber'] == 'sn1'
|
||||||
|
computer, _ = user2.get(res=Tag, item='foo/device')
|
||||||
|
assert computer['serialNumber'] == 'sn2'
|
||||||
|
|
||||||
|
_, status = client.get(res=Tag, item='foo/device', status=MultipleResourcesFound)
|
||||||
|
assert status.status_code == 422
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.mvp
|
@pytest.mark.mvp
|
||||||
|
@ -216,8 +314,7 @@ def test_tag_secondary_workbench_link_find(user: UserClient):
|
||||||
t = Tag('foo', secondary='bar', owner_id=user.user['id'])
|
t = Tag('foo', secondary='bar', owner_id=user.user['id'])
|
||||||
db.session.add(t)
|
db.session.add(t)
|
||||||
db.session.flush()
|
db.session.flush()
|
||||||
assert Tag.from_an_id('bar').one() == t
|
assert Tag.from_an_id('bar').one() == Tag.from_an_id('foo').one()
|
||||||
assert Tag.from_an_id('foo').one() == t
|
|
||||||
with pytest.raises(ResourceNotFound):
|
with pytest.raises(ResourceNotFound):
|
||||||
Tag.from_an_id('nope').one()
|
Tag.from_an_id('nope').one()
|
||||||
|
|
||||||
|
|
Reference in a new issue