Merge "Fix INF-356 DC client auth token is invalid"
[pti/o2.git] / 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     )