Add the PATCH method for the monitoring API
[pti/o2.git] / 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     )