import pytest\r
import redis\r
import requests\r
+from flask import Flask\r
+from flask_restx import Api\r
from sqlalchemy import create_engine\r
from sqlalchemy.orm import sessionmaker, clear_mappers\r
from tenacity import retry, stop_after_delay\r
+from unittest.mock import MagicMock\r
+\r
+from o2common.config import config\r
\r
from o2ims.adapter.orm import metadata, start_o2ims_mappers\r
-from o2ims import config\r
+# from o2ims.adapter.clients.orm_stx import start_o2ims_stx_mappers\r
+\r
+from o2app.adapter import unit_of_work\r
+from o2ims.views.ocloud_route import configure_namespace\r
+\r
+from o2app.bootstrap import bootstrap\r
+\r
+\r
+@pytest.fixture\r
+def mock_uow():\r
+ session = MagicMock()\r
+ uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory=session)\r
+ return session, uow\r
+\r
+\r
+@pytest.fixture\r
+def mock_flask_uow(mock_uow):\r
+ session, uow = mock_uow\r
+ app = Flask(__name__)\r
+ app.config["TESTING"] = True\r
+ api = Api(app)\r
+ bus = bootstrap(False, uow)\r
+ configure_namespace(api, bus)\r
+ return session, app\r
\r
\r
@pytest.fixture\r
yield sessionmaker(bind=in_memory_sqlite_db)\r
\r
\r
+@pytest.fixture\r
+def sqlite_uow(sqlite_session_factory):\r
+ uow = unit_of_work.SqlAlchemyUnitOfWork(\r
+ session_factory=sqlite_session_factory)\r
+ # with uow:\r
+ # start_o2ims_mappers(uow.session.get_bind())\r
+ # uow.commit()\r
+ yield uow\r
+ # clear_mappers()\r
+ with uow:\r
+ engine = uow.session.get_bind()\r
+ metadata.drop_all(engine)\r
+\r
+\r
+@pytest.fixture\r
+def sqlite_flask_uow(sqlite_uow):\r
+ app = Flask(__name__)\r
+ app.config["TESTING"] = True\r
+ api = Api(app)\r
+ bus = bootstrap(False, sqlite_uow)\r
+ configure_namespace(api, bus)\r
+ yield sqlite_uow, app\r
+\r
+\r
@pytest.fixture\r
def mappers():\r
start_o2ims_mappers()\r
+ # start_o2ims_stx_mappers()\r
yield\r
clear_mappers()\r
\r
\r
@pytest.fixture(scope="session")\r
def postgres_db():\r
- engine = create_engine(config.get_postgres_uri(), isolation_level="SERIALIZABLE")\r
+ engine = create_engine(config.get_postgres_uri(),\r
+ isolation_level="SERIALIZABLE")\r
wait_for_postgres_to_come_up(engine)\r
metadata.create_all(engine)\r
return engine\r
return postgres_session_factory()\r
\r
\r
+@pytest.fixture\r
+def postgres_uow(postgres_session_factory):\r
+ uow = unit_of_work.SqlAlchemyUnitOfWork(\r
+ session_factory=postgres_session_factory)\r
+ yield uow\r
+\r
+\r
+@pytest.fixture\r
+def postgres_flask_uow(postgres_uow):\r
+ app = Flask(__name__)\r
+ app.config["TESTING"] = True\r
+ api = Api(app)\r
+ bus = bootstrap(False, postgres_uow)\r
+ configure_namespace(api, bus)\r
+ yield postgres_uow, app\r
+\r
+\r
@pytest.fixture\r
def restart_api():\r
- (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py").touch()\r
+ (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py")\\r
+ .touch()\r
time.sleep(0.5)\r
wait_for_webapp_to_come_up()\r
\r