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