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 o2ims import config
\r
18 from o2ims.adapter.orm import metadata, start_o2ims_mappers
\r
19 from o2ims.adapter.clients.orm_stx import start_o2ims_stx_mappers
\r
20 from o2ims.adapter import unit_of_work
\r
21 from o2ims.views.ocloud_route import configure_namespace
\r
22 from o2ims.bootstrap import bootstrap
\r
27 session = MagicMock()
\r
28 uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory=session)
\r
33 def mock_flask_uow(mock_uow):
\r
34 session, uow = mock_uow
\r
35 app = Flask(__name__)
\r
36 # app.config["TESTING"] = True
\r
38 bus = bootstrap(False, uow)
\r
39 configure_namespace(api, bus)
\r
40 client = app.test_client()
\r
41 return session, client
\r
45 def in_memory_sqlite_db():
\r
46 engine = create_engine("sqlite:///:memory:")
\r
47 # engine = create_engine("sqlite:///:memory:", echo=True)
\r
48 metadata.create_all(engine)
\r
53 def sqlite_session_factory(in_memory_sqlite_db):
\r
54 yield sessionmaker(bind=in_memory_sqlite_db)
\r
58 def sqlite_uow(sqlite_session_factory):
\r
59 uow = unit_of_work.SqlAlchemyUnitOfWork(
\r
60 session_factory=sqlite_session_factory)
\r
62 # start_o2ims_mappers(uow.session.get_bind())
\r
67 engine = uow.session.get_bind()
\r
68 metadata.drop_all(engine)
\r
72 def sqlite_flask_uow(sqlite_uow):
\r
73 app = Flask(__name__)
\r
74 # app.config["TESTING"] = True
\r
76 bus = bootstrap(False, sqlite_uow)
\r
77 configure_namespace(api, bus)
\r
78 yield app.test_client()
\r
83 start_o2ims_mappers()
\r
84 start_o2ims_stx_mappers()
\r
89 @retry(stop=stop_after_delay(10))
\r
90 def wait_for_postgres_to_come_up(engine):
\r
91 return engine.connect()
\r
94 @retry(stop=stop_after_delay(10))
\r
95 def wait_for_webapp_to_come_up():
\r
96 return requests.get(config.get_api_url())
\r
99 @retry(stop=stop_after_delay(10))
\r
100 def wait_for_redis_to_come_up():
\r
101 r = redis.Redis(**config.get_redis_host_and_port())
\r
105 @pytest.fixture(scope="session")
\r
107 engine = create_engine(config.get_postgres_uri(),
\r
108 isolation_level="SERIALIZABLE")
\r
109 wait_for_postgres_to_come_up(engine)
\r
110 metadata.create_all(engine)
\r
115 def postgres_session_factory(postgres_db):
\r
116 yield sessionmaker(bind=postgres_db)
\r
120 def postgres_session(postgres_session_factory):
\r
121 return postgres_session_factory()
\r
125 def postgres_uow(postgres_session_factory):
\r
126 uow = unit_of_work.SqlAlchemyUnitOfWork(
\r
127 session_factory=postgres_session_factory)
\r
132 def postgres_flask_uow(postgres_uow):
\r
133 app = Flask(__name__)
\r
135 bus = bootstrap(False, postgres_uow)
\r
136 configure_namespace(api, bus)
\r
137 yield app.test_client()
\r
142 (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py")\
\r
145 wait_for_webapp_to_come_up()
\r
149 def restart_redis_pubsub():
\r
150 wait_for_redis_to_come_up()
\r
151 if not shutil.which("docker-compose"):
\r
152 print("skipping restart, assumes running in container")
\r
155 ["docker-compose", "restart", "-t", "0", "redis_pubsub"],
\r