tests/conftest.py

   1 # pylint: disable=redefined-outer-name
   2 import shutil
   3 import subprocess
   4 import time
   5 from pathlib import Path
   6
   7 import pytest
   8 import redis
   9 import requests
  10 from flask import Flask
  11 from flask_restx import Api
  12 from sqlalchemy import create_engine
  13 from sqlalchemy.orm import sessionmaker, clear_mappers
  14 from tenacity import retry, stop_after_delay
  15 from unittest.mock import MagicMock
  16
  17 from o2common.config import config
  18
  19 from o2ims.adapter.orm import metadata, start_o2ims_mappers
  20 # from o2ims.adapter.clients.orm_stx import start_o2ims_stx_mappers
  21
  22 from o2app.adapter import unit_of_work
  23 from o2ims.views import configure_namespace
  24
  25 from o2app.bootstrap import bootstrap
  26
  27
  28 @pytest.fixture
  29 def mock_uow():
  30     session = MagicMock()
  31     uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory=session)
  32     return session, uow
  33
  34
  35 @pytest.fixture
  36 def mock_flask_uow(mock_uow):
  37     session, uow = mock_uow
  38     app = Flask(__name__)
  39     app.config["TESTING"] = True
  40     api = Api(app)
  41     bootstrap(False, uow)
  42     configure_namespace(api)
  43     return session, app
  44
  45
  46 @pytest.fixture
  47 def in_memory_sqlite_db():
  48     engine = create_engine("sqlite:///:memory:")
  49     # engine = create_engine("sqlite:///:memory:", echo=True)
  50     metadata.create_all(engine)
  51     return engine
  52
  53
  54 @pytest.fixture
  55 def sqlite_session_factory(in_memory_sqlite_db):
  56     yield sessionmaker(bind=in_memory_sqlite_db)
  57
  58
  59 @pytest.fixture
  60 def sqlite_uow(sqlite_session_factory):
  61     uow = unit_of_work.SqlAlchemyUnitOfWork(
  62         session_factory=sqlite_session_factory)
  63     # with uow:
  64     #     start_o2ims_mappers(uow.session.get_bind())
  65     #     uow.commit()
  66     yield uow
  67     # clear_mappers()
  68     with uow:
  69         engine = uow.session.get_bind()
  70         metadata.drop_all(engine)
  71
  72
  73 @pytest.fixture
  74 def sqlite_flask_uow(sqlite_uow):
  75     app = Flask(__name__)
  76     app.config["TESTING"] = True
  77     api = Api(app)
  78     bootstrap(False, sqlite_uow)
  79     configure_namespace(api)
  80     yield sqlite_uow, app
  81
  82
  83 @pytest.fixture
  84 def mappers():
  85     start_o2ims_mappers()
  86     # start_o2ims_stx_mappers()
  87     yield
  88     clear_mappers()
  89
  90
  91 @retry(stop=stop_after_delay(10))
  92 def wait_for_postgres_to_come_up(engine):
  93     return engine.connect()
  94
  95
  96 @retry(stop=stop_after_delay(10))
  97 def wait_for_webapp_to_come_up():
  98     return requests.get(config.get_api_url())
  99
 100
 101 @retry(stop=stop_after_delay(10))
 102 def wait_for_redis_to_come_up():
 103     r = redis.Redis(**config.get_redis_host_and_port())
 104     return r.ping()
 105
 106
 107 @pytest.fixture(scope="session")
 108 def postgres_db():
 109     engine = create_engine(config.get_postgres_uri(),
 110                            isolation_level="SERIALIZABLE")
 111     wait_for_postgres_to_come_up(engine)
 112     metadata.create_all(engine)
 113     return engine
 114
 115
 116 @pytest.fixture
 117 def postgres_session_factory(postgres_db):
 118     yield sessionmaker(bind=postgres_db)
 119
 120
 121 @pytest.fixture
 122 def postgres_session(postgres_session_factory):
 123     return postgres_session_factory()
 124
 125
 126 @pytest.fixture
 127 def postgres_uow(postgres_session_factory):
 128     uow = unit_of_work.SqlAlchemyUnitOfWork(
 129         session_factory=postgres_session_factory)
 130     yield uow
 131
 132
 133 @pytest.fixture
 134 def postgres_flask_uow(postgres_uow):
 135     app = Flask(__name__)
 136     app.config["TESTING"] = True
 137     api = Api(app)
 138     bootstrap(False, postgres_uow)
 139     configure_namespace(api)
 140     yield postgres_uow, app
 141
 142
 143 @pytest.fixture
 144 def restart_api():
 145     (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py")\
 146         .touch()
 147     time.sleep(0.5)
 148     wait_for_webapp_to_come_up()
 149
 150
 151 @pytest.fixture
 152 def restart_redis_pubsub():
 153     wait_for_redis_to_come_up()
 154     if not shutil.which("docker-compose"):
 155         print("skipping restart, assumes running in container")
 156         return
 157     subprocess.run(
 158         ["docker-compose", "restart", "-t", "0", "redis_pubsub"],
 159         check=True,
 160     )