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