1 # Copyright (C) 2022 Wind River Systems, Inc.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
15 # pylint: disable=redefined-outer-name
19 from pathlib import Path
24 from flask import Flask
25 from flask_restx import Api
26 from sqlalchemy import create_engine
27 from sqlalchemy.orm import sessionmaker, clear_mappers
28 from tenacity import retry, stop_after_delay
29 from unittest.mock import MagicMock
31 from o2app.bootstrap import bootstrap
32 from o2ims.views import configure_namespace
33 from o2app.adapter import unit_of_work
34 from o2ims.adapter.orm import metadata, start_o2ims_mappers
35 from o2common.config import config
38 # os.environ['O2APP_CONFIG'] = 'configs/o2app.conf'
39 # os.environ['ALARM_YAML'] = 'configs/alarm.yaml'
45 uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory=session)
50 def mock_flask_uow(mock_uow):
51 session, uow = mock_uow
53 app.config["TESTING"] = True
56 configure_namespace(api)
61 def in_memory_sqlite_db():
62 engine = create_engine("sqlite:///:memory:")
63 # engine = create_engine("sqlite:///:memory:", echo=True)
64 metadata.create_all(engine)
69 def sqlite_session_factory(in_memory_sqlite_db):
70 yield sessionmaker(bind=in_memory_sqlite_db)
74 def sqlite_uow(sqlite_session_factory):
75 uow = unit_of_work.SqlAlchemyUnitOfWork(
76 session_factory=sqlite_session_factory)
78 # start_o2ims_mappers(uow.session.get_bind())
83 engine = uow.session.get_bind()
84 metadata.drop_all(engine)
88 def sqlite_flask_uow(sqlite_uow):
90 app.config["TESTING"] = True
92 bootstrap(False, sqlite_uow)
93 configure_namespace(api)
100 # start_o2ims_stx_mappers()
105 @retry(stop=stop_after_delay(10))
106 def wait_for_postgres_to_come_up(engine):
107 return engine.connect()
110 @retry(stop=stop_after_delay(10))
111 def wait_for_webapp_to_come_up():
112 return requests.get(config.get_api_url())
115 @retry(stop=stop_after_delay(10))
116 def wait_for_redis_to_come_up():
117 r = redis.Redis(**config.get_redis_host_and_port())
121 @pytest.fixture(scope="session")
123 engine = create_engine(config.get_postgres_uri(),
124 isolation_level="SERIALIZABLE")
125 wait_for_postgres_to_come_up(engine)
126 metadata.create_all(engine)
131 def postgres_session_factory(postgres_db):
132 yield sessionmaker(bind=postgres_db)
136 def postgres_session(postgres_session_factory):
137 return postgres_session_factory()
141 def postgres_uow(postgres_session_factory):
142 uow = unit_of_work.SqlAlchemyUnitOfWork(
143 session_factory=postgres_session_factory)
148 def postgres_flask_uow(postgres_uow):
149 app = Flask(__name__)
150 app.config["TESTING"] = True
152 bootstrap(False, postgres_uow)
153 configure_namespace(api)
154 yield postgres_uow, app
159 (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py")\
162 wait_for_webapp_to_come_up()
166 def restart_redis_pubsub():
167 wait_for_redis_to_come_up()
168 if not shutil.which("docker-compose"):
169 print("skipping restart, assumes running in container")
172 ["docker-compose", "restart", "-t", "0", "redis_pubsub"],