1 # pylint: disable=redefined-outer-name
\r
5 from pathlib import Path
\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
14 from o2ims.adapter.orm import metadata, start_o2ims_mappers
\r
15 from o2ims import config
\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
27 def sqlite_session_factory(in_memory_sqlite_db):
\r
28 yield sessionmaker(bind=in_memory_sqlite_db)
\r
33 start_o2ims_mappers()
\r
38 @retry(stop=stop_after_delay(10))
\r
39 def wait_for_postgres_to_come_up(engine):
\r
40 return engine.connect()
\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
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
54 @pytest.fixture(scope="session")
\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
63 def postgres_session_factory(postgres_db):
\r
64 yield sessionmaker(bind=postgres_db)
\r
68 def postgres_session(postgres_session_factory):
\r
69 return postgres_session_factory()
\r
74 (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py").touch()
\r
76 wait_for_webapp_to_come_up()
\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
86 ["docker-compose", "restart", "-t", "0", "redis_pubsub"],
\r