Filter devices by being inside lots
This commit is contained in:
parent
7e4a0981a1
commit
863578559c
|
@ -4,7 +4,8 @@ import marshmallow
|
|||
from flask import current_app as app, render_template, request
|
||||
from flask.json import jsonify
|
||||
from flask_sqlalchemy import Pagination
|
||||
from marshmallow import fields as f, validate as v
|
||||
from marshmallow import fields, fields as f, validate as v
|
||||
from sqlalchemy.orm import aliased
|
||||
from teal import query
|
||||
from teal.cache import cache
|
||||
from teal.resource import View
|
||||
|
@ -12,9 +13,10 @@ from teal.resource import View
|
|||
from ereuse_devicehub import auth
|
||||
from ereuse_devicehub.db import db
|
||||
from ereuse_devicehub.resources import search
|
||||
from ereuse_devicehub.resources.device.models import Device, Manufacturer
|
||||
from ereuse_devicehub.resources.device.models import Component, Computer, Device, Manufacturer
|
||||
from ereuse_devicehub.resources.device.search import DeviceSearch
|
||||
from ereuse_devicehub.resources.event.models import Rate
|
||||
from ereuse_devicehub.resources.lot.models import Lot, LotDevice
|
||||
from ereuse_devicehub.resources.tag.model import Tag
|
||||
|
||||
|
||||
|
@ -39,16 +41,28 @@ class TagQ(query.Query):
|
|||
org = query.ILike(Tag.org)
|
||||
|
||||
|
||||
class LotQ(query.Query):
|
||||
id = query.Or(query.QueryField(Lot.descendantsq, fields.UUID()))
|
||||
|
||||
|
||||
class Filters(query.Query):
|
||||
_parent = aliased(Computer)
|
||||
_device_inside_lot = (Device.id == LotDevice.device_id) & (Lot.id == LotDevice.lot_id)
|
||||
_component_inside_lot_through_parent = (Device.id == Component.id) \
|
||||
& (Component.parent_id == _parent.id) \
|
||||
& (_parent.id == LotDevice.device_id)
|
||||
|
||||
type = query.Or(OfType(Device.type))
|
||||
model = query.ILike(Device.model)
|
||||
manufacturer = query.ILike(Device.manufacturer)
|
||||
serialNumber = query.ILike(Device.serial_number)
|
||||
rating = query.Join(Device.id == Rate.device_id, RateQ)
|
||||
tag = query.Join(Device.id == Tag.id, TagQ)
|
||||
tag = query.Join(Device.id == Tag.device_id, TagQ)
|
||||
lot = query.Join(_device_inside_lot | _component_inside_lot_through_parent, LotQ)
|
||||
|
||||
|
||||
class Sorting(query.Sort):
|
||||
id = query.SortField(Device.id)
|
||||
created = query.SortField(Device.created)
|
||||
|
||||
|
||||
|
|
|
@ -77,14 +77,27 @@ class Lot(Thing):
|
|||
.join(self.__class__.paths) \
|
||||
.filter(Path.path.lquery(exp.cast('*.{}.*{{1}}'.format(id), LQUERY)))
|
||||
|
||||
@property
|
||||
def descendants(self):
|
||||
return self.descendantsq(self.id)
|
||||
|
||||
@classmethod
|
||||
def descendantsq(cls, id):
|
||||
_id = UUIDLtree.convert(id)
|
||||
return (cls.id == Path.lot_id) & Path.path.lquery(exp.cast('*.{}.*'.format(_id), LQUERY))
|
||||
|
||||
@property
|
||||
def parents(self):
|
||||
return self.parentsq(self.id)
|
||||
|
||||
@classmethod
|
||||
def parentsq(cls, id: UUID):
|
||||
"""The parent lots."""
|
||||
id = UUIDLtree.convert(self.id)
|
||||
id = UUIDLtree.convert(id)
|
||||
i = db.func.index(Path.path, id)
|
||||
parent_id = db.func.replace(exp.cast(db.func.subpath(Path.path, i - 1, i), TEXT), '_', '-')
|
||||
join_clause = parent_id == exp.cast(Lot.id, TEXT)
|
||||
return self.query.join(Path, join_clause).filter(
|
||||
return cls.query.join(Path, join_clause).filter(
|
||||
Path.path.lquery(exp.cast('*{{1}}.{}.*'.format(id), LQUERY))
|
||||
)
|
||||
|
||||
|
|
|
@ -43,14 +43,27 @@ class Lot(Thing):
|
|||
def children(self) -> LotQuery:
|
||||
pass
|
||||
|
||||
@property
|
||||
def descendants(self) -> LotQuery:
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def descendantsq(cls, id) -> LotQuery:
|
||||
pass
|
||||
|
||||
@property
|
||||
def parents(self) -> LotQuery:
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def parentsq(cls, id) -> LotQuery:
|
||||
pass
|
||||
|
||||
@property
|
||||
def url(self) -> urlutils.URL:
|
||||
pass
|
||||
|
||||
|
||||
class Path:
|
||||
id = ... # type: Column
|
||||
lot_id = ... # type: Column
|
||||
|
|
|
@ -8,6 +8,7 @@ from ereuse_devicehub.resources.device.models import Desktop, Device, Laptop, So
|
|||
from ereuse_devicehub.resources.device.views import Filters, Sorting
|
||||
from ereuse_devicehub.resources.enums import ComputerChassis
|
||||
from ereuse_devicehub.resources.event.models import Snapshot
|
||||
from ereuse_devicehub.resources.lot.models import Lot
|
||||
from tests import conftest
|
||||
from tests.conftest import file
|
||||
|
||||
|
@ -98,6 +99,50 @@ def test_device_query_filter_sort(user: UserClient):
|
|||
assert tuple(d['type'] for d in i['items']) == ('Desktop', 'Laptop', 'Desktop')
|
||||
|
||||
|
||||
@pytest.mark.usefixtures(device_query_dummy.__name__)
|
||||
def test_device_query_filter_lots(user: UserClient):
|
||||
parent, _ = user.post({'name': 'Parent'}, res=Lot)
|
||||
child, _ = user.post({'name': 'Child'}, res=Lot)
|
||||
parent, _ = user.post({},
|
||||
res=Lot,
|
||||
item='{}/children'.format(parent['id']),
|
||||
query=[('id', child['id'])])
|
||||
i, _ = user.get(res=Device, query=[
|
||||
('filter', {'type': ['Computer']})
|
||||
])
|
||||
lot, _ = user.post({},
|
||||
res=Lot,
|
||||
item='{}/devices'.format(parent['id']),
|
||||
query=[('id', d['id']) for d in i['items'][:-1]])
|
||||
lot, _ = user.post({},
|
||||
res=Lot,
|
||||
item='{}/devices'.format(child['id']),
|
||||
query=[('id', i['items'][-1]['id'])])
|
||||
i, _ = user.get(res=Device, query=[
|
||||
('filter', {'lot': {'id': [parent['id']]}}),
|
||||
('sort', {'id': Sorting.ASCENDING})
|
||||
])
|
||||
assert len(i['items']) == 4
|
||||
assert tuple(x['id'] for x in i['items']) == (1, 2, 3, 4), \
|
||||
'The parent lot contains 2 items plus indirectly the third one, and 1st device the HDD.'
|
||||
|
||||
s, _ = user.get(res=Device, query=[
|
||||
('filter', {'lot': {'id': [child['id']]}})
|
||||
])
|
||||
assert s['items'][0]['chassis'] == 'Microtower', 'The child lot only contains the last device.'
|
||||
s, _ = user.get(res=Device, query=[
|
||||
('filter', {'lot': {'id': [child['id'], parent['id']]}})
|
||||
])
|
||||
assert all(x['id'] == id for x, id in zip(i['items'], (1, 2, 3, 4))), \
|
||||
'Adding both lots is redundant in this case and we have the 4 elements.'
|
||||
i, _ = user.get(res=Device, query=[
|
||||
('filter', {'lot': {'id': [parent['id']]}, 'type': ['Computer']}),
|
||||
('sort', {'id': Sorting.ASCENDING})
|
||||
])
|
||||
assert len(i['items']) == 3
|
||||
assert tuple(x['id'] for x in i['items']) == (1, 2, 3), 'Only computers now'
|
||||
|
||||
|
||||
def test_device_query(user: UserClient):
|
||||
"""Checks result of inventory."""
|
||||
user.post(conftest.file('basic.snapshot'), res=Snapshot)
|
||||
|
|
Reference in New Issue