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 sqlalchemy import create_engine\r
  11 from sqlalchemy.orm import sessionmaker, clear_mappers\r
  12 from tenacity import retry, stop_after_delay\r
  13 \r
  14 from o2ims.adapter.orm import metadata, start_o2ims_mappers\r
  15 from o2ims.adapter.clients.orm_stx import start_o2ims_stx_mappers\r
  16 from o2ims import config\r
  17 \r
  18 \r
  19 @pytest.fixture\r
  20 def in_memory_sqlite_db():\r
  21     engine = create_engine("sqlite:///:memory:")\r
  22     # engine = create_engine("sqlite:///:memory:", echo=True)\r
  23     metadata.create_all(engine)\r
  24     return engine\r
  25 \r
  26 \r
  27 @pytest.fixture\r
  28 def sqlite_session_factory(in_memory_sqlite_db):\r
  29     yield sessionmaker(bind=in_memory_sqlite_db)\r
  30 \r
  31 \r
  32 @pytest.fixture\r
  33 def mappers():\r
  34     start_o2ims_mappers()\r
  35     start_o2ims_stx_mappers()\r
  36     yield\r
  37     clear_mappers()\r
  38 \r
  39 @pytest.fixture\r
  40 def fake_stx_client():\r
  41     pass\r
  42 \r
  43 @retry(stop=stop_after_delay(10))\r
  44 def wait_for_postgres_to_come_up(engine):\r
  45     return engine.connect()\r
  46 \r
  47 \r
  48 @retry(stop=stop_after_delay(10))\r
  49 def wait_for_webapp_to_come_up():\r
  50     return requests.get(config.get_api_url())\r
  51 \r
  52 \r
  53 @retry(stop=stop_after_delay(10))\r
  54 def wait_for_redis_to_come_up():\r
  55     r = redis.Redis(**config.get_redis_host_and_port())\r
  56     return r.ping()\r
  57 \r
  58 \r
  59 @pytest.fixture(scope="session")\r
  60 def postgres_db():\r
  61     engine = create_engine(config.get_postgres_uri(), isolation_level="SERIALIZABLE")\r
  62     wait_for_postgres_to_come_up(engine)\r
  63     metadata.create_all(engine)\r
  64     return engine\r
  65 \r
  66 \r
  67 @pytest.fixture\r
  68 def postgres_session_factory(postgres_db):\r
  69     yield sessionmaker(bind=postgres_db)\r
  70 \r
  71 \r
  72 @pytest.fixture\r
  73 def postgres_session(postgres_session_factory):\r
  74     return postgres_session_factory()\r
  75 \r
  76 \r
  77 @pytest.fixture\r
  78 def restart_api():\r
  79     (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py").touch()\r
  80     time.sleep(0.5)\r
  81     wait_for_webapp_to_come_up()\r
  82 \r
  83 \r
  84 @pytest.fixture\r
  85 def restart_redis_pubsub():\r
  86     wait_for_redis_to_come_up()\r
  87     if not shutil.which("docker-compose"):\r
  88         print("skipping restart, assumes running in container")\r
  89         return\r
  90     subprocess.run(\r
  91         ["docker-compose", "restart", "-t", "0", "redis_pubsub"],\r
  92         check=True,\r
  93     )\r