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.ocloud_route 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     bus = bootstrap(False, uow)\r
  42     configure_namespace(api, bus)\r
  43     client = app.test_client()\r
  44     return session, client\r
  45 \r
  46 \r
  47 @pytest.fixture\r
  48 def in_memory_sqlite_db():\r
  49     engine = create_engine("sqlite:///:memory:")\r
  50     # engine = create_engine("sqlite:///:memory:", echo=True)\r
  51     metadata.create_all(engine)\r
  52     return engine\r
  53 \r
  54 \r
  55 @pytest.fixture\r
  56 def sqlite_session_factory(in_memory_sqlite_db):\r
  57     yield sessionmaker(bind=in_memory_sqlite_db)\r
  58 \r
  59 \r
  60 @pytest.fixture\r
  61 def sqlite_uow(sqlite_session_factory):\r
  62     uow = unit_of_work.SqlAlchemyUnitOfWork(\r
  63         session_factory=sqlite_session_factory)\r
  64     # with uow:\r
  65     #     start_o2ims_mappers(uow.session.get_bind())\r
  66     #     uow.commit()\r
  67     yield uow\r
  68     # clear_mappers()\r
  69     with uow:\r
  70         engine = uow.session.get_bind()\r
  71         metadata.drop_all(engine)\r
  72 \r
  73 \r
  74 @pytest.fixture\r
  75 def sqlite_flask_uow(sqlite_uow):\r
  76     app = Flask(__name__)\r
  77     # app.config["TESTING"] = True\r
  78     api = Api(app)\r
  79     bus = bootstrap(False, sqlite_uow)\r
  80     configure_namespace(api, bus)\r
  81     yield app.test_client()\r
  82 \r
  83 \r
  84 @pytest.fixture\r
  85 def mappers():\r
  86     start_o2ims_mappers()\r
  87     # start_o2ims_stx_mappers()\r
  88     yield\r
  89     clear_mappers()\r
  90 \r
  91 \r
  92 @retry(stop=stop_after_delay(10))\r
  93 def wait_for_postgres_to_come_up(engine):\r
  94     return engine.connect()\r
  95 \r
  96 \r
  97 @retry(stop=stop_after_delay(10))\r
  98 def wait_for_webapp_to_come_up():\r
  99     return requests.get(config.get_api_url())\r
 100 \r
 101 \r
 102 @retry(stop=stop_after_delay(10))\r
 103 def wait_for_redis_to_come_up():\r
 104     r = redis.Redis(**config.get_redis_host_and_port())\r
 105     return r.ping()\r
 106 \r
 107 \r
 108 @pytest.fixture(scope="session")\r
 109 def postgres_db():\r
 110     engine = create_engine(config.get_postgres_uri(),\r
 111                            isolation_level="SERIALIZABLE")\r
 112     wait_for_postgres_to_come_up(engine)\r
 113     metadata.create_all(engine)\r
 114     return engine\r
 115 \r
 116 \r
 117 @pytest.fixture\r
 118 def postgres_session_factory(postgres_db):\r
 119     yield sessionmaker(bind=postgres_db)\r
 120 \r
 121 \r
 122 @pytest.fixture\r
 123 def postgres_session(postgres_session_factory):\r
 124     return postgres_session_factory()\r
 125 \r
 126 \r
 127 @pytest.fixture\r
 128 def postgres_uow(postgres_session_factory):\r
 129     uow = unit_of_work.SqlAlchemyUnitOfWork(\r
 130         session_factory=postgres_session_factory)\r
 131     yield uow\r
 132 \r
 133 \r
 134 @pytest.fixture\r
 135 def postgres_flask_uow(postgres_uow):\r
 136     app = Flask(__name__)\r
 137     api = Api(app)\r
 138     bus = bootstrap(False, postgres_uow)\r
 139     configure_namespace(api, bus)\r
 140     yield app.test_client()\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