1 # pylint: disable=redefined-outer-name
\r
5 from pathlib import Path
\r
10 from flask import Flask
\r
11 from sqlalchemy import create_engine
\r
12 from sqlalchemy.orm import sessionmaker, clear_mappers
\r
13 from tenacity import retry, stop_after_delay
\r
14 from unittest.mock import MagicMock
\r
16 from o2ims import config
\r
17 from o2ims.adapter.orm import metadata, start_o2ims_mappers
\r
18 from o2ims.adapter.clients.orm_stx import start_o2ims_stx_mappers
\r
19 from o2ims.adapter import unit_of_work
\r
20 from o2ims.views.ocloud_route import configure_routes
\r
21 from o2ims.bootstrap import bootstrap
\r
26 session = MagicMock()
\r
27 uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory=session)
\r
32 def mock_flask_uow(mock_uow):
\r
33 session, uow = mock_uow
\r
34 app = Flask(__name__)
\r
35 # app.config["TESTING"] = True
\r
36 bus = bootstrap(False, uow)
\r
37 configure_routes(app, bus)
\r
38 client = app.test_client()
\r
39 return session, client
\r
43 def in_memory_sqlite_db():
\r
44 engine = create_engine("sqlite:///:memory:")
\r
45 # engine = create_engine("sqlite:///:memory:", echo=True)
\r
46 metadata.create_all(engine)
\r
51 def sqlite_session_factory(in_memory_sqlite_db):
\r
52 yield sessionmaker(bind=in_memory_sqlite_db)
\r
56 def sqlite_uow(sqlite_session_factory):
\r
57 uow = unit_of_work.SqlAlchemyUnitOfWork(
\r
58 session_factory=sqlite_session_factory)
\r
60 # start_o2ims_mappers(uow.session.get_bind())
\r
65 engine = uow.session.get_bind()
\r
66 metadata.drop_all(engine)
\r
70 def sqlite_flask_uow(sqlite_uow):
\r
71 app = Flask(__name__)
\r
72 # app.config["TESTING"] = True
\r
73 bus = bootstrap(False, sqlite_uow)
\r
74 configure_routes(app, bus)
\r
75 yield app.test_client()
\r
80 start_o2ims_mappers()
\r
81 start_o2ims_stx_mappers()
\r
86 @retry(stop=stop_after_delay(10))
\r
87 def wait_for_postgres_to_come_up(engine):
\r
88 return engine.connect()
\r
91 @retry(stop=stop_after_delay(10))
\r
92 def wait_for_webapp_to_come_up():
\r
93 return requests.get(config.get_api_url())
\r
96 @retry(stop=stop_after_delay(10))
\r
97 def wait_for_redis_to_come_up():
\r
98 r = redis.Redis(**config.get_redis_host_and_port())
\r
102 @pytest.fixture(scope="session")
\r
104 engine = create_engine(config.get_postgres_uri(),
\r
105 isolation_level="SERIALIZABLE")
\r
106 wait_for_postgres_to_come_up(engine)
\r
107 metadata.create_all(engine)
\r
112 def postgres_session_factory(postgres_db):
\r
113 yield sessionmaker(bind=postgres_db)
\r
117 def postgres_session(postgres_session_factory):
\r
118 return postgres_session_factory()
\r
122 def postgres_uow(postgres_session_factory):
\r
123 uow = unit_of_work.SqlAlchemyUnitOfWork(
\r
124 session_factory=postgres_session_factory)
\r
129 def postgres_flask_uow(postgres_uow):
\r
130 app = Flask(__name__)
\r
131 bus = bootstrap(False, postgres_uow)
\r
132 configure_routes(app, bus)
\r
133 yield app.test_client()
\r
138 (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py")\
\r
141 wait_for_webapp_to_come_up()
\r
145 def restart_redis_pubsub():
\r
146 wait_for_redis_to_come_up()
\r
147 if not shutil.which("docker-compose"):
\r
148 print("skipping restart, assumes running in container")
\r
151 ["docker-compose", "restart", "-t", "0", "redis_pubsub"],
\r