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

return jwt.encode to snapshot code

Browse source
This commit is contained in:
Cayo Puigdefabregas 2022-06-07 15:24:45 +02:00
parent eb66baf842
commit 99d0a61e87
1 changed files with 122 additions and 83 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

153
ereuse_devicehub/dummy/dummy.py
View file

@ -1,41 +1,37 @@
import itertools import itertools
import json import json
import jwt
from pathlib import Path from pathlib import Path
from typing import Set from typing import Set
import click import click
import click_spinner import click_spinner
import ereuse_utils.cli import ereuse_utils.cli
import jwt
import yaml import yaml
from ereuse_utils.test import ANY from ereuse_utils.test import ANY
from ereuse_devicehub.client import UserClient from ereuse_devicehub.client import UserClient
from ereuse_devicehub.db import db from ereuse_devicehub.db import db
from ereuse_devicehub.parser.models import SnapshotsLog
from ereuse_devicehub.resources.action import models as m from ereuse_devicehub.resources.action import models as m
from ereuse_devicehub.resources.agent.models import Person from ereuse_devicehub.resources.agent.models import Person
from ereuse_devicehub.resources.device.models import Device from ereuse_devicehub.resources.device.models import Device
from ereuse_devicehub.resources.enums import SessionType
from ereuse_devicehub.resources.lot.models import Lot from ereuse_devicehub.resources.lot.models import Lot
from ereuse_devicehub.resources.tag.model import Tag from ereuse_devicehub.resources.tag.model import Tag
from ereuse_devicehub.resources.user import User from ereuse_devicehub.resources.user import User
from ereuse_devicehub.resources.user.models import Session from ereuse_devicehub.resources.user.models import Session
from ereuse_devicehub.resources.enums import SessionType
from ereuse_devicehub.parser.models import SnapshotsLog
class Dummy: class Dummy:
TAGS = ( TAGS = ('tag1', 'tag2', 'tag3')
'tag1',
'tag2',
'tag3'
)
"""Tags to create.""" """Tags to create."""
ET = ( ET = (
('DT-AAAAA', 'A0000000000001'), ('DT-AAAAA', 'A0000000000001'),
('DT-BBBBB', 'A0000000000002'), ('DT-BBBBB', 'A0000000000002'),
('DT-CCCCC', 'A0000000000003'), ('DT-CCCCC', 'A0000000000003'),
('DT-BRRAB', '04970DA2A15984'), ('DT-BRRAB', '04970DA2A15984'),
('DT-XXXXX', '04e4bc5af95980') ('DT-XXXXX', '04e4bc5af95980'),
) )
"""eTags to create.""" """eTags to create."""
ORG = 'eReuse.org CAT', '-t', 'G-60437761', '-c', 'ES' ORG = 'eReuse.org CAT', '-t', 'G-60437761', '-c', 'ES'
@ -44,27 +40,33 @@ class Dummy:
def __init__(self, app) -> None: def __init__(self, app) -> None:
super().__init__() super().__init__()
self.app = app self.app = app
self.app.cli.command('dummy', short_help='Creates dummy devices and users.')(self.run) self.app.cli.command('dummy', short_help='Creates dummy devices and users.')(
self.run
)
@click.option('--tag-url', '-tu', @click.option(
'--tag-url',
'-tu',
type=ereuse_utils.cli.URL(scheme=True, host=True, path=False), type=ereuse_utils.cli.URL(scheme=True, host=True, path=False),
default='http://localhost:8081', default='http://localhost:8081',
help='The base url (scheme and host) of the tag provider.') help='The base url (scheme and host) of the tag provider.',
@click.option('--tag-token', '-tt', )
@click.option(
'--tag-token',
'-tt',
type=click.UUID, type=click.UUID,
default='899c794e-1737-4cea-9232-fdc507ab7106', default='899c794e-1737-4cea-9232-fdc507ab7106',
help='The token provided by the tag provider. It is an UUID.') help='The token provided by the tag provider. It is an UUID.',
@click.confirmation_option(prompt='This command (re)creates the DB from scratch.' )
'Do you want to continue?') @click.confirmation_option(
prompt='This command (re)creates the DB from scratch.'
'Do you want to continue?'
)
def run(self, tag_url, tag_token): def run(self, tag_url, tag_token):
runner = self.app.test_cli_runner() runner = self.app.test_cli_runner()
self.app.init_db('Dummy', self.app.init_db(
'ACME', 'Dummy', 'ACME', 'acme-id', tag_url, tag_token, erase=True, common=True
'acme-id', )
tag_url,
tag_token,
erase=True,
common=True)
print('Creating stuff...'.ljust(30), end='') print('Creating stuff...'.ljust(30), end='')
assert SnapshotsLog.query.filter().all() == [] assert SnapshotsLog.query.filter().all() == []
with click_spinner.spinner(): with click_spinner.spinner():
@ -79,16 +81,31 @@ class Dummy:
for id in self.TAGS: for id in self.TAGS:
user1.post({'id': id}, res=Tag) user1.post({'id': id}, res=Tag)
for id, sec in self.ET: for id, sec in self.ET:
runner.invoke('tag', 'add', id, runner.invoke(
'-p', 'https://t.devicetag.io', 'tag',
'-s', sec, 'add',
'-u', user1.user["id"], id,
'-o', org_id) '-p',
'https://t.devicetag.io',
'-s',
sec,
'-u',
user1.user["id"],
'-o',
org_id,
)
# create tag for pc-laudem # create tag for pc-laudem
runner.invoke('tag', 'add', 'tagA', runner.invoke(
'-p', 'https://t.devicetag.io', 'tag',
'-u', user1.user["id"], 'add',
'-s', 'tagA-secondary') 'tagA',
'-p',
'https://t.devicetag.io',
'-u',
user1.user["id"],
'-s',
'tagA-secondary',
)
files = tuple(Path(__file__).parent.joinpath('files').iterdir()) files = tuple(Path(__file__).parent.joinpath('files').iterdir())
print('done.') print('done.')
sample_pc = None # We treat this one as a special sample for demonstrations sample_pc = None # We treat this one as a special sample for demonstrations
@ -97,28 +114,39 @@ class Dummy:
for path in bar: for path in bar:
with path.open() as f: with path.open() as f:
snapshot = yaml.load(f) snapshot = yaml.load(f)
# s, _ = user1.post(res=m.Snapshot, data=self.json_encode(snapshot)) s, _ = user1.post(res=m.Snapshot, data=self.json_encode(snapshot))
s, _ = user1.post(res=m.Snapshot, data=snapshot)
if s.get('uuid', None) == 'ec23c11b-80b6-42cd-ac5c-73ba7acddbc4': if s.get('uuid', None) == 'ec23c11b-80b6-42cd-ac5c-73ba7acddbc4':
sample_pc = s['device']['id'] sample_pc = s['device']['id']
sample_pc_devicehub_id = s['device']['devicehubID'] sample_pc_devicehub_id = s['device']['devicehubID']
else: else:
pcs.add(s['device']['id']) pcs.add(s['device']['id'])
if s.get('uuid', None) == 'de4f495e-c58b-40e1-a33e-46ab5e84767e': # oreo if (
s.get('uuid', None) == 'de4f495e-c58b-40e1-a33e-46ab5e84767e'
): # oreo
# Make one hdd ErasePhysical # Make one hdd ErasePhysical
hdd = next(hdd for hdd in s['components'] if hdd['type'] == 'HardDrive') hdd = next(
user1.post({'type': 'ErasePhysical', 'method': 'Shred', 'device': hdd['id']}, hdd for hdd in s['components'] if hdd['type'] == 'HardDrive'
res=m.Action) )
user1.post(
{
'type': 'ErasePhysical',
'method': 'Shred',
'device': hdd['id'],
},
res=m.Action,
)
assert sample_pc assert sample_pc
print('PC sample is', sample_pc) print('PC sample is', sample_pc)
# Link tags and eTags # Link tags and eTags
for tag, pc in zip((self.TAGS[1], self.TAGS[2], self.ET[0][0], self.ET[1][1]), pcs): for tag, pc in zip(
(self.TAGS[1], self.TAGS[2], self.ET[0][0], self.ET[1][1]), pcs
):
user1.put({}, res=Tag, item='{}/device/{}'.format(tag, pc), status=204) user1.put({}, res=Tag, item='{}/device/{}'.format(tag, pc), status=204)
# Perform generic actions # Perform generic actions
for pc, model in zip(pcs, for pc, model in zip(
{m.ToRepair, m.Repair, m.ToPrepare, m.Ready, m.ToPrepare, pcs, {m.ToRepair, m.Repair, m.ToPrepare, m.Ready, m.ToPrepare, m.Prepare}
m.Prepare}): ):
user1.post({'type': model.t, 'devices': [pc]}, res=m.Action) user1.post({'type': model.t, 'devices': [pc]}, res=m.Action)
# Perform a Sell to several devices # Perform a Sell to several devices
@ -138,26 +166,34 @@ class Dummy:
lot_user4, _ = user4.post({'name': 'LoteJordi'}, res=Lot) lot_user4, _ = user4.post({'name': 'LoteJordi'}, res=Lot)
lot, _ = user1.post({}, lot, _ = user1.post(
{},
res=Lot, res=Lot,
item='{}/devices'.format(lot_user['id']), item='{}/devices'.format(lot_user['id']),
query=[('id', pc) for pc in itertools.islice(pcs, 1, 4)]) query=[('id', pc) for pc in itertools.islice(pcs, 1, 4)],
)
# assert len(lot['devices']) # assert len(lot['devices'])
lot2, _ = user2.post({}, lot2, _ = user2.post(
{},
res=Lot, res=Lot,
item='{}/devices'.format(lot_user2['id']), item='{}/devices'.format(lot_user2['id']),
query=[('id', pc) for pc in itertools.islice(pcs, 4, 6)]) query=[('id', pc) for pc in itertools.islice(pcs, 4, 6)],
)
lot3, _ = user3.post({}, lot3, _ = user3.post(
{},
res=Lot, res=Lot,
item='{}/devices'.format(lot_user3['id']), item='{}/devices'.format(lot_user3['id']),
query=[('id', pc) for pc in itertools.islice(pcs, 11, 14)]) query=[('id', pc) for pc in itertools.islice(pcs, 11, 14)],
)
lot4, _ = user4.post({}, lot4, _ = user4.post(
{},
res=Lot, res=Lot,
item='{}/devices'.format(lot_user4['id']), item='{}/devices'.format(lot_user4['id']),
query=[('id', pc) for pc in itertools.islice(pcs, 14, 16)]) query=[('id', pc) for pc in itertools.islice(pcs, 14, 16)],
)
# Keep this at the bottom # Keep this at the bottom
inventory, _ = user1.get(res=Device) inventory, _ = user1.get(res=Device)
@ -174,8 +210,10 @@ class Dummy:
user1.post({'type': m.ToPrepare.t, 'devices': [sample_pc]}, res=m.Action) user1.post({'type': m.ToPrepare.t, 'devices': [sample_pc]}, res=m.Action)
user1.post({'type': m.Prepare.t, 'devices': [sample_pc]}, res=m.Action) user1.post({'type': m.Prepare.t, 'devices': [sample_pc]}, res=m.Action)
user1.post({'type': m.Ready.t, 'devices': [sample_pc]}, res=m.Action) user1.post({'type': m.Ready.t, 'devices': [sample_pc]}, res=m.Action)
user1.post({'type': m.Price.t, 'device': sample_pc, 'currency': 'EUR', 'price': 85}, user1.post(
res=m.Action) {'type': m.Price.t, 'device': sample_pc, 'currency': 'EUR', 'price': 85},
res=m.Action,
)
# todo test reserve # todo test reserve
# user1.post( # Sell device # user1.post( # Sell device
@ -206,19 +244,20 @@ class Dummy:
db.session.add(session_external) db.session.add(session_external)
db.session.commit() db.session.commit()
client = UserClient(self.app, user.email, password, client = UserClient(
response_wrapper=self.app.response_class) self.app, user.email, password, response_wrapper=self.app.response_class
)
client.login() client.login()
return client return client
def json_encode(self, dev: str) -> dict: def json_encode(self, dev: str) -> dict:
"""Encode json.""" """Encode json."""
data = {"type": "Snapshot"} data = {"type": "Snapshot"}
data['data'] = jwt.encode(dev, data['data'] = jwt.encode(
dev,
self.app.config['JWT_PASS'], self.app.config['JWT_PASS'],
algorithm="HS256", algorithm="HS256",
json_encoder=ereuse_utils.JSONEncoder json_encoder=ereuse_utils.JSONEncoder,
) )
return data return data