2018-08-26 17:04:42 +00:00
|
|
|
import io
|
2019-01-23 15:55:04 +00:00
|
|
|
import uuid
|
2018-08-26 17:04:42 +00:00
|
|
|
from contextlib import redirect_stdout
|
|
|
|
from datetime import datetime
|
2018-04-27 17:16:43 +00:00
|
|
|
from pathlib import Path
|
|
|
|
|
2019-01-23 15:55:04 +00:00
|
|
|
import boltons.urlutils
|
2022-03-14 12:32:39 +00:00
|
|
|
import ereuse_utils
|
|
|
|
import jwt
|
2018-04-10 15:06:39 +00:00
|
|
|
import pytest
|
2018-04-27 17:16:43 +00:00
|
|
|
import yaml
|
2022-03-14 12:32:39 +00:00
|
|
|
from decouple import config
|
2018-10-05 12:35:51 +00:00
|
|
|
from psycopg2 import IntegrityError
|
2018-07-14 14:41:22 +00:00
|
|
|
from sqlalchemy.exc import ProgrammingError
|
2018-04-10 15:06:39 +00:00
|
|
|
|
2018-04-27 17:16:43 +00:00
|
|
|
from ereuse_devicehub.client import Client, UserClient
|
2018-04-10 15:06:39 +00:00
|
|
|
from ereuse_devicehub.config import DevicehubConfig
|
|
|
|
from ereuse_devicehub.db import db
|
|
|
|
from ereuse_devicehub.devicehub import Devicehub
|
2022-03-14 12:32:39 +00:00
|
|
|
from ereuse_devicehub.inventory.views import devices
|
2018-08-03 16:15:08 +00:00
|
|
|
from ereuse_devicehub.resources.agent.models import Person
|
2021-04-15 19:16:32 +00:00
|
|
|
from ereuse_devicehub.resources.enums import SessionType
|
2022-03-14 12:32:39 +00:00
|
|
|
from ereuse_devicehub.resources.tag import Tag
|
|
|
|
from ereuse_devicehub.resources.user.models import Session, User
|
|
|
|
from ereuse_devicehub.views import core
|
2018-04-10 15:06:39 +00:00
|
|
|
|
2018-08-26 17:04:42 +00:00
|
|
|
STARTT = datetime(year=2000, month=1, day=1, hour=1)
|
|
|
|
"""A dummy starting time to use in tests."""
|
|
|
|
ENDT = datetime(year=2000, month=1, day=1, hour=2)
|
|
|
|
"""A dummy ending time to use in tests."""
|
|
|
|
T = {'start_time': STARTT, 'end_time': ENDT}
|
|
|
|
"""A dummy start_time/end_time to use as function keywords."""
|
2021-07-05 13:04:44 +00:00
|
|
|
P = config('JWT_PASS', '')
|
2018-08-26 17:04:42 +00:00
|
|
|
|
2018-04-10 15:06:39 +00:00
|
|
|
|
|
|
|
class TestConfig(DevicehubConfig):
|
2018-07-08 13:00:28 +00:00
|
|
|
SQLALCHEMY_DATABASE_URI = 'postgresql://dhub:ereuse@localhost/dh_test'
|
2018-04-27 17:16:43 +00:00
|
|
|
TESTING = True
|
2018-10-05 15:13:23 +00:00
|
|
|
SERVER_NAME = 'localhost'
|
2020-10-13 13:37:21 +00:00
|
|
|
TMP_SNAPSHOTS = '/tmp/snapshots'
|
2020-12-29 16:37:13 +00:00
|
|
|
TMP_LIVES = '/tmp/lives'
|
2021-02-22 20:15:25 +00:00
|
|
|
EMAIL_ADMIN = 'foo@foo.com'
|
2021-05-12 09:47:03 +00:00
|
|
|
PATH_DOCUMENTS_STORAGE = '/tmp/trade_documents'
|
2021-07-05 13:04:44 +00:00
|
|
|
JWT_PASS = config('JWT_PASS', '')
|
2018-04-10 15:06:39 +00:00
|
|
|
|
|
|
|
|
2018-10-05 12:35:51 +00:00
|
|
|
@pytest.fixture(scope='session')
|
2018-04-10 15:06:39 +00:00
|
|
|
def config():
|
|
|
|
return TestConfig()
|
|
|
|
|
|
|
|
|
2018-10-05 12:35:51 +00:00
|
|
|
@pytest.fixture(scope='session')
|
2018-04-27 17:16:43 +00:00
|
|
|
def _app(config: TestConfig) -> Devicehub:
|
2022-03-14 12:32:39 +00:00
|
|
|
# dh_config = DevicehubConfig()
|
|
|
|
# config = TestConfig(dh_config)
|
|
|
|
app = Devicehub(inventory='test', config=config, db=db)
|
|
|
|
app.register_blueprint(core)
|
|
|
|
app.register_blueprint(devices)
|
|
|
|
app.config["SQLALCHEMY_RECORD_QUERIES"] = True
|
|
|
|
app.config['PROFILE'] = True
|
|
|
|
# app.wsgi_app = ProfilerMiddleware(app.wsgi_app, restrictions=[30])
|
|
|
|
return app
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
|
|
def _app2(config: TestConfig) -> Devicehub:
|
2019-01-23 15:55:04 +00:00
|
|
|
return Devicehub(inventory='test', config=config, db=db)
|
2018-04-27 17:16:43 +00:00
|
|
|
|
|
|
|
|
2018-04-10 15:06:39 +00:00
|
|
|
@pytest.fixture()
|
2018-04-27 17:16:43 +00:00
|
|
|
def app(request, _app: Devicehub) -> Devicehub:
|
|
|
|
# More robust than 'yield'
|
2018-07-14 14:41:22 +00:00
|
|
|
def _drop(*args, **kwargs):
|
|
|
|
with _app.app_context():
|
|
|
|
db.drop_all()
|
|
|
|
|
2019-01-23 15:55:04 +00:00
|
|
|
def _init():
|
2022-03-14 12:32:39 +00:00
|
|
|
_app.init_db(
|
|
|
|
name='Test Inventory',
|
|
|
|
org_name='FooOrg',
|
|
|
|
org_id='foo-org-id',
|
|
|
|
tag_url=boltons.urlutils.URL('https://example.com'),
|
|
|
|
tag_token=uuid.UUID('52dacef0-6bcb-4919-bfed-f10d2c96ecee'),
|
|
|
|
erase=False,
|
|
|
|
common=True,
|
|
|
|
)
|
2019-01-23 15:55:04 +00:00
|
|
|
|
2018-07-14 14:41:22 +00:00
|
|
|
with _app.app_context():
|
|
|
|
try:
|
2018-08-26 17:04:42 +00:00
|
|
|
with redirect_stdout(io.StringIO()):
|
2019-01-23 15:55:04 +00:00
|
|
|
_init()
|
2019-02-11 20:34:45 +00:00
|
|
|
except (ProgrammingError, IntegrityError, AssertionError):
|
2018-07-14 14:41:22 +00:00
|
|
|
print('Database was not correctly emptied. Re-empty and re-installing...')
|
|
|
|
_drop()
|
2019-01-23 15:55:04 +00:00
|
|
|
_init()
|
2018-07-14 14:41:22 +00:00
|
|
|
|
|
|
|
request.addfinalizer(_drop)
|
2018-04-27 17:16:43 +00:00
|
|
|
return _app
|
2018-04-10 15:06:39 +00:00
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture()
|
|
|
|
def client(app: Devicehub) -> Client:
|
|
|
|
return app.test_client()
|
2018-04-27 17:16:43 +00:00
|
|
|
|
|
|
|
|
2018-04-30 17:58:19 +00:00
|
|
|
@pytest.fixture()
|
|
|
|
def app_context(app: Devicehub):
|
|
|
|
with app.app_context():
|
|
|
|
yield
|
|
|
|
|
|
|
|
|
2018-04-27 17:16:43 +00:00
|
|
|
@pytest.fixture()
|
|
|
|
def user(app: Devicehub) -> UserClient:
|
|
|
|
"""Gets a client with a logged-in dummy user."""
|
|
|
|
with app.app_context():
|
2018-06-20 21:18:15 +00:00
|
|
|
password = 'foo'
|
|
|
|
user = create_user(password=password)
|
2022-03-14 12:32:39 +00:00
|
|
|
client = UserClient(
|
|
|
|
app, user.email, password, response_wrapper=app.response_class
|
|
|
|
)
|
2020-08-05 10:22:06 +00:00
|
|
|
client.login()
|
|
|
|
return client
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture()
|
|
|
|
def user2(app: Devicehub) -> UserClient:
|
|
|
|
"""Gets a client with a logged-in dummy user."""
|
|
|
|
with app.app_context():
|
|
|
|
password = 'foo'
|
|
|
|
email = 'foo2@foo.com'
|
|
|
|
user = create_user(email=email, password=password)
|
2022-03-14 12:32:39 +00:00
|
|
|
client = UserClient(
|
|
|
|
app, user.email, password, response_wrapper=app.response_class
|
|
|
|
)
|
2018-08-03 16:15:08 +00:00
|
|
|
client.login()
|
2018-04-27 17:16:43 +00:00
|
|
|
return client
|
|
|
|
|
|
|
|
|
|
|
|
def create_user(email='foo@foo.com', password='foo') -> User:
|
|
|
|
user = User(email=email, password=password)
|
2018-08-03 16:15:08 +00:00
|
|
|
user.individuals.add(Person(name='Timmy'))
|
2021-04-15 19:16:32 +00:00
|
|
|
session_external = Session(user=user, type=SessionType.External)
|
|
|
|
session_internal = Session(user=user, type=SessionType.Internal)
|
2018-04-27 17:16:43 +00:00
|
|
|
db.session.add(user)
|
2021-04-15 19:16:32 +00:00
|
|
|
db.session.add(session_internal)
|
|
|
|
db.session.add(session_external)
|
2018-04-27 17:16:43 +00:00
|
|
|
db.session.commit()
|
|
|
|
return user
|
|
|
|
|
|
|
|
|
2018-04-30 17:58:19 +00:00
|
|
|
@pytest.fixture()
|
|
|
|
def auth_app_context(app: Devicehub):
|
|
|
|
"""Creates an app context with a set user."""
|
|
|
|
with app.app_context():
|
|
|
|
user = create_user()
|
|
|
|
|
|
|
|
class Auth: # Mock
|
|
|
|
username = user.token
|
|
|
|
password = ''
|
|
|
|
|
|
|
|
app.auth.perform_auth(Auth())
|
2018-10-11 09:22:59 +00:00
|
|
|
yield app
|
2018-04-30 17:58:19 +00:00
|
|
|
|
|
|
|
|
2021-07-02 13:25:19 +00:00
|
|
|
def json_encode(dev: str) -> dict:
|
|
|
|
"""Encode json."""
|
2021-07-05 10:31:43 +00:00
|
|
|
data = {"type": "Snapshot"}
|
2022-03-14 12:32:39 +00:00
|
|
|
data['data'] = jwt.encode(
|
|
|
|
dev, P, algorithm="HS256", json_encoder=ereuse_utils.JSONEncoder
|
2021-07-02 13:25:19 +00:00
|
|
|
)
|
|
|
|
|
2021-07-05 10:31:43 +00:00
|
|
|
return data
|
|
|
|
|
|
|
|
|
2021-07-02 13:25:19 +00:00
|
|
|
def yaml2json(name: str) -> dict:
|
2018-05-30 10:49:40 +00:00
|
|
|
"""Opens and parses a YAML file from the ``files`` subdir."""
|
2018-04-27 17:16:43 +00:00
|
|
|
with Path(__file__).parent.joinpath('files').joinpath(name + '.yaml').open() as f:
|
|
|
|
return yaml.load(f)
|
2018-05-30 10:49:40 +00:00
|
|
|
|
|
|
|
|
2021-07-02 13:25:19 +00:00
|
|
|
def file(name: str) -> dict:
|
|
|
|
"""Opens and parses a YAML file from the ``files`` subdir. And decode"""
|
|
|
|
return json_encode(yaml2json(name))
|
|
|
|
|
|
|
|
|
2020-10-15 11:19:47 +00:00
|
|
|
def file_workbench(name: str) -> dict:
|
|
|
|
"""Opens and parses a YAML file from the ``files`` subdir."""
|
2022-03-14 12:32:39 +00:00
|
|
|
with Path(__file__).parent.joinpath('workbench_files').joinpath(
|
|
|
|
name + '.json'
|
|
|
|
).open() as f:
|
2020-10-15 11:19:47 +00:00
|
|
|
return yaml.load(f)
|
|
|
|
|
|
|
|
|
2018-05-30 10:49:40 +00:00
|
|
|
@pytest.fixture()
|
|
|
|
def tag_id(app: Devicehub) -> str:
|
|
|
|
"""Creates a tag and returns its id."""
|
|
|
|
with app.app_context():
|
2020-07-23 05:56:51 +00:00
|
|
|
if User.query.count():
|
|
|
|
user = User.query.one()
|
|
|
|
else:
|
|
|
|
user = create_user()
|
2020-07-22 16:40:27 +00:00
|
|
|
t = Tag(id='foo', owner_id=user.id)
|
2018-05-30 10:49:40 +00:00
|
|
|
db.session.add(t)
|
|
|
|
db.session.commit()
|
|
|
|
return t.id
|