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
  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
  36
  37 # import os
  38 # os.environ['O2APP_CONFIG'] = 'configs/o2app.conf'
  39 # os.environ['ALARM_YAML'] = 'configs/alarm.yaml'
  40
  41
  42 @pytest.fixture
  43 def mock_uow():
  44     session = MagicMock()
  45     uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory=session)
  46     return session, uow
  47
  48
  49 @pytest.fixture
  50 def mock_flask_uow(mock_uow):
  51     session, uow = mock_uow
  52     app = Flask(__name__)
  53     app.config["TESTING"] = True
  54     api = Api(app)
  55     bootstrap(False, uow)
  56     configure_namespace(api)
  57     return session, app
  58
  59
  60 @pytest.fixture
  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)
  65     return engine
  66
  67
  68 @pytest.fixture
  69 def sqlite_session_factory(in_memory_sqlite_db):
  70     yield sessionmaker(bind=in_memory_sqlite_db)
  71
  72
  73 @pytest.fixture
  74 def sqlite_uow(sqlite_session_factory):
  75     uow = unit_of_work.SqlAlchemyUnitOfWork(
  76         session_factory=sqlite_session_factory)
  77     # with uow:
  78     #     start_o2ims_mappers(uow.session.get_bind())
  79     #     uow.commit()
  80     yield uow
  81     # clear_mappers()
  82     with uow:
  83         engine = uow.session.get_bind()
  84         metadata.drop_all(engine)
  85
  86
  87 @pytest.fixture
  88 def sqlite_flask_uow(sqlite_uow):
  89     app = Flask(__name__)
  90     app.config["TESTING"] = True
  91     api = Api(app)
  92     bootstrap(False, sqlite_uow)
  93     configure_namespace(api)
  94     yield sqlite_uow, app
  95
  96
  97 @pytest.fixture
  98 def mappers():
  99     start_o2ims_mappers()
 100     # start_o2ims_stx_mappers()
 101     yield
 102     clear_mappers()
 103
 104
 105 @retry(stop=stop_after_delay(10))
 106 def wait_for_postgres_to_come_up(engine):
 107     return engine.connect()
 108
 109
 110 @retry(stop=stop_after_delay(10))
 111 def wait_for_webapp_to_come_up():
 112     return requests.get(config.get_api_url())
 113
 114
 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())
 118     return r.ping()
 119
 120
 121 @pytest.fixture(scope="session")
 122 def postgres_db():
 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)
 127     return engine
 128
 129
 130 @pytest.fixture
 131 def postgres_session_factory(postgres_db):
 132     yield sessionmaker(bind=postgres_db)
 133
 134
 135 @pytest.fixture
 136 def postgres_session(postgres_session_factory):
 137     return postgres_session_factory()
 138
 139
 140 @pytest.fixture
 141 def postgres_uow(postgres_session_factory):
 142     uow = unit_of_work.SqlAlchemyUnitOfWork(
 143         session_factory=postgres_session_factory)
 144     yield uow
 145
 146
 147 @pytest.fixture
 148 def postgres_flask_uow(postgres_uow):
 149     app = Flask(__name__)
 150     app.config["TESTING"] = True
 151     api = Api(app)
 152     bootstrap(False, postgres_uow)
 153     configure_namespace(api)
 154     yield postgres_uow, app
 155
 156
 157 @pytest.fixture
 158 def restart_api():
 159     (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py")\
 160         .touch()
 161     time.sleep(0.5)
 162     wait_for_webapp_to_come_up()
 163
 164
 165 @pytest.fixture
 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")
 170         return
 171     subprocess.run(
 172         ["docker-compose", "restart", "-t", "0", "redis_pubsub"],
 173         check=True,
 174     )