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