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