320 lines
8.4 KiB
Python
320 lines
8.4 KiB
Python
"""
|
|
Pytest configuration and fixtures for production-faithful Scouter E2E tests.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from pymongo import MongoClient
|
|
from redis import Redis
|
|
from sqlalchemy import create_engine, text
|
|
from testcontainers.mongodb import MongoDbContainer
|
|
from testcontainers.postgres import PostgresContainer
|
|
from testcontainers.redis import RedisContainer
|
|
from temporalio.testing import WorkflowEnvironment
|
|
from temporalio.worker import PollerBehaviorAutoscaling, Worker
|
|
|
|
from e2e.helpers import SCOUTER_TASK_QUEUE, postgres_connection_parts
|
|
from e2e.pi_web_api_test_server import PIWebAPITestServer
|
|
from scouter.activities.activities import Activities
|
|
from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter
|
|
from scouter.workflow.scouter import Scouter
|
|
from scouter.workflow.sub_workflows.core_scouter import CoreScouter
|
|
from sientia_do.notifications.handlers import CoreNotificationHandler
|
|
from sientia_do.observability.logger import Logger
|
|
|
|
E2E_DATABASE = 'scouter_e2e_test'
|
|
E2E_NOTIFICATION_COLLECTION = 'notification_queue'
|
|
DB_SCHEMA_PATH = Path(__file__).resolve().parent / 'db_schema.sql'
|
|
|
|
def _activity_list(activities: Activities) -> list:
|
|
"""Return bound activity callables for the E2E worker."""
|
|
return [
|
|
activities.load_latest_data,
|
|
activities.get_last_data_timestamp,
|
|
activities.put_last_data_timestamp,
|
|
activities.get_tag_values,
|
|
activities.data_quality_gate,
|
|
activities.aggregate_data,
|
|
activities.group_and_hold_data,
|
|
activities.export_data_to_postgres,
|
|
activities.write_metrics,
|
|
activities.store_data_package,
|
|
]
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def postgres_container():
|
|
"""
|
|
Session-scoped PostgreSQL testcontainer.
|
|
|
|
Return:
|
|
Running PostgresContainer instance
|
|
"""
|
|
container = PostgresContainer('postgres:15')
|
|
container.start()
|
|
yield container
|
|
container.stop()
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def mongo_container():
|
|
"""
|
|
Session-scoped MongoDB testcontainer.
|
|
|
|
Return:
|
|
Running MongoDbContainer instance
|
|
"""
|
|
container = MongoDbContainer('mongo:7')
|
|
container.start()
|
|
yield container
|
|
container.stop()
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def redis_container():
|
|
"""
|
|
Session-scoped Redis testcontainer.
|
|
|
|
Return:
|
|
Running RedisContainer instance
|
|
"""
|
|
container = RedisContainer('redis:7')
|
|
container.start()
|
|
yield container
|
|
container.stop()
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def postgres_engine(postgres_container):
|
|
"""
|
|
SQLAlchemy engine bound to the Postgres testcontainer for the session.
|
|
|
|
Return:
|
|
SQLAlchemy Engine
|
|
"""
|
|
engine = create_engine(postgres_container.get_connection_url())
|
|
yield engine
|
|
engine.dispose()
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def mongo_uri(mongo_container):
|
|
"""
|
|
MongoDB connection string for the testcontainer.
|
|
|
|
Return:
|
|
Connection URI string
|
|
"""
|
|
return mongo_container.get_connection_url()
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def redis_client(redis_container):
|
|
"""
|
|
Redis client connected to the testcontainer.
|
|
|
|
Return:
|
|
redis.Redis client with decode_responses=True
|
|
"""
|
|
host = redis_container.get_container_host_ip()
|
|
port = int(redis_container.get_exposed_port(6379))
|
|
client = Redis(host=host, port=port, decode_responses=True)
|
|
yield client
|
|
client.close()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_postgres_schema_and_table(postgres_engine):
|
|
"""
|
|
Apply db_schema.sql before each test so laborious_data is empty and current.
|
|
"""
|
|
sql = DB_SCHEMA_PATH.read_text(encoding='utf-8')
|
|
with postgres_engine.begin() as conn:
|
|
conn.exec_driver_sql(sql)
|
|
yield
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_mongo_collections(mongo_uri):
|
|
"""
|
|
Drop all collections in the E2E Mongo database between tests.
|
|
"""
|
|
client = MongoClient(mongo_uri)
|
|
try:
|
|
db = client[E2E_DATABASE]
|
|
for name in db.list_collection_names():
|
|
db.drop_collection(name)
|
|
finally:
|
|
client.close()
|
|
yield
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_redis(redis_client):
|
|
"""
|
|
Flush the Redis testcontainer between tests.
|
|
"""
|
|
redis_client.flushdb()
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_logger():
|
|
"""
|
|
Logger stand-in (only permitted MagicMock in the E2E harness).
|
|
|
|
Return:
|
|
MagicMock with Logger spec
|
|
"""
|
|
logger = MagicMock(spec=Logger)
|
|
logger.info = MagicMock()
|
|
logger.debug = MagicMock()
|
|
logger.error = MagicMock()
|
|
logger.warning = MagicMock()
|
|
logger.custom_info = MagicMock()
|
|
return logger
|
|
|
|
|
|
@pytest.fixture
|
|
def notification_handler(mock_logger, mongo_uri):
|
|
"""
|
|
Real CoreNotificationHandler backed by the Mongo testcontainer.
|
|
|
|
Return:
|
|
CoreNotificationHandler instance
|
|
"""
|
|
handler = CoreNotificationHandler(
|
|
connection_string=mongo_uri,
|
|
database=E2E_DATABASE,
|
|
logger=mock_logger,
|
|
project_name='scouter-e2e',
|
|
notification_topic=E2E_NOTIFICATION_COLLECTION,
|
|
)
|
|
yield handler
|
|
handler.shutdown()
|
|
|
|
|
|
@pytest.fixture
|
|
def notification_inserts(notification_handler):
|
|
"""
|
|
Spy wrapper around notification collection insert_one (still writes to Mongo).
|
|
|
|
Return:
|
|
MagicMock wrapping insert_one
|
|
"""
|
|
collection = notification_handler.mongo_collection
|
|
spy = MagicMock(wraps=collection.insert_one)
|
|
collection.insert_one = spy
|
|
return spy
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def pi_web_api_server():
|
|
"""
|
|
Session-scoped in-process PI Web API HTTP stub.
|
|
|
|
Return:
|
|
Started PIWebAPITestServer instance
|
|
"""
|
|
server = PIWebAPITestServer()
|
|
server.start()
|
|
yield server
|
|
server.stop()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def cleanup_pi_web_api_server(pi_web_api_server):
|
|
"""
|
|
Reset PI Web API stub state between tests.
|
|
"""
|
|
pi_web_api_server.clear()
|
|
yield
|
|
|
|
|
|
@pytest_asyncio.fixture(scope='session')
|
|
async def temporal_env():
|
|
"""
|
|
Session-scoped WorkflowEnvironment using the real local Temporal dev server.
|
|
|
|
Return:
|
|
WorkflowEnvironment from start_local()
|
|
"""
|
|
async with await WorkflowEnvironment.start_local() as env:
|
|
yield env
|
|
|
|
|
|
@pytest.fixture
|
|
def test_activities(
|
|
postgres_container,
|
|
mongo_uri,
|
|
redis_container,
|
|
pi_web_api_server,
|
|
mock_logger,
|
|
notification_handler,
|
|
):
|
|
"""
|
|
Production Activities wired to testcontainers and the PI Web API stub.
|
|
|
|
Return:
|
|
Live Activities instance (no unittest.mock.patch)
|
|
"""
|
|
pg_parts = postgres_connection_parts(postgres_container.get_connection_url())
|
|
redis_host = redis_container.get_container_host_ip()
|
|
redis_port = int(redis_container.get_exposed_port(6379))
|
|
|
|
activities = Activities(
|
|
postgres_config={
|
|
**pg_parts,
|
|
'min_connections': 1,
|
|
'max_connections': 5,
|
|
},
|
|
redis_config={
|
|
'host': redis_host,
|
|
'port': redis_port,
|
|
'username': '',
|
|
'password': '',
|
|
},
|
|
mongodb_config={
|
|
'connection_string': mongo_uri,
|
|
'database_name': E2E_DATABASE,
|
|
},
|
|
api_config={
|
|
'base_url': pi_web_api_server.base_url,
|
|
'auth_type': 'bearer',
|
|
'auth_token': 'e2e-test-token',
|
|
},
|
|
logger=mock_logger,
|
|
notification_handler=notification_handler,
|
|
)
|
|
yield activities
|
|
activities.shutdown()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def temporal_worker(temporal_env, test_activities):
|
|
"""
|
|
Temporal worker registering all Scouter workflows and activities on the E2E queue.
|
|
|
|
Return:
|
|
Running temporalio.worker.Worker
|
|
"""
|
|
async with Worker(
|
|
temporal_env.client,
|
|
task_queue=SCOUTER_TASK_QUEUE,
|
|
workflows=[Scouter, PIWebAPIScouter, CoreScouter],
|
|
activities=_activity_list(test_activities),
|
|
activity_executor=ThreadPoolExecutor(max_workers=50, thread_name_prefix='e2e-activity'),
|
|
max_concurrent_workflow_tasks=50,
|
|
max_concurrent_activities=50,
|
|
max_concurrent_local_activities=50,
|
|
workflow_task_poller_behavior=PollerBehaviorAutoscaling(minimum=1, initial=2, maximum=10),
|
|
activity_task_poller_behavior=PollerBehaviorAutoscaling(minimum=1, initial=2, maximum=10),
|
|
) as worker:
|
|
yield worker
|