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