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
28 #os.environ['ALARM_YAML'] = 'configs/alarm.yaml'
34 uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory=session)
39 def mock_flask_uow(mock_uow):
40 session, uow = mock_uow
42 app.config["TESTING"] = True
45 configure_namespace(api)
50 def in_memory_sqlite_db():
51 engine = create_engine("sqlite:///:memory:")
52 # engine = create_engine("sqlite:///:memory:", echo=True)
53 metadata.create_all(engine)
58 def sqlite_session_factory(in_memory_sqlite_db):
59 yield sessionmaker(bind=in_memory_sqlite_db)
63 def sqlite_uow(sqlite_session_factory):
64 uow = unit_of_work.SqlAlchemyUnitOfWork(
65 session_factory=sqlite_session_factory)
67 # start_o2ims_mappers(uow.session.get_bind())
72 engine = uow.session.get_bind()
73 metadata.drop_all(engine)
77 def sqlite_flask_uow(sqlite_uow):
79 app.config["TESTING"] = True
81 bootstrap(False, sqlite_uow)
82 configure_namespace(api)
89 # start_o2ims_stx_mappers()
94 @retry(stop=stop_after_delay(10))
95 def wait_for_postgres_to_come_up(engine):
96 return engine.connect()
99 @retry(stop=stop_after_delay(10))
100 def wait_for_webapp_to_come_up():
101 return requests.get(config.get_api_url())
104 @retry(stop=stop_after_delay(10))
105 def wait_for_redis_to_come_up():
106 r = redis.Redis(**config.get_redis_host_and_port())
110 @pytest.fixture(scope="session")
112 engine = create_engine(config.get_postgres_uri(),
113 isolation_level="SERIALIZABLE")
114 wait_for_postgres_to_come_up(engine)
115 metadata.create_all(engine)
120 def postgres_session_factory(postgres_db):
121 yield sessionmaker(bind=postgres_db)
125 def postgres_session(postgres_session_factory):
126 return postgres_session_factory()
130 def postgres_uow(postgres_session_factory):
131 uow = unit_of_work.SqlAlchemyUnitOfWork(
132 session_factory=postgres_session_factory)
137 def postgres_flask_uow(postgres_uow):
138 app = Flask(__name__)
139 app.config["TESTING"] = True
141 bootstrap(False, postgres_uow)
142 configure_namespace(api)
143 yield postgres_uow, app
148 (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py")\
151 wait_for_webapp_to_come_up()
155 def restart_redis_pubsub():
156 wait_for_redis_to_come_up()
157 if not shutil.which("docker-compose"):
158 print("skipping restart, assumes running in container")
161 ["docker-compose", "restart", "-t", "0", "redis_pubsub"],