1 # pylint: disable=redefined-outer-name
5 from pathlib import Path
10 from flask import Flask
11 from flask_restx import Api
12 from sqlalchemy import create_engine
13 from sqlalchemy.orm import sessionmaker, clear_mappers
14 from tenacity import retry, stop_after_delay
15 from unittest.mock import MagicMock
17 from o2common.config import config
19 from o2ims.adapter.orm import metadata, start_o2ims_mappers
20 # from o2ims.adapter.clients.orm_stx import start_o2ims_stx_mappers
22 from o2app.adapter import unit_of_work
23 from o2ims.views import configure_namespace
25 from o2app.bootstrap import bootstrap
31 uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory=session)
36 def mock_flask_uow(mock_uow):
37 session, uow = mock_uow
39 app.config["TESTING"] = True
42 configure_namespace(api)
47 def in_memory_sqlite_db():
48 engine = create_engine("sqlite:///:memory:")
49 # engine = create_engine("sqlite:///:memory:", echo=True)
50 metadata.create_all(engine)
55 def sqlite_session_factory(in_memory_sqlite_db):
56 yield sessionmaker(bind=in_memory_sqlite_db)
60 def sqlite_uow(sqlite_session_factory):
61 uow = unit_of_work.SqlAlchemyUnitOfWork(
62 session_factory=sqlite_session_factory)
64 # start_o2ims_mappers(uow.session.get_bind())
69 engine = uow.session.get_bind()
70 metadata.drop_all(engine)
74 def sqlite_flask_uow(sqlite_uow):
76 app.config["TESTING"] = True
78 bootstrap(False, sqlite_uow)
79 configure_namespace(api)
86 # start_o2ims_stx_mappers()
91 @retry(stop=stop_after_delay(10))
92 def wait_for_postgres_to_come_up(engine):
93 return engine.connect()
96 @retry(stop=stop_after_delay(10))
97 def wait_for_webapp_to_come_up():
98 return requests.get(config.get_api_url())
101 @retry(stop=stop_after_delay(10))
102 def wait_for_redis_to_come_up():
103 r = redis.Redis(**config.get_redis_host_and_port())
107 @pytest.fixture(scope="session")
109 engine = create_engine(config.get_postgres_uri(),
110 isolation_level="SERIALIZABLE")
111 wait_for_postgres_to_come_up(engine)
112 metadata.create_all(engine)
117 def postgres_session_factory(postgres_db):
118 yield sessionmaker(bind=postgres_db)
122 def postgres_session(postgres_session_factory):
123 return postgres_session_factory()
127 def postgres_uow(postgres_session_factory):
128 uow = unit_of_work.SqlAlchemyUnitOfWork(
129 session_factory=postgres_session_factory)
134 def postgres_flask_uow(postgres_uow):
135 app = Flask(__name__)
136 app.config["TESTING"] = True
138 bootstrap(False, postgres_uow)
139 configure_namespace(api)
140 yield postgres_uow, app
145 (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py")\
148 wait_for_webapp_to_come_up()
152 def restart_redis_pubsub():
153 wait_for_redis_to_come_up()
154 if not shutil.which("docker-compose"):
155 print("skipping restart, assumes running in container")
158 ["docker-compose", "restart", "-t", "0", "redis_pubsub"],