devicehub-teal/tests/conftest.py

import io
import uuid
from contextlib import redirect_stdout
from datetime import datetime
from pathlib import Path

import boltons.urlutils
import ereuse_utils
import jwt
import pytest
import yaml
from decouple import config
from psycopg2 import IntegrityError
from sqlalchemy.exc import ProgrammingError

from ereuse_devicehub.client import Client, UserClient, UserClientFlask
from ereuse_devicehub.config import DevicehubConfig
from ereuse_devicehub.db import db
from ereuse_devicehub.devicehub import Devicehub
from ereuse_devicehub.inventory.views import devices
from ereuse_devicehub.labels.views import labels
from ereuse_devicehub.resources.agent.models import Person
from ereuse_devicehub.resources.enums import SessionType
from ereuse_devicehub.resources.tag import Tag
from ereuse_devicehub.resources.user.models import Session, User
from ereuse_devicehub.views import core
from ereuse_devicehub.workbench.views import workbench

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."""
P = config('JWT_PASS', '')


class TestConfig(DevicehubConfig):
    SQLALCHEMY_DATABASE_URI = 'postgresql://dhub:ereuse@localhost/dh_test'
    TESTING = True
    SERVER_NAME = 'localhost'
    TMP_SNAPSHOTS = '/tmp/snapshots'
    TMP_LIVES = '/tmp/lives'
    EMAIL_ADMIN = 'foo@foo.com'
    PATH_DOCUMENTS_STORAGE = '/tmp/trade_documents'
    JWT_PASS = config('JWT_PASS', '')


@pytest.fixture(scope='session')
def config():
    return TestConfig()


@pytest.fixture(scope='session')
def _app(config: TestConfig) -> Devicehub:
    # dh_config = DevicehubConfig()
    # config = TestConfig(dh_config)
    app = Devicehub(inventory='test', config=config, db=db)
    app.register_blueprint(core)
    app.register_blueprint(devices)
    app.register_blueprint(labels)
    app.register_blueprint(workbench)
    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:
    return Devicehub(inventory='test', config=config, db=db)


@pytest.fixture()
def app(request, _app: Devicehub) -> Devicehub:
    # More robust than 'yield'
    def _drop(*args, **kwargs):
        with _app.app_context():
            db.drop_all()

    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,
        )

    with _app.app_context():
        try:
            with redirect_stdout(io.StringIO()):
                _init()
        except (ProgrammingError, IntegrityError, AssertionError):
            print('Database was not correctly emptied. Re-empty and re-installing...')
            _drop()
            _init()

    request.addfinalizer(_drop)
    return _app


@pytest.fixture()
def client(app: Devicehub) -> Client:
    return app.test_client()


@pytest.fixture()
def app_context(app: Devicehub):
    with app.app_context():
        yield


@pytest.fixture()
def user(app: Devicehub) -> UserClient:
    """Gets a client with a logged-in dummy user."""
    with app.app_context():
        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()
        return client


@pytest.fixture()
def user3(app: Devicehub) -> UserClientFlask:
    """Gets a client with a logged-in dummy user."""
    with app.app_context():
        password = 'foo'
        user = create_user(password=password)
        client = UserClientFlask(app, user.email, password)
        return client


@pytest.fixture()
def user4(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 = UserClientFlask(app, user.email, password)
        return client


def create_user(email='foo@foo.com', password='foo') -> User:
    user = User(email=email, password=password)
    user.individuals.add(Person(name='Timmy'))
    session_external = Session(user=user, type=SessionType.External)
    session_internal = Session(user=user, type=SessionType.Internal)
    db.session.add(user)
    db.session.add(session_internal)
    db.session.add(session_external)
    db.session.commit()
    return user


@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


def json_encode(dev: str) -> dict:
    """Encode json."""
    data = {"type": "Snapshot"}
    data['data'] = jwt.encode(
        dev, P, algorithm="HS256", json_encoder=ereuse_utils.JSONEncoder
    )

    return data


def yaml2json(name: str) -> dict:
    """Opens and parses a YAML file from the ``files`` subdir."""
    with Path(__file__).parent.joinpath('files').joinpath(name + '.yaml').open() as f:
        return yaml.load(f)


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():
        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