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