tests/conftest.py

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