Merge pull request #167 from eReuse/feature/new-recycling-action
Feature/new recycling action
This commit is contained in:
cayop
GitHub
commit
3e0e1c5716
|
|
@ -13,6 +13,7 @@ ml).
|
|||
|
||||
## [1.0.10-beta]
|
||||
- [bugfix] #168 can to do a trade without devices.
|
||||
- [addend] #167 new actions of status devices: use, recycling, refurbish and management.
|
||||
|
||||
## [1.0.9-beta]
|
||||
- [addend] #159 external document as proof of erase of disk
|
||||
|
|
|
|||
|
|
@ -1 +1 @@
|
|||
__version__ = "1.0.9-beta"
|
||||
__version__ = "1.0.10-beta"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,38 @@
|
|||
"""adding state actions
|
||||
|
||||
Revision ID: a0978ac6cf4a
|
||||
Revises: 7ecb8ff7abad
|
||||
Create Date: 2021-09-24 12:03:01.661679
|
||||
|
||||
"""
|
||||
from alembic import op, context
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.dialects import postgresql
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = 'a0978ac6cf4a'
|
||||
down_revision = '3ac2bc1897ce'
|
||||
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():
|
||||
op.create_table('action_status',
|
||||
sa.Column('rol_user_id', postgresql.UUID(as_uuid=True), nullable=False),
|
||||
sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
|
||||
sa.ForeignKeyConstraint(['id'], [f'{get_inv()}.action.id'], ),
|
||||
sa.ForeignKeyConstraint(['rol_user_id'], ['common.user.id'], ),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
schema=f'{get_inv()}'
|
||||
)
|
||||
|
||||
|
||||
def downgrade():
|
||||
op.drop_table('action_status', schema=f'{get_inv()}')
|
||||
|
|
@ -194,6 +194,26 @@ class ReadyDef(ActionDef):
|
|||
SCHEMA = schemas.Ready
|
||||
|
||||
|
||||
class RecyclingDef(ActionDef):
|
||||
VIEW = None
|
||||
SCHEMA = schemas.Recycling
|
||||
|
||||
|
||||
class UseDef(ActionDef):
|
||||
VIEW = None
|
||||
SCHEMA = schemas.Use
|
||||
|
||||
|
||||
class RefurbishDef(ActionDef):
|
||||
VIEW = None
|
||||
SCHEMA = schemas.Refurbish
|
||||
|
||||
|
||||
class ManagementDef(ActionDef):
|
||||
VIEW = None
|
||||
SCHEMA = schemas.Management
|
||||
|
||||
|
||||
class ToPrepareDef(ActionDef):
|
||||
VIEW = None
|
||||
SCHEMA = schemas.ToPrepare
|
||||
|
|
|
|||
|
|
@ -1350,6 +1350,33 @@ class DataWipe(JoinedTableMixin, ActionWithMultipleDevices):
|
|||
primaryjoin='DataWipe.document_id == DataWipeDocument.id')
|
||||
|
||||
|
||||
class ActionStatus(JoinedTableMixin, ActionWithMultipleTradeDocuments):
|
||||
"""This is a meta-action than mark the status of the devices"""
|
||||
|
||||
rol_user_id = db.Column(UUID(as_uuid=True),
|
||||
db.ForeignKey(User.id),
|
||||
nullable=False,
|
||||
default=lambda: g.user.id)
|
||||
rol_user = db.relationship(User, primaryjoin=rol_user_id == User.id)
|
||||
rol_user_comment = """The user that ."""
|
||||
|
||||
|
||||
class Recycling(ActionStatus):
|
||||
"""This action mark devices as recycling"""
|
||||
|
||||
|
||||
class Use(ActionStatus):
|
||||
"""This action mark one devices or container as use"""
|
||||
|
||||
|
||||
class Refurbish(ActionStatus):
|
||||
"""This action mark one devices or container as refurbish"""
|
||||
|
||||
|
||||
class Management(ActionStatus):
|
||||
"""This action mark one devices or container as management"""
|
||||
|
||||
|
||||
class Prepare(ActionWithMultipleDevices):
|
||||
"""Work has been performed to the device to a defined point of
|
||||
acceptance.
|
||||
|
|
@ -1471,6 +1498,20 @@ class CancelReservation(Organize):
|
|||
"""The act of cancelling a reservation."""
|
||||
|
||||
|
||||
class ActionStatusDocuments(JoinedTableMixin, ActionWithMultipleTradeDocuments):
|
||||
"""This is a meta-action that marks the state of the devices."""
|
||||
rol_user_id = db.Column(UUID(as_uuid=True),
|
||||
db.ForeignKey(User.id),
|
||||
nullable=False,
|
||||
default=lambda: g.user.id)
|
||||
rol_user = db.relationship(User, primaryjoin=rol_user_id == User.id)
|
||||
rol_user_comment = """The user that ."""
|
||||
|
||||
|
||||
class RecyclingDocument(ActionStatusDocuments):
|
||||
"""This action mark one document or container as recycling"""
|
||||
|
||||
|
||||
class ConfirmDocument(JoinedTableMixin, ActionWithMultipleTradeDocuments):
|
||||
"""Users confirm the one action trade this confirmation it's link to trade
|
||||
and the document that confirm
|
||||
|
|
|
|||
|
|
@ -424,6 +424,50 @@ class Ready(ActionWithMultipleDevices):
|
|||
__doc__ = m.Ready.__doc__
|
||||
|
||||
|
||||
class ActionStatus(Action):
|
||||
rol_user = NestedOn(s_user.User, dump_only=True, exclude=('token',))
|
||||
devices = NestedOn(s_device.Device,
|
||||
many=True,
|
||||
required=False, # todo test ensuring len(devices) >= 1
|
||||
only_query='id',
|
||||
collection_class=OrderedSet)
|
||||
documents = NestedOn(s_document.TradeDocument,
|
||||
many=True,
|
||||
required=False, # todo test ensuring len(devices) >= 1
|
||||
only_query='id',
|
||||
collection_class=OrderedSet)
|
||||
|
||||
@pre_load
|
||||
def put_devices(self, data: dict):
|
||||
if not 'devices' in data.keys():
|
||||
data['devices'] = []
|
||||
|
||||
@post_load
|
||||
def put_rol_user(self, data: dict):
|
||||
for dev in data['devices']:
|
||||
if dev.trading in [None, 'Revoke', 'ConfirmRevoke']:
|
||||
return data
|
||||
trade = [ac for ac in dev.actions if ac.t == 'Trade'][-1]
|
||||
if trade.user_to != g.user:
|
||||
data['rol_user'] = trade.user_to
|
||||
|
||||
|
||||
class Recycling(ActionStatus):
|
||||
__doc__ = m.Recycling.__doc__
|
||||
|
||||
|
||||
class Use(ActionStatus):
|
||||
__doc__ = m.Use.__doc__
|
||||
|
||||
|
||||
class Refurbish(ActionStatus):
|
||||
__doc__ = m.Refurbish.__doc__
|
||||
|
||||
|
||||
class Management(ActionStatus):
|
||||
__doc__ = m.Management.__doc__
|
||||
|
||||
|
||||
class ToPrepare(ActionWithMultipleDevices):
|
||||
__doc__ = m.ToPrepare.__doc__
|
||||
|
||||
|
|
|
|||
|
|
@ -261,6 +261,46 @@ class Device(Thing):
|
|||
with suppress(LookupError, ValueError):
|
||||
return self.last_action_of(*states.Trading.actions())
|
||||
|
||||
@property
|
||||
def status(self):
|
||||
"""Show the actual status of device for this owner.
|
||||
The status depend of one of this 4 actions:
|
||||
- Use
|
||||
- Refurbish
|
||||
- Recycling
|
||||
- Management
|
||||
"""
|
||||
from ereuse_devicehub.resources.device import states
|
||||
with suppress(LookupError, ValueError):
|
||||
return self.last_action_of(*states.Status.actions())
|
||||
|
||||
@property
|
||||
def history_status(self):
|
||||
"""Show the history of the status actions of the device.
|
||||
The status depend of one of this 4 actions:
|
||||
- Use
|
||||
- Refurbish
|
||||
- Recycling
|
||||
- Management
|
||||
"""
|
||||
from ereuse_devicehub.resources.device import states
|
||||
status_actions = [ac.t for ac in states.Status.actions()]
|
||||
history = []
|
||||
for ac in self.actions:
|
||||
if not ac.t in status_actions:
|
||||
continue
|
||||
if not history:
|
||||
history.append(ac)
|
||||
continue
|
||||
if ac.rol_user == history[-1].rol_user:
|
||||
# get only the last action consecutive for the same user
|
||||
history = history[:-1] + [ac]
|
||||
continue
|
||||
|
||||
history.append(ac)
|
||||
|
||||
return history
|
||||
|
||||
@property
|
||||
def trading(self):
|
||||
"""The trading state, or None if no Trade action has
|
||||
|
|
|
|||
|
|
@ -83,3 +83,16 @@ class Usage(State):
|
|||
Allocate = e.Allocate
|
||||
Deallocate = e.Deallocate
|
||||
InUse = e.Live
|
||||
|
||||
|
||||
class Status(State):
|
||||
"""Define status of device for one user.
|
||||
:cvar Use: The device is in use for one final user.
|
||||
:cvar Refurbish: The device is owned by one refurbisher.
|
||||
:cvar Recycling: The device is sended to recycling.
|
||||
:cvar Management: The device is owned by one Manager.
|
||||
"""
|
||||
Use = e.Use
|
||||
Refurbish = e.Refurbish
|
||||
Recycling = e.Recycling
|
||||
Management = e.Management
|
||||
|
|
|
|||
|
|
@ -256,6 +256,282 @@ def test_generic_action(action_model_state: Tuple[models.Action, states.Trading]
|
|||
assert snapshot['device']['updated'] != device['updated']
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
@pytest.mark.parametrize('action_model',
|
||||
(pytest.param(ams, id=ams.__class__.__name__)
|
||||
for ams in [
|
||||
models.Recycling,
|
||||
models.Use,
|
||||
models.Refurbish,
|
||||
models.Management
|
||||
]))
|
||||
def test_simple_status_actions(action_model: models.Action, user: UserClient, user2: UserClient):
|
||||
"""Simple test of status action."""
|
||||
snap, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
|
||||
|
||||
action = {'type': action_model.t, 'devices': [snap['device']['id']]}
|
||||
action, _ = user.post(action, res=models.Action)
|
||||
device, _ = user.get(res=Device, item=snap['device']['devicehubID'])
|
||||
assert device['actions'][-1]['id'] == action['id']
|
||||
assert action['author']['id'] == user.user['id']
|
||||
assert action['rol_user']['id'] == user.user['id']
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
@pytest.mark.parametrize('action_model',
|
||||
(pytest.param(ams, id=ams.__class__.__name__)
|
||||
for ams in [
|
||||
models.Recycling,
|
||||
models.Use,
|
||||
models.Refurbish,
|
||||
models.Management
|
||||
]))
|
||||
def test_outgoinlot_status_actions(action_model: models.Action, user: UserClient, user2: UserClient):
|
||||
"""Test of status actions in outgoinlot."""
|
||||
snap, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
|
||||
device, _ = user.get(res=Device, item=snap['device']['devicehubID'])
|
||||
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
|
||||
user.post({},
|
||||
res=Lot,
|
||||
item='{}/devices'.format(lot['id']),
|
||||
query=[('id', device['id'])])
|
||||
|
||||
request_post = {
|
||||
'type': 'Trade',
|
||||
'devices': [device['id']],
|
||||
'userFromEmail': user.email,
|
||||
'userToEmail': user2.email,
|
||||
'price': 10,
|
||||
'date': "2020-12-01T02:00:00+00:00",
|
||||
'lot': lot['id'],
|
||||
'confirms': True,
|
||||
}
|
||||
|
||||
user.post(res=models.Action, data=request_post)
|
||||
action = {'type': action_model.t, 'devices': [device['id']]}
|
||||
action, _ = user.post(action, res=models.Action)
|
||||
device, _ = user.get(res=Device, item=snap['device']['devicehubID'])
|
||||
|
||||
assert device['actions'][-1]['id'] == action['id']
|
||||
assert action['author']['id'] == user.user['id']
|
||||
assert action['rol_user']['id'] == user2.user['id']
|
||||
|
||||
# Remove device from lot
|
||||
lot, _ = user.delete({},
|
||||
res=Lot,
|
||||
item='{}/devices'.format(lot['id']),
|
||||
query=[('id', device['id'])], status=200)
|
||||
|
||||
action = {'type': action_model.t, 'devices': [device['id']]}
|
||||
action, _ = user.post(action, res=models.Action)
|
||||
device, _ = user.get(res=Device, item=snap['device']['devicehubID'])
|
||||
|
||||
assert device['actions'][-1]['id'] == action['id']
|
||||
assert action['author']['id'] == user.user['id']
|
||||
assert action['rol_user']['id'] == user.user['id']
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
@pytest.mark.parametrize('action_model',
|
||||
(pytest.param(ams, id=ams.__class__.__name__)
|
||||
for ams in [
|
||||
models.Recycling,
|
||||
models.Use,
|
||||
models.Refurbish,
|
||||
models.Management
|
||||
]))
|
||||
def test_incominglot_status_actions(action_model: models.Action, user: UserClient, user2: UserClient):
|
||||
"""Test of status actions in outgoinlot."""
|
||||
snap, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
|
||||
device, _ = user.get(res=Device, item=snap['device']['devicehubID'])
|
||||
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
|
||||
user.post({},
|
||||
res=Lot,
|
||||
item='{}/devices'.format(lot['id']),
|
||||
query=[('id', device['id'])])
|
||||
|
||||
request_post = {
|
||||
'type': 'Trade',
|
||||
'devices': [device['id']],
|
||||
'userFromEmail': user2.email,
|
||||
'userToEmail': user.email,
|
||||
'price': 10,
|
||||
'date': "2020-12-01T02:00:00+00:00",
|
||||
'lot': lot['id'],
|
||||
'confirms': True,
|
||||
}
|
||||
|
||||
user.post(res=models.Action, data=request_post)
|
||||
action = {'type': action_model.t, 'devices': [device['id']]}
|
||||
action, _ = user.post(action, res=models.Action)
|
||||
device, _ = user.get(res=Device, item=snap['device']['devicehubID'])
|
||||
|
||||
assert device['actions'][-1]['id'] == action['id']
|
||||
assert action['author']['id'] == user.user['id']
|
||||
assert action['rol_user']['id'] == user.user['id']
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
@pytest.mark.usefixtures(conftest.app_context.__name__)
|
||||
def test_history_status_actions(user: UserClient, user2: UserClient):
|
||||
"""Test for check the status actions."""
|
||||
snap, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
|
||||
device = Device.query.filter_by(id=snap['device']['id']).one()
|
||||
|
||||
# Case 1
|
||||
action = {'type': models.Recycling.t, 'devices': [device.id]}
|
||||
action, _ = user.post(action, res=models.Action)
|
||||
|
||||
assert str(device.actions[-1].id) == action['id']
|
||||
assert action['id'] == str(device.status.id)
|
||||
assert device.status.t == models.Recycling.t
|
||||
assert [action['id']] == [str(ac.id) for ac in device.history_status]
|
||||
|
||||
# Case 2
|
||||
action2 = {'type': models.Refurbish.t, 'devices': [device.id]}
|
||||
action2, _ = user.post(action2, res=models.Action)
|
||||
assert action2['id'] == str(device.status.id)
|
||||
assert device.status.t == models.Refurbish.t
|
||||
assert [action2['id']] == [str(ac.id) for ac in device.history_status]
|
||||
|
||||
# Case 3
|
||||
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
|
||||
user.post({},
|
||||
res=Lot,
|
||||
item='{}/devices'.format(lot['id']),
|
||||
query=[('id', device.id)])
|
||||
|
||||
request_post = {
|
||||
'type': 'Trade',
|
||||
'devices': [device.id],
|
||||
'userFromEmail': user.email,
|
||||
'userToEmail': user2.email,
|
||||
'price': 10,
|
||||
'date': "2020-12-01T02:00:00+00:00",
|
||||
'lot': lot['id'],
|
||||
'confirms': True,
|
||||
}
|
||||
|
||||
user.post(res=models.Action, data=request_post)
|
||||
action3 = {'type': models.Use.t, 'devices': [device.id]}
|
||||
action3, _ = user.post(action3, res=models.Action)
|
||||
assert action3['id'] == str(device.status.id)
|
||||
assert device.status.t == models.Use.t
|
||||
assert [action2['id'], action3['id']] == [str(ac.id) for ac in device.history_status]
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
@pytest.mark.usefixtures(conftest.app_context.__name__)
|
||||
def test_use_changing_owner(user: UserClient, user2: UserClient):
|
||||
"""Check if is it possible to do a use action for one device
|
||||
when you are not the owner.
|
||||
"""
|
||||
snap, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
|
||||
device = Device.query.filter_by(id=snap['device']['id']).one()
|
||||
|
||||
assert device.owner.email == user.email
|
||||
|
||||
# Trade
|
||||
lot, _ = user.post({'name': 'MyLot'}, res=Lot)
|
||||
user.post({},
|
||||
res=Lot,
|
||||
item='{}/devices'.format(lot['id']),
|
||||
query=[('id', device.id)])
|
||||
|
||||
request_post = {
|
||||
'type': 'Trade',
|
||||
'devices': [device.id],
|
||||
'userFromEmail': user.email,
|
||||
'userToEmail': user2.email,
|
||||
'price': 10,
|
||||
'date': "2020-12-01T02:00:00+00:00",
|
||||
'lot': lot['id'],
|
||||
'confirms': True,
|
||||
}
|
||||
|
||||
user.post(res=models.Action, data=request_post)
|
||||
trade = models.Trade.query.one()
|
||||
|
||||
# Doble confirmation and change of owner
|
||||
request_confirm = {
|
||||
'type': 'Confirm',
|
||||
'action': trade.id,
|
||||
'devices': [device.id]
|
||||
}
|
||||
|
||||
user2.post(res=models.Action, data=request_confirm)
|
||||
assert device.owner.email == user2.email
|
||||
|
||||
# Adding action Use
|
||||
action3 = {'type': models.Use.t, 'devices': [device.id]}
|
||||
action3, _ = user.post(action3, res=models.Action)
|
||||
assert action3['id'] == str(device.status.id)
|
||||
assert device.status.t == models.Use.t
|
||||
assert device.owner.email == user2.email
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
@pytest.mark.usefixtures(conftest.app_context.__name__)
|
||||
def test_recycling_container(user: UserClient):
|
||||
"""Test of status action recycling for a container."""
|
||||
lot, _ = user.post({'name': 'MyLotOut'}, res=Lot)
|
||||
url = 'http://www.ereuse.org/',
|
||||
request_post = {
|
||||
'filename': 'test.pdf',
|
||||
'hash': 'bbbbbbbb',
|
||||
'url': url,
|
||||
'weight': 150,
|
||||
'lot': lot['id']
|
||||
}
|
||||
tradedocument, _ = user.post(res=TradeDocument, data=request_post)
|
||||
action = {'type': models.Recycling.t, 'devices': [], 'documents': [tradedocument['id']]}
|
||||
action, _ = user.post(action, res=models.Action)
|
||||
trade = TradeDocument.query.one()
|
||||
assert str(trade.actions[0].id) == action['id']
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
@pytest.mark.parametrize('action_model',
|
||||
(pytest.param(ams, id=ams.__class__.__name__)
|
||||
for ams in [
|
||||
models.Recycling,
|
||||
models.Use,
|
||||
models.Refurbish,
|
||||
models.Management
|
||||
]))
|
||||
def test_status_without_lot(action_model: models.Action, user: UserClient):
|
||||
"""Test of status actions for devices without lot."""
|
||||
snap, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
|
||||
action = {'type': action_model.t, 'devices': [snap['device']['id']]}
|
||||
action, _ = user.post(action, res=models.Action)
|
||||
device, _ = user.get(res=Device, item=snap['device']['devicehubID'])
|
||||
assert device['actions'][-1]['id'] == action['id']
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
@pytest.mark.parametrize('action_model',
|
||||
(pytest.param(ams, id=ams.__class__.__name__)
|
||||
for ams in [
|
||||
models.Recycling,
|
||||
models.Use,
|
||||
models.Refurbish,
|
||||
models.Management
|
||||
]))
|
||||
def test_status_in_temporary_lot(action_model: models.Action, user: UserClient):
|
||||
"""Test of status actions for devices in a temporary lot."""
|
||||
snap, _ = user.post(file('basic.snapshot'), res=models.Snapshot)
|
||||
device_id = snap['device']['id']
|
||||
lot, _ = user.post({'name': 'MyLotOut'}, res=Lot)
|
||||
lot, _ = user.post({},
|
||||
res=Lot,
|
||||
item='{}/devices'.format(lot['id']),
|
||||
query=[('id', device_id)])
|
||||
action = {'type': action_model.t, 'devices': [device_id]}
|
||||
action, _ = user.post(action, res=models.Action)
|
||||
device, _ = user.get(res=Device, item=snap['device']['devicehubID'])
|
||||
assert device['actions'][-1]['id'] == action['id']
|
||||
|
||||
|
||||
@pytest.mark.mvp
|
||||
@pytest.mark.usefixtures(conftest.app_context.__name__)
|
||||
def test_live(user: UserClient, client: Client, app: Devicehub):
|
||||
|
|
|
|||
|
|
@ -122,4 +122,4 @@ def test_api_docs(client: Client):
|
|||
'scheme': 'basic',
|
||||
'name': 'Authorization'
|
||||
}
|
||||
assert len(docs['definitions']) == 127
|
||||
assert len(docs['definitions']) == 131
|
||||
|
|
|
|||
Reference in New Issue