-# pylint: disable=redefined-outer-name\r
-import shutil\r
-import subprocess\r
-import time\r
-from pathlib import Path\r
-\r
-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.adapter.clients.orm_stx import start_o2ims_stx_mappers\r
-\r
-from o2app.adapter import unit_of_work\r
-from o2ims.views 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
- bootstrap(False, uow)\r
- configure_namespace(api)\r
- return session, app\r
-\r
-\r
-@pytest.fixture\r
-def in_memory_sqlite_db():\r
- engine = create_engine("sqlite:///:memory:")\r
- # engine = create_engine("sqlite:///:memory:", echo=True)\r
- metadata.create_all(engine)\r
- return engine\r
-\r
-\r
-@pytest.fixture\r
-def sqlite_session_factory(in_memory_sqlite_db):\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
- bootstrap(False, sqlite_uow)\r
- configure_namespace(api)\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
-@retry(stop=stop_after_delay(10))\r
-def wait_for_postgres_to_come_up(engine):\r
- return engine.connect()\r
-\r
-\r
-@retry(stop=stop_after_delay(10))\r
-def wait_for_webapp_to_come_up():\r
- return requests.get(config.get_api_url())\r
-\r
-\r
-@retry(stop=stop_after_delay(10))\r
-def wait_for_redis_to_come_up():\r
- r = redis.Redis(**config.get_redis_host_and_port())\r
- return r.ping()\r
-\r
-\r
-@pytest.fixture(scope="session")\r
-def postgres_db():\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
-\r
-\r
-@pytest.fixture\r
-def postgres_session_factory(postgres_db):\r
- yield sessionmaker(bind=postgres_db)\r
-\r
-\r
-@pytest.fixture\r
-def postgres_session(postgres_session_factory):\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
- bootstrap(False, postgres_uow)\r
- configure_namespace(api)\r
- yield postgres_uow, app\r
-\r
-\r
-@pytest.fixture\r
-def restart_api():\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
-\r
-@pytest.fixture\r
-def restart_redis_pubsub():\r
- wait_for_redis_to_come_up()\r
- if not shutil.which("docker-compose"):\r
- print("skipping restart, assumes running in container")\r
- return\r
- subprocess.run(\r
- ["docker-compose", "restart", "-t", "0", "redis_pubsub"],\r
- check=True,\r
- )\r
+# pylint: disable=redefined-outer-name
+import shutil
+import subprocess
+import time
+from pathlib import Path
+
+import pytest
+import redis
+import requests
+from flask import Flask
+from flask_restx import Api
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, clear_mappers
+from tenacity import retry, stop_after_delay
+from unittest.mock import MagicMock
+
+from o2common.config import config
+
+from o2ims.adapter.orm import metadata, start_o2ims_mappers
+# from o2ims.adapter.clients.orm_stx import start_o2ims_stx_mappers
+
+from o2app.adapter import unit_of_work
+from o2ims.views import configure_namespace
+
+from o2app.bootstrap import bootstrap
+
+
+@pytest.fixture
+def mock_uow():
+ session = MagicMock()
+ uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory=session)
+ return session, uow
+
+
+@pytest.fixture
+def mock_flask_uow(mock_uow):
+ session, uow = mock_uow
+ app = Flask(__name__)
+ app.config["TESTING"] = True
+ api = Api(app)
+ bootstrap(False, uow)
+ configure_namespace(api)
+ return session, app
+
+
+@pytest.fixture
+def in_memory_sqlite_db():
+ engine = create_engine("sqlite:///:memory:")
+ # engine = create_engine("sqlite:///:memory:", echo=True)
+ metadata.create_all(engine)
+ return engine
+
+
+@pytest.fixture
+def sqlite_session_factory(in_memory_sqlite_db):
+ yield sessionmaker(bind=in_memory_sqlite_db)
+
+
+@pytest.fixture
+def sqlite_uow(sqlite_session_factory):
+ uow = unit_of_work.SqlAlchemyUnitOfWork(
+ session_factory=sqlite_session_factory)
+ # with uow:
+ # start_o2ims_mappers(uow.session.get_bind())
+ # uow.commit()
+ yield uow
+ # clear_mappers()
+ with uow:
+ engine = uow.session.get_bind()
+ metadata.drop_all(engine)
+
+
+@pytest.fixture
+def sqlite_flask_uow(sqlite_uow):
+ app = Flask(__name__)
+ app.config["TESTING"] = True
+ api = Api(app)
+ bootstrap(False, sqlite_uow)
+ configure_namespace(api)
+ yield sqlite_uow, app
+
+
+@pytest.fixture
+def mappers():
+ start_o2ims_mappers()
+ # start_o2ims_stx_mappers()
+ yield
+ clear_mappers()
+
+
+@retry(stop=stop_after_delay(10))
+def wait_for_postgres_to_come_up(engine):
+ return engine.connect()
+
+
+@retry(stop=stop_after_delay(10))
+def wait_for_webapp_to_come_up():
+ return requests.get(config.get_api_url())
+
+
+@retry(stop=stop_after_delay(10))
+def wait_for_redis_to_come_up():
+ r = redis.Redis(**config.get_redis_host_and_port())
+ return r.ping()
+
+
+@pytest.fixture(scope="session")
+def postgres_db():
+ engine = create_engine(config.get_postgres_uri(),
+ isolation_level="SERIALIZABLE")
+ wait_for_postgres_to_come_up(engine)
+ metadata.create_all(engine)
+ return engine
+
+
+@pytest.fixture
+def postgres_session_factory(postgres_db):
+ yield sessionmaker(bind=postgres_db)
+
+
+@pytest.fixture
+def postgres_session(postgres_session_factory):
+ return postgres_session_factory()
+
+
+@pytest.fixture
+def postgres_uow(postgres_session_factory):
+ uow = unit_of_work.SqlAlchemyUnitOfWork(
+ session_factory=postgres_session_factory)
+ yield uow
+
+
+@pytest.fixture
+def postgres_flask_uow(postgres_uow):
+ app = Flask(__name__)
+ app.config["TESTING"] = True
+ api = Api(app)
+ bootstrap(False, postgres_uow)
+ configure_namespace(api)
+ yield postgres_uow, app
+
+
+@pytest.fixture
+def restart_api():
+ (Path(__file__).parent / "../src/o2ims/entrypoints/flask_application.py")\
+ .touch()
+ time.sleep(0.5)
+ wait_for_webapp_to_come_up()
+
+
+@pytest.fixture
+def restart_redis_pubsub():
+ wait_for_redis_to_come_up()
+ if not shutil.which("docker-compose"):
+ print("skipping restart, assumes running in container")
+ return
+ subprocess.run(
+ ["docker-compose", "restart", "-t", "0", "redis_pubsub"],
+ check=True,
+ )
Code Review / pti / o2.git / blobdiff