devicehub-teal/tests/conftest.py

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

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

from ereuse_devicehub import ereuse_utils
from ereuse_devicehub.api.views import api
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.mail.flask_mail import Mail
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', '')
    MAIL_SUPPRESS_SEND = True


@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(api)
    app.register_blueprint(workbench)
    app.config["SQLALCHEMY_RECORD_QUERIES"] = True
    app.config['PROFILE'] = True
    # app.wsgi_app = ProfilerMiddleware(app.wsgi_app, restrictions=[30])
    mail = Mail(app)
    app.mail = mail
    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_json(name):
    with Path(__file__).parent.joinpath('files').joinpath(name).open() as f:
        return json.loads(f.read())


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