tests/conftest.py

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