tests/conftest.py

   1 # Copyright (C) 2022-2024 Wind River Systems, Inc.
   2 #
   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
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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.
  14
  15 # pylint: disable=redefined-outer-name
  16 import shutil
  17 import subprocess
  18 import sys
  19 import time
  20 from pathlib import Path
  21
  22 import pytest
  23 import redis
  24 import requests
  25 from flask import Flask
  26 from flask_restx import Api
  27 from sqlalchemy import create_engine
  28 from sqlalchemy.orm import sessionmaker, clear_mappers
  29 from tenacity import retry, stop_after_delay
  30 from unittest.mock import MagicMock
  31 from mock_alchemy.mocking import UnifiedAlchemyMagicMock
  32
  33 # Mock cgtsclient, dcmanagerclient, fmclient
  34 modules_to_mock = [
  35     'cgtsclient',
  36     'cgtsclient.client',
  37     'cgtsclient.exc',
  38     'dcmanagerclient',
  39     'dcmanagerclient.api',
  40     'dcmanagerclient.api.client',
  41     'fmclient',
  42     'fmclient.client',
  43     'fmclient.common',
  44     'fmclient.common.exceptions'
  45 ]
  46
  47 for module_name in modules_to_mock:
  48     sys.modules[module_name] = MagicMock()
  49
  50 from o2app.bootstrap import bootstrap
  51 from o2ims.views import configure_namespace
  52 from o2app.adapter import unit_of_work
  53 from o2ims.adapter.orm import metadata, start_o2ims_mappers
  54 from o2common.config import config
  55
  56 # import os
  57 # os.environ['O2APP_CONFIG'] = 'configs/o2app.conf'
  58 # os.environ['ALARM_YAML'] = 'configs/alarm.yaml'
  59
  60
  61 @pytest.fixture
  62 def mock_uow():
  63     session = MagicMock()
  64     uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory=session)
  65     return session, uow
  66
  67
  68 @pytest.fixture
  69 def mock_alchemy_uow():
  70     session = UnifiedAlchemyMagicMock()
  71     uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory=session)
  72     return session, uow
  73
  74
  75 @pytest.fixture
  76 def mock_flask_uow(mock_uow):
  77     session, uow = mock_uow
  78     app = Flask(__name__)
  79     app.config["TESTING"] = True
  80     api = Api(app)
  81     bootstrap(False, uow)
  82     configure_namespace(api)
  83     return session, app
  84
  85
  86 @pytest.fixture
  87 def in_memory_sqlite_db():
  88     engine = create_engine("sqlite:///:memory:")
  89     # engine = create_engine("sqlite:///:memory:", echo=True)
  90     metadata.create_all(engine)
  91     return engine
  92
  93
  94 @pytest.fixture
  95 def sqlite_session_factory(in_memory_sqlite_db):
  96     yield sessionmaker(bind=in_memory_sqlite_db)
  97
  98
  99 @pytest.fixture
 100 def sqlite_uow(sqlite_session_factory):
 101     uow = unit_of_work.SqlAlchemyUnitOfWork(
 102         session_factory=sqlite_session_factory)
 103     # with uow:
 104     #     start_o2ims_mappers(uow.session.get_bind())
 105     #     uow.commit()
 106     yield uow
 107     # clear_mappers()
 108     with uow:
 109         engine = uow.session.get_bind()
 110         metadata.drop_all(engine)
 111
 112
 113 @pytest.fixture
 114 def sqlite_flask_uow(sqlite_uow):
 115     app = Flask(__name__)
 116     app.config["TESTING"] = True
 117     api = Api(app)
 118     bootstrap(False, sqlite_uow)
 119     configure_namespace(api)
 120     yield sqlite_uow, app
 121
 122
 123 @pytest.fixture
 124 def mappers():
 125     start_o2ims_mappers()
 126     # start_o2ims_stx_mappers()
 127     yield
 128     clear_mappers()
 129
 130
 131 @retry(stop=stop_after_delay(10))
 132 def wait_for_postgres_to_come_up(engine):
 133     return engine.connect()
 134
 135
 136 @retry(stop=stop_after_delay(10))
 137 def wait_for_webapp_to_come_up():
 138     return requests.get(config.get_api_url())
 139
 140
 141 @retry(stop=stop_after_delay(10))
 142 def wait_for_redis_to_come_up():
 143     r = redis.Redis(**config.get_redis_host_and_port())
 144     return r.ping()
 145
 146
 147 @pytest.fixture(scope="session")
 148 def postgres_db():
 149     engine = create_engine(config.get_postgres_uri(),
 150                            isolation_level="SERIALIZABLE")
 151     wait_for_postgres_to_come_up(engine)
 152     metadata.create_all(engine)
 153     return engine
 154
 155
 156 @pytest.fixture
 157 def postgres_session_factory(postgres_db):
 158     yield sessionmaker(bind=postgres_db)
 159
 160
 161 @pytest.fixture
 162 def postgres_session(postgres_session_factory):
 163     return postgres_session_factory()
 164
 165
 166 @pytest.fixture
 167 def postgres_uow(postgres_session_factory):
 168     uow = unit_of_work.SqlAlchemyUnitOfWork(
 169         session_factory=postgres_session_factory)
 170     yield uow
 171
 172
 173 @pytest.fixture
 174 def postgres_flask_uow(postgres_uow):
 175     app = Flask(__name__)
 176     app.config["TESTING"] = True
 177     api = Api(app)
 178     bootstrap(False, postgres_uow)
 179     configure_namespace(api)
 180     yield postgres_uow, app
 181
 182
 183 @pytest.fixture
 184 def restart_api():
 185     (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py")\
 186         .touch()
 187     time.sleep(0.5)
 188     wait_for_webapp_to_come_up()
 189
 190
 191 @pytest.fixture
 192 def restart_redis_pubsub():
 193     wait_for_redis_to_come_up()
 194     if not shutil.which("docker-compose"):
 195         print("skipping restart, assumes running in container")
 196         return
 197     subprocess.run(
 198         ["docker-compose", "restart", "-t", "0", "redis_pubsub"],
 199         check=True,
 200     )