This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
devicehub-teal/tests/conftest.py

181 lines
5.2 KiB
Python
Raw Normal View History

import io
2019-01-23 15:55:04 +00:00
import uuid
2021-07-02 13:25:19 +00:00
import jwt
import ereuse_utils
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
2018-04-10 15:06:39 +00:00
import pytest
2018-04-27 17:16:43 +00:00
import yaml
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
from ereuse_devicehub.resources.agent.models import Person
from ereuse_devicehub.resources.tag import Tag
2018-04-27 17:16:43 +00:00
from ereuse_devicehub.resources.user.models import User
2021-04-15 19:16:32 +00:00
from ereuse_devicehub.resources.user.models import Session
from ereuse_devicehub.resources.enums import SessionType
2018-04-10 15:06:39 +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-02 13:25:19 +00:00
P = '7KU4ZzsEfe'
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
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'
2018-04-10 15:06:39 +00:00
@pytest.fixture(scope='session')
2018-04-10 15:06:39 +00:00
def config():
return TestConfig()
@pytest.fixture(scope='session')
2018-04-27 17:16:43 +00:00
def _app(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():
_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)
2018-07-14 14:41:22 +00:00
with _app.app_context():
try:
with redirect_stdout(io.StringIO()):
2019-01-23 15:55:04 +00:00
_init()
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)
client = UserClient(app, user.email, password, response_wrapper=app.response_class)
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)
client = UserClient(app, user.email, password, response_wrapper=app.response_class)
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)
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())
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."""
return jwt.encode(dev,
P,
algorithm="HS256",
json_encoder=ereuse_utils.JSONEncoder
)
def yaml2json(name: str) -> dict:
"""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)
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))
def file_workbench(name: str) -> dict:
"""Opens and parses a YAML file from the ``files`` subdir."""
with Path(__file__).parent.joinpath('workbench_files').joinpath(name + '.json').open() as f:
return yaml.load(f)
@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()
t = Tag(id='foo', owner_id=user.id)
db.session.add(t)
db.session.commit()
return t.id