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_routes\r
  22 from o2ims.views.ocloud_route import configure_namespace\r
  23 from o2ims.bootstrap import bootstrap\r
  24 \r
  25 \r
  26 @pytest.fixture\r
  27 def mock_uow():\r
  28     session = MagicMock()\r
  29     uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory=session)\r
  30     return session, uow\r
  31 \r
  32 \r
  33 @pytest.fixture\r
  34 def mock_flask_uow(mock_uow):\r
  35     session, uow = mock_uow\r
  36     app = Flask(__name__)\r
  37     # app.config["TESTING"] = True\r
  38     api = Api(app)\r
  39     bus = bootstrap(False, uow)\r
  40     # configure_routes(app, bus)\r
  41     configure_namespace(api, bus)\r
  42     client = app.test_client()\r
  43     return session, client\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     bus = bootstrap(False, sqlite_uow)\r
  79     # configure_routes(app, bus)\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_routes(app, bus)\r
 140     configure_namespace(api, bus)\r
 141     yield app.test_client()\r
 142 \r
 143 \r
 144 @pytest.fixture\r
 145 def restart_api():\r
 146     (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py")\\r
 147         .touch()\r
 148     time.sleep(0.5)\r
 149     wait_for_webapp_to_come_up()\r
 150 \r
 151 \r
 152 @pytest.fixture\r
 153 def restart_redis_pubsub():\r
 154     wait_for_redis_to_come_up()\r
 155     if not shutil.which("docker-compose"):\r
 156         print("skipping restart, assumes running in container")\r
 157         return\r
 158     subprocess.run(\r
 159         ["docker-compose", "restart", "-t", "0", "redis_pubsub"],\r
 160         check=True,\r
 161     )\r