SIENTIAPDE-1646

Update project configuration and dependencies

- Added .mypy_cache and .cursor to .gitignore.
- Changed asyncio_default_fixture_loop_scope and asyncio_default_test_loop_scope to "session" in pyproject.toml.
- Updated e2e testing dependencies in requirements-dev.txt, replacing fakeredis and mongomock with pytest-httpserver.
- Updated requirements.txt to use sientia_do instead of a specific git commit.
- Modified sonar-project.properties to remove a file from coverage exclusions.
- Enhanced E2E test fixtures in e2e/conftest.py for better container management.
- Cleaned up e2e test files related to CoreScouter and PIWebAPIScouter workflows.
This commit is contained in:
vitor-aignosi
2026-05-25 12:58:03 -03:00
parent 909ad25b63
commit 34dbc886f3
65 changed files with 2591 additions and 2849 deletions

37
e2e/README.md Normal file
View File

@@ -0,0 +1,37 @@
# Scouter end-to-end tests
Production-faithful E2E tests for `Scouter`, `PIWebAPIScouter`, and `CoreScouter` against real backing services.
## Requirements
- Docker (for testcontainers: MongoDB, Redis, PostgreSQL)
- Python 3.11+ with dev dependencies: `pip install -r requirements-dev.txt`
## Run locally
```bash
pytest e2e/ --override-ini testpaths=e2e -m e2e -v
```
Stop on first failure:
```bash
pytest e2e/ --override-ini testpaths=e2e -m e2e -x
```
## Coverage (separate from unit tests)
```bash
COVERAGE_FILE=.coverage.e2e pytest e2e/ --override-ini testpaths=e2e -m e2e --cov=scouter --cov-branch
coverage combine .coverage .coverage.e2e && coverage report
```
## Scenario catalog
See [scenarios.md](scenarios.md) for numbered scenarios and how they map to `test_scenario_*` functions. Section `## 0` of that file lists the harness smoke tests in `test_harness_smoke.py` (infra liveness checks, not business scenarios).
## Production code is not mocked
E2E uses real testcontainers, an in-process PI Web API HTTP server, `WorkflowEnvironment.start_local()`, and production `Activities` wiring. The only stand-ins are `mock_logger` and the optional `notification_inserts` spy. If a scenario fails due to a production defect, it is marked `xfail(strict=True)` and tracked in `openspec/changes/standardize-and-complete-e2e-tests/notes.md` when applicable.
Unit tests under `tests/` remain Docker-free and run with the default `pytest` invocation.

View File

@@ -1,121 +1,177 @@
"""
Pytest configuration and fixtures for E2E tests.
Pytest configuration and fixtures for production-faithful Scouter E2E tests.
"""
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from unittest.mock import MagicMock
import pandas as pd
import pytest
import pytest_asyncio
from pandas import DataFrame
from pymongo import MongoClient
from redis import Redis
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from testcontainers.mongodb import MongoDbContainer
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
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
from sientia_do.observability.metrics_controller import MetricsController
from e2e.fixtures.fake_mongodb_repository import FakeMongoDBRepository
from e2e.fixtures.fake_redis_repository import FakeRedisRepository
E2E_DATABASE = 'scouter_e2e_test'
E2E_NOTIFICATION_COLLECTION = 'notification_queue'
DB_SCHEMA_PATH = Path(__file__).resolve().parent / 'db_schema.sql'
# Test constants
TEST_MONGODB_CONNECTION_STRING = 'mongodb://localhost:27017'
TEST_DATABASE_NAME = 'test_db'
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_asyncio.fixture(scope='session')
@pytest.fixture(scope='session')
def postgres_container():
"""
Create a PostgreSQL container using testcontainers.
This fixture creates a real PostgreSQL database in a Docker container
that will be used for all tests in the session.
Session-scoped PostgreSQL testcontainer.
Return:
Running PostgresContainer instance
"""
postgres = PostgresContainer('postgres:15')
postgres.start()
yield postgres
postgres.stop()
container = PostgresContainer('postgres:15')
container.start()
yield container
container.stop()
@pytest_asyncio.fixture
@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):
"""
Create SQLAlchemy engine for PostgreSQL test database.
This fixture creates a connection to the PostgreSQL container
created by the postgres_container fixture.
SQLAlchemy engine bound to the Postgres testcontainer for the session.
Return:
SQLAlchemy Engine
"""
# Get connection URL from container
connection_string = postgres_container.get_connection_url()
engine = create_engine(connection_string)
engine = create_engine(postgres_container.get_connection_url())
yield engine
engine.dispose()
def _create_schema_and_table(engine):
@pytest.fixture(scope='session')
def mongo_uri(mongo_container):
"""
Helper function to create schema and table in the given engine.
This is used by both the autouse fixture and test_activities to ensure
the schema exists before Activities tries to use it.
Note: For tests, we create a non-partitioned table to avoid issues
with pandas to_sql recognizing partitioned tables.
MongoDB connection string for the testcontainer.
Return:
Connection URI string
"""
schema_name = 'sientia_data'
table_name = 'laborious_data'
# Use begin() to ensure transaction is properly committed
with engine.begin() as conn:
# Create schema
conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {schema_name}"))
# Create table WITHOUT partitioning (simpler for tests)
# Same structure as production, but without PARTITION BY RANGE
create_table_sql = f"""
CREATE TABLE IF NOT EXISTS {schema_name}.{table_name} (
id SERIAL NOT NULL,
model_id int4 NOT NULL,
variable text NOT NULL,
value numeric NULL,
"timestamp" timestamptz NOT NULL,
created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL,
PRIMARY KEY (id, created_at)
);
"""
conn.execute(text(create_table_sql))
# Transaction is automatically committed when exiting the 'with' block
return mongo_container.get_connection_url()
@pytest_asyncio.fixture(autouse=True)
@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):
"""
Automatically create necessary schema and table before each test.
This fixture runs automatically (autouse=True) and ensures
that the sientia_data schema and laborious_data table exist
with the correct structure before tests execute.
Note: For tests, we use a non-partitioned table which is simpler and avoids issues
with pandas to_sql recognizing partitioned tables.
Apply db_schema.sql before each test so laborious_data is empty and current.
"""
_create_schema_and_table(postgres_engine)
sql = DB_SCHEMA_PATH.read_text(encoding='utf-8')
with postgres_engine.begin() as conn:
conn.exec_driver_sql(sql)
yield
@pytest_asyncio.fixture
@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():
"""Mock logger for testing."""
"""
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()
@@ -125,238 +181,139 @@ def mock_logger():
return logger
@pytest_asyncio.fixture
def mock_mongo_client():
@pytest.fixture
def notification_handler(mock_logger, mongo_uri):
"""
Mock MongoDB client to avoid real connections.
This fixture mocks the pymongo.MongoClient used by CoreNotificationHandler,
allowing us to use a real NotificationHandler instance without connecting to MongoDB.
Real CoreNotificationHandler backed by the Mongo testcontainer.
Return:
CoreNotificationHandler instance
"""
mock_client = MagicMock()
mock_db = MagicMock()
mock_collection = MagicMock()
# Configure the mock chain: client[database] -> db[collection] -> collection
mock_client.__getitem__.return_value = mock_db
mock_db.__getitem__.return_value = mock_collection
# Mock server_info() to avoid connection attempts
mock_client.server_info = MagicMock()
# Mock insert_one for notifications
mock_collection.insert_one = MagicMock()
return mock_client
@pytest_asyncio.fixture
def notification_handler(mock_logger, mock_mongo_client):
"""
Create a real NotificationHandler instance with mocked MongoDB client.
This fixture creates a real CoreNotificationHandler instance but mocks
the underlying MongoDB connection to avoid real database connections.
"""
# Patch MongoClient where it's imported in the handlers module
with patch('sientia_do.notifications.handlers.MongoClient', return_value=mock_mongo_client):
handler = CoreNotificationHandler(
connection_string=TEST_MONGODB_CONNECTION_STRING,
database=TEST_DATABASE_NAME,
logger=mock_logger,
project_name='scouter',
)
yield handler
handler.shutdown()
@pytest_asyncio.fixture
def metrics_controller(mock_logger):
"""
Create a real MetricsController instance.
MetricsController doesn't require external services, so we can use
a real instance without mocking anything.
"""
controller = MetricsController(logger=mock_logger)
yield controller
# MetricsController might have cleanup, but it's optional
@pytest_asyncio.fixture
def mock_pi_web_api_client():
"""Mock PI Web API client."""
mock_client = MagicMock()
# Mock DataFrame response similar to real API
# The real PIWebAPIClient returns timestamp as datetime, so we need to match that
mock_df = DataFrame({
'timestamp': [
'2024-01-01 12:00:00+0000',
'2024-01-01 12:01:00+0000',
'2024-01-01 12:02:00+0000',
],
'name': ['tag1', 'tag2', 'tag3'],
'value': [10.5, 20.3, 30.7],
'tag': ['webid1', 'webid2', 'webid3'],
})
# Convert timestamp to datetime (UTC, floored to seconds) to match real client behavior
mock_df['timestamp'] = pd.to_datetime(mock_df['timestamp'], utc=True).dt.floor('s')
mock_client.get_latest_values_df = AsyncMock(return_value=mock_df)
mock_client.close = MagicMock()
return mock_client
@pytest_asyncio.fixture(scope='function')
def test_activities(
postgres_engine,
postgres_container,
mock_logger,
notification_handler,
metrics_controller,
mock_pi_web_api_client,
):
"""
Create Activities instance with test dependencies.
This fixture creates a real Activities instance with:
- PostgreSQL database (via testcontainers)
- FakeRedis instead of real Redis
- FakeMongoDB instead of real MongoDB
- Mocked PI Web API client
- Real NotificationHandler and MetricsController (with mocked underlying services)
"""
# Get connection details from container
connection_string = postgres_container.get_connection_url()
# Parse connection string to get individual components
# Format: postgresql://testuser:testpass@localhost:5432/test
from urllib.parse import urlparse
parsed = urlparse(connection_string)
# Ensure schema and table exist BEFORE creating Activities
# This ensures the schema exists when Activities initializes its engine
_create_schema_and_table(postgres_engine)
# Create fake repositories that will be used from the start
fake_redis_repo = FakeRedisRepository(
host='localhost',
port=6379,
username='',
password='',
handler = CoreNotificationHandler(
connection_string=mongo_uri,
database=E2E_DATABASE,
logger=mock_logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
project_name='scouter-e2e',
notification_topic=E2E_NOTIFICATION_COLLECTION,
)
fake_mongo_repo = FakeMongoDBRepository(
connection_string=TEST_MONGODB_CONNECTION_STRING,
database_name=TEST_DATABASE_NAME,
logger=mock_logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
# Patch create_engine to return our postgres_engine instead of creating a new one
# This ensures Activities uses the same engine from the start
original_create_engine = create_engine
def patched_create_engine(connection_string, *args, **kwargs):
# Check if this is the connection string that Activities would create
# Activities creates: postgresql://user:password@host:port/dbname
expected_conn_str = f"postgresql://{parsed.username or 'test'}:{parsed.password or 'test'}@localhost:{postgres_container.get_exposed_port(5432)}/{parsed.path.lstrip('/') if parsed.path else 'test'}"
# If it matches our test container connection, return our engine
if connection_string == expected_conn_str:
return postgres_engine
# Otherwise, use the original create_engine
return original_create_engine(connection_string, *args, **kwargs)
# Patch RedisRepository to return our fake repository from the start
def patched_redis_repository(*args, **kwargs):
return fake_redis_repo
# Patch MongoDBRepository to return our fake repository from the start
def patched_mongodb_repository(*args, **kwargs):
return fake_mongo_repo
# Patch PIWebAPIClient to return our mock from the start
def patched_pi_web_api_client(*args, **kwargs):
return mock_pi_web_api_client
# Create Activities with test configurations
# All patches ensure it uses our test instances from the start
with patch('sientia_do.temporal.activities.postgres.create_engine', new=patched_create_engine), \
patch('scouter.activities.redis.RedisRepository', new=patched_redis_repository), \
patch('scouter.activities.mongodb.MongoDBRepository', new=patched_mongodb_repository), \
patch('scouter.activities.api.PIWebAPIClient', new=patched_pi_web_api_client):
activities = Activities(
postgres_config={
'host': 'localhost', # Container exposes to localhost
'port': postgres_container.get_exposed_port(5432),
'user': parsed.username or 'test',
'password': parsed.password or 'test',
'dbname': parsed.path.lstrip('/') if parsed.path else 'test',
'min_connections': 1,
'max_connections': 5,
},
redis_config={
'host': 'localhost',
'port': 6379,
'username': '',
'password': '',
},
mongodb_config={
'connection_string': TEST_MONGODB_CONNECTION_STRING,
'database_name': TEST_DATABASE_NAME,
},
api_config={
'base_url': 'http://localhost:8080',
'auth_type': 'bearer',
'auth_token': 'test_token',
},
logger=mock_logger,
notification_handler=notification_handler,
)
# Verify that Activities is using our instances from the start
assert activities.engine is postgres_engine, "Activities should use the same engine as postgres_engine"
assert activities.redis_repository is fake_redis_repo, "Activities should use the same fake Redis repository"
assert activities.mongodb_repository is fake_mongo_repo, "Activities should use the same fake MongoDB repository"
assert activities.pi_web_api_client is mock_pi_web_api_client, "Activities should use the same mock PI Web API client"
yield activities
# Cleanup
activities.shutdown()
yield handler
handler.shutdown()
@pytest_asyncio.fixture(scope='function')
async def temporal_test_env():
"""Create Temporal test environment."""
env = await WorkflowEnvironment.start_time_skipping()
async with env:
@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_asyncio.fixture(scope='function')
async def temporal_worker(temporal_test_env, test_activities):
"""Create Temporal worker with test activities."""
@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_test_env.client,
task_queue='test-queue',
workflows=[PIWebAPIScouter, CoreScouter],
activities=[
test_activities.get_tag_values,
test_activities.data_quality_gate,
test_activities.aggregate_data,
test_activities.group_and_hold_data,
test_activities.export_data_to_postgres,
test_activities.write_metrics,
test_activities.store_data_package,
],
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

16
e2e/db_schema.sql Normal file
View File

@@ -0,0 +1,16 @@
-- Single source of truth for E2E Postgres DDL (non-partitioned mirror of production sientia_data.laborious_data).
DROP TABLE IF EXISTS sientia_data.laborious_data;
DROP SCHEMA IF EXISTS sientia_data CASCADE;
CREATE SCHEMA sientia_data;
CREATE TABLE sientia_data.laborious_data (
id SERIAL NOT NULL,
model_id int4 NOT NULL,
variable text NOT NULL,
value numeric NULL,
"timestamp" timestamptz NOT NULL,
created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL,
PRIMARY KEY (id, created_at)
);

View File

@@ -1,241 +0,0 @@
"""
Fake MongoDB Repository adapter for testing.
This adapter implements the MongoDBRepository interface using mongomock
to provide an in-memory MongoDB server for testing.
"""
import time
from typing import Any
from mongomock import MongoClient
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.observability.logger import Logger
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
def clear_mongo_id(docs: list) -> list:
"""
Remove MongoDB internal `_id` fields from nested structures.
Same implementation as in the real MongoDBRepository.
Args:
docs: The list of documents or nested structures to clean.
Returns:
The cleaned documents with `_id` fields removed wherever present.
"""
for doc in docs:
if isinstance(doc, list):
clear_mongo_id(doc)
elif isinstance(doc, dict):
if '_id' in doc:
del doc['_id']
for _key, value in doc.items():
if isinstance(value, list):
clear_mongo_id(value)
elif isinstance(value, dict):
clear_mongo_id([value])
return docs
class FakeMongoDBRepository(SientiaMonitoring):
"""
Fake MongoDB Repository that uses mongomock for testing.
Implements the same interface as MongoDBRepository but uses
mongomock for in-memory MongoDB operations.
"""
def __init__(
self,
connection_string: str,
database_name: str,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
"""
Initialize fake MongoDB repository with mongomock.
Args:
connection_string: MongoDB connection string (ignored in fake mode)
database_name: Target database name
logger: Logger instance
notification_handler: Notification handler
metrics_controller: Metrics controller
"""
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
self.database_name = database_name
# Use mongomock instead of real MongoDB
self.mongo_client = MongoClient()
self.database = self.mongo_client[self.database_name]
logger.info('Fake MongoDB connection initialized')
def close(self):
"""
Closes fake MongoDB connection and shuts down monitoring.
"""
try:
if self.mongo_client:
self.logger.info('Closing fake MongoDB connection...')
self.mongo_client.close()
self.logger.info('Fake MongoDB connection closed successfully')
except Exception as e:
self.logger.error(f'Failed to close fake MongoDB connection: {e}')
SientiaMonitoring.shutdown(self)
def __del__(self):
"""
Destructor that closes the connection when destroying the instance.
"""
self.close()
async def find(
self,
collection_name: str,
filters: dict[str, Any],
metadata: dict[str, Any],
) -> list[dict[str, Any]]:
"""
Finds documents in a MongoDB collection based on provided filters.
Args:
collection_name: Name of the collection to search
filters: Query filters to apply
metadata: Dictionary with additional metadata
Return:
List of documents matching the filters (with `_id` removed)
"""
try:
start_time = time.time()
collection = self.database[collection_name]
documents = list(collection.find(filters, {'_id': 0}))
except Exception as e:
self.logger.error(f'Failed to find documents in fake MongoDB: {e}')
raise e
return clear_mongo_id(documents)
async def aggregate(
self,
collection_name: str,
pipeline: list[dict[str, Any]],
metadata: dict[str, Any],
) -> list[dict[str, Any]]:
"""
Executes an aggregation on a MongoDB collection.
Args:
collection_name: Name of the collection to aggregate
pipeline: MongoDB aggregation pipeline
metadata: Dictionary with additional metadata
Return:
List of documents resulting from aggregation (with `_id` removed)
"""
try:
start_time = time.time()
collection = self.database[collection_name]
documents = list(collection.aggregate(pipeline))
except Exception as e:
self.logger.error(f'Failed to aggregate documents in fake MongoDB: {e}')
raise e
return clear_mongo_id(documents)
async def update_many(
self,
collection_name: str,
filters: dict[str, Any],
update: dict[str, Any],
metadata: dict[str, Any],
) -> None:
"""
Updates multiple documents in a MongoDB collection.
Args:
collection_name: Collection name
filters: Filters to identify documents to update
update: Update operations to apply
metadata: Dictionary with additional metadata
"""
try:
collection = self.database[collection_name]
collection.update_many(filters, update)
except Exception as e:
self.logger.error(f'Failed to update documents in fake MongoDB: {e}')
raise e
async def insert_many(
self,
collection_name: str,
documents: list[dict[str, Any]],
metadata: dict[str, Any],
) -> None:
"""
Inserts multiple documents into a MongoDB collection.
Args:
collection_name: Collection name
documents: List of documents to insert
metadata: Dictionary with additional metadata
"""
try:
collection = self.database[collection_name]
collection.insert_many(documents)
except Exception as e:
self.logger.error(f'Failed to insert documents in fake MongoDB: {e}')
raise e
async def insert(
self,
collection_name: str,
document: dict[str, Any],
metadata: dict[str, Any],
) -> None:
"""
Inserts a document into a MongoDB collection.
Args:
collection_name: Collection name
document: Document to insert
metadata: Dictionary with additional metadata
"""
try:
collection = self.database[collection_name]
collection.insert_one(document)
except Exception as e:
self.logger.error(f'Failed to insert document in fake MongoDB: {e}')
raise e
async def delete_many(
self,
collection_name: str,
filters: dict[str, Any],
metadata: dict[str, Any],
) -> None:
"""
Removes multiple documents from a MongoDB collection.
Args:
collection_name: Collection name
filters: Filters to identify documents to remove
metadata: Dictionary with additional metadata
"""
try:
collection = self.database[collection_name]
collection.delete_many(filters)
except Exception as e:
self.logger.error(f'Failed to delete documents in fake MongoDB: {e}')
raise e

View File

@@ -1,170 +0,0 @@
"""
Fake Redis Repository adapter for testing.
This adapter implements the RedisRepository interface using fakeredis
to provide an in-memory Redis server for testing.
"""
import json
from typing import Any
import fakeredis
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.observability.logger import Logger
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
class FakeRedisRepository(SientiaMonitoring):
"""
Fake Redis Repository that uses fakeredis for testing.
Implements the same interface as RedisRepository but uses
fakeredis for in-memory Redis operations.
"""
def __init__(
self,
host: str,
port: int,
username: str,
password: str,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
"""
Initialize fake Redis repository with fakeredis.
Args:
host: Redis server address (ignored in fake mode)
port: Redis server port (ignored in fake mode)
username: Username (ignored in fake mode)
password: Password (ignored in fake mode)
logger: Logger instance
notification_handler: Notification handler
metrics_controller: Metrics controller
"""
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
# Create fake Redis server
self.redis_client = fakeredis.FakeStrictRedis(
decode_responses=True
)
def _get_redis(self):
"""Get fakeredis connection."""
return self.redis_client
async def get(self, key: str, metadata: dict | None = None):
"""
Gets a value from fake Redis by key.
Args:
key: Key of the value to retrieve
metadata: Optional dictionary with additional metadata
Return:
Deserialized value from Redis or None if not found
"""
redis = self._get_redis()
try:
history = redis.get(key)
return json.loads(history) if history else None
except Exception as e:
self.error(f'Error getting data from fake redis: {e}', metadata or {})
raise e
async def set(
self,
key: str,
data: dict,
ttl: int = 600,
nx: bool = False,
metadata: dict | None = None,
) -> bool:
"""
Sets a value in fake Redis with optional TTL.
Args:
key: Key of the value to set
data: Dictionary with data to store
ttl: Time to live in seconds (default: 600)
nx: If True, only sets if key doesn't exist (default: False)
metadata: Optional dictionary with additional metadata
Return:
True if value was set, False otherwise
"""
redis = self._get_redis()
try:
result = redis.set(
key, json.dumps(data), ex=ttl, nx=nx
)
return bool(result)
except Exception as e:
self.error(f'Error setting data in fake redis: {e}', metadata or {})
raise e
async def delete(self, key: str, metadata: dict | None = None):
"""
Removes a key from fake Redis.
Args:
key: Key to be removed
metadata: Optional dictionary with additional metadata
"""
redis = self._get_redis()
try:
redis.delete(key)
except Exception as e:
self.error(f'Error deleting data from fake redis: {e}', metadata or {})
raise e
async def expire(self, key: str, ttl: int, metadata: dict | None = None):
"""
Sets the time to live (TTL) of an existing Redis key.
Args:
key: Key whose TTL will be set
ttl: Time to live in seconds
metadata: Optional dictionary with additional metadata
"""
redis = self._get_redis()
try:
redis.expire(key, ttl)
except Exception as e:
self.error(f'Error expiring data from fake redis: {e}', metadata or {})
raise e
async def keys(self, pattern: str, metadata: dict | None = None):
"""
Gets all keys matching the specified pattern.
Args:
pattern: Search pattern for keys
metadata: Optional dictionary with additional metadata
Return:
List of keys matching the pattern
"""
redis = self._get_redis()
try:
keys = redis.keys(pattern)
return keys
except Exception as e:
self.error(f'Error getting keys from fake redis: {e}', metadata or {})
raise e
def close(self):
"""
Closes fake Redis connection and shuts down monitoring.
"""
if self.redis_client:
self.redis_client.close()
SientiaMonitoring.shutdown(self)

404
e2e/helpers.py Normal file
View File

@@ -0,0 +1,404 @@
"""
Shared helpers for Scouter end-to-end tests.
"""
from __future__ import annotations
import json
import re
import uuid
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
from pymongo import MongoClient
from redis import Redis
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ
from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine
from temporalio.client import Client
SCENARIO_INPUTS_DIR = Path(__file__).resolve().parent / 'scenario_inputs'
SCOUTER_TASK_QUEUE = 'scouter-test-queue'
_NOW_MARKER = re.compile(r'^@now(?:([+-])(\d+)([smhd]))?$')
def _resolve_timestamp_marker(value: str) -> datetime:
"""
Resolve @now relative timestamp markers to timezone-aware datetimes.
Args:
- value: Marker string such as @now, @now-1h, or @now+30m
Return:
Resolved datetime in UTC
"""
match = _NOW_MARKER.match(value.strip())
if not match:
raise ValueError(f'Invalid timestamp marker: {value}')
now = datetime.now(UTC)
if match.group(1) is None:
return now
sign, amount, unit = match.group(1), int(match.group(2)), match.group(3)
delta_kwargs = {'seconds': 0, 'minutes': 0, 'hours': 0, 'days': 0}
if unit == 's':
delta_kwargs['seconds'] = amount
elif unit == 'm':
delta_kwargs['minutes'] = amount
elif unit == 'h':
delta_kwargs['hours'] = amount
elif unit == 'd':
delta_kwargs['days'] = amount
delta = timedelta(**delta_kwargs)
return now - delta if sign == '-' else now + delta
def _resolve_payload(node: Any) -> Any:
"""
Recursively resolve @now markers inside JSON-loaded structures.
Args:
- node: JSON node (dict, list, or scalar)
Return:
Structure with markers replaced by datetimes or formatted strings
"""
if isinstance(node, dict):
return {key: _resolve_payload(value) for key, value in node.items()}
if isinstance(node, list):
return [_resolve_payload(item) for item in node]
if isinstance(node, str) and node.startswith('@now'):
resolved = _resolve_timestamp_marker(node)
return resolved.strftime('%Y-%m-%d %H:%M:%S.%f%z')
return node
def load_scenario_input(name: str, **overrides: Any) -> dict[str, Any]:
"""
Load a scenario JSON file and apply optional overrides.
Args:
- name: Scenario slug with or without .json suffix
- overrides: Top-level keys merged into the loaded document
Return:
Parsed scenario document with @now markers resolved
"""
slug = name.removesuffix('.json')
path = SCENARIO_INPUTS_DIR / f'{slug}.json'
with path.open(encoding='utf-8') as handle:
payload = json.load(handle)
resolved = _resolve_payload(payload)
if overrides:
resolved.update(overrides)
return resolved
def make_workflow_id(prefix: str) -> str:
"""
Build a unique Temporal workflow id for E2E runs.
Args:
- prefix: Human-readable prefix for the workflow id
Return:
Unique workflow id string
"""
return f'{prefix}-{uuid.uuid4().hex[:12]}'
async def start_and_await_workflow(
client: Client,
workflow_run: Any,
input_data: dict[str, Any],
workflow_id: str,
*,
task_queue: str = SCOUTER_TASK_QUEUE,
timeout: float = 300.0,
) -> Any:
"""
Start a workflow on the E2E task queue and await its result.
Args:
- client: Temporal client from WorkflowEnvironment
- workflow_run: Workflow run method (e.g. Scouter.run)
- input_data: Workflow input payload
- workflow_id: Unique workflow id
- task_queue: Task queue name
- timeout: Maximum seconds to wait for completion
Return:
Workflow result (None for Scouter-family workflows)
"""
return await client.execute_workflow(
workflow_run,
input_data,
id=workflow_id,
task_queue=task_queue,
execution_timeout=timedelta(seconds=timeout),
)
def _coerce_mongo_document(document: dict[str, Any]) -> dict[str, Any]:
"""
Convert string timestamps in seed documents to BSON datetimes for Mongo filters.
Args:
- document: Raw document dict from scenario JSON
Return:
Document with inserted_at as datetime when present
"""
doc = dict(document)
inserted_at = doc.get('inserted_at')
if isinstance(inserted_at, str):
doc['inserted_at'] = datetime.strptime(inserted_at, DATETIME_FORMAT_MS_WITH_TZ).replace(
tzinfo=UTC
)
return doc
def seed_raw_collection(
mongo_uri: str,
database: str,
schedule_name: str,
documents: list[dict[str, Any]],
) -> None:
"""
Insert raw Mongo documents into raw_<schedule_name>.
Args:
- mongo_uri: MongoDB connection string
- database: Database name
- schedule_name: Schedule slug used in collection name
- documents: Documents to insert
"""
collection_name = f'raw_{schedule_name}'
client = MongoClient(mongo_uri)
try:
collection = client[database][collection_name]
if documents:
collection.insert_many([_coerce_mongo_document(doc) for doc in documents])
finally:
client.close()
def seed_last_data_timestamp(
redis_client: Redis,
workflow_name: str,
schedule_name: str,
value: str,
) -> None:
"""
Pre-seed last_data_timestamp Redis key the same way RedisRepository.set stores strings.
Args:
- redis_client: Connected Redis client
- workflow_name: Workflow name segment in the key
- schedule_name: Schedule name segment in the key
- value: Timestamp string to store
"""
key = f'last_data_timestamp:{workflow_name}:{schedule_name}'
redis_client.set(key, json.dumps(value))
def count_laborious_rows(engine: Engine, model_id: str | int | None = None) -> int:
"""
Count rows in sientia_data.laborious_data, optionally filtered by model_id.
Args:
- engine: SQLAlchemy engine bound to the Postgres testcontainer
- model_id: Optional model id filter
Return:
Row count
"""
query = 'SELECT COUNT(*) FROM sientia_data.laborious_data'
params: dict[str, Any] = {}
if model_id is not None:
query += ' WHERE model_id = :model_id'
params['model_id'] = int(model_id)
with engine.connect() as conn:
return conn.execute(text(query), params).scalar() or 0
def fetch_laborious_rows(engine: Engine, model_id: str | int | None = None) -> list[dict[str, Any]]:
"""
Fetch laborious_data rows as plain dicts.
Args:
- engine: SQLAlchemy engine bound to the Postgres testcontainer
- model_id: Optional model id filter
Return:
List of row dicts with variable and value keys
"""
query = 'SELECT model_id, variable, value, timestamp FROM sientia_data.laborious_data'
params: dict[str, Any] = {}
if model_id is not None:
query += ' WHERE model_id = :model_id'
params['model_id'] = int(model_id)
query += ' ORDER BY variable'
with engine.connect() as conn:
rows = conn.execute(text(query), params).mappings().all()
return [dict(row) for row in rows]
def count_held_data_keys(redis_client: Redis) -> int:
"""
Count Redis keys matching held_data_*.
Args:
- redis_client: Connected Redis client
Return:
Number of matching keys
"""
return len(redis_client.keys('held_data_*'))
def count_notifications(
mongo_uri: str,
database: str,
*,
notification_id: str | None = None,
level: str | None = None,
) -> int:
"""
Count notification documents in the E2E notification collection.
Args:
- mongo_uri: MongoDB connection string
- database: Database name
- notification_id: Optional notification_id filter
- level: Optional level filter (WARNING, ERROR, ...)
Return:
Matching document count
"""
client = MongoClient(mongo_uri)
try:
collection = client[database]['notification_queue']
query: dict[str, Any] = {}
if notification_id:
query['notification_id'] = notification_id
if level:
query['level'] = level
return collection.count_documents(query)
finally:
client.close()
def default_model_tags(
*,
names: list[str],
aggr: str = 'avg',
data_range: list[float] | None = None,
frequency: int = 60000,
) -> dict[str, dict[str, Any]]:
"""
Build a minimal model_tags map for E2E scenarios.
Args:
- names: Tag names to include
- aggr: Aggregation function (aggr_func field)
- data_range: Optional [min, max] validation range
- frequency: Collection frequency in milliseconds
Return:
model_tags dict keyed by tag name
"""
if data_range is None:
data_range = [0, 100]
return {
name: {
'webid': f'webid_{name}',
'aggr_func': aggr,
'data_range': data_range,
'frequency': frequency,
}
for name in names
}
def default_scouter_input(
*,
model_id: str = '1',
model_name: str = 'Test Model',
schedule_name: str = 'test-schedule',
workflow_name: str = 'scouter',
**overrides: Any,
) -> dict[str, Any]:
"""
Return a base workflow input dict for Scouter / CoreScouter E2E runs.
Args:
- model_id: Model identifier
- model_name: Human-readable model name
- schedule_name: Schedule slug
- workflow_name: Parent workflow name
- overrides: Additional keys merged into the payload
Return:
Workflow input dictionary
"""
payload: dict[str, Any] = {
'topic': 'e2e-topic',
'model_id': model_id,
'model_name': model_name,
'schedule_name': schedule_name,
'workflow_name': workflow_name,
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'model_tags': default_model_tags(names=['tag1']),
}
payload.update(overrides)
return payload
def apply_pi_web_api_server_config(server: Any, config: dict[str, Any] | None) -> None:
"""
Configure the in-process PI Web API stub from a scenario pi_web_api_server block.
Args:
- server: PIWebAPITestServer instance
- config: Optional mode/rows/timeout_sleep_seconds dict from scenario JSON
"""
if not config:
return
server.set_mode(
config.get('mode', 'success'),
rows=config.get('rows'),
timeout_sleep_seconds=config.get('timeout_sleep_seconds', 60),
)
def postgres_connection_parts(connection_url: str) -> dict[str, Any]:
"""
Parse a SQLAlchemy Postgres URL into Activities postgres_config fields.
Args:
- connection_url: SQLAlchemy connection URL from testcontainers
Return:
Dict with host, port, user, password, dbname keys
"""
parsed = urlparse(connection_url)
return {
'host': parsed.hostname or 'localhost',
'port': parsed.port or 5432,
'user': parsed.username or 'test',
'password': parsed.password or 'test',
'dbname': (parsed.path or '/test').lstrip('/'),
}

View File

@@ -0,0 +1,124 @@
"""
In-process HTTP server emulating PI Web API streamsets/recorded responses for E2E tests.
"""
from __future__ import annotations
import json
import time
from typing import Any, Literal
from pytest_httpserver import HTTPServer
from werkzeug import Request
from werkzeug.wrappers import Response
PIWebAPIMode = Literal['success', 'empty', 'error', 'timeout']
STREAMSETS_RECORDED_PATH = '/streamsets/recorded'
class PIWebAPITestServer:
"""
Thread-backed PI Web API stub using pytest-httpserver (real HTTP for pycurl clients).
"""
def __init__(self) -> None:
self._httpserver = HTTPServer(host='127.0.0.1', port=0)
self._mode: PIWebAPIMode = 'success'
self._rows: list[dict[str, Any]] = []
self._timeout_sleep_seconds = 60
self.requests: list[Request] = []
@property
def host(self) -> str:
return self._httpserver.host
@property
def port(self) -> int:
return self._httpserver.port
@property
def base_url(self) -> str:
return f'http://{self.host}:{self.port}'
def start(self) -> None:
"""Start the HTTP server and register the streamsets handler."""
self._httpserver.start()
self._register_handler()
def stop(self) -> None:
"""Stop the HTTP server."""
self._httpserver.stop()
def clear(self) -> None:
"""Clear recorded requests and reset mode to success with no rows."""
self.requests.clear()
self._mode = 'success'
self._rows = []
self._httpserver.clear()
self._register_handler()
def set_mode(
self,
mode: PIWebAPIMode,
rows: list[dict[str, Any]] | None = None,
*,
timeout_sleep_seconds: int = 60,
) -> None:
"""
Configure the next responses from the stub server.
Args:
- mode: Response mode (success, empty, error, timeout)
- rows: Optional list of row dicts with keys name, webid, timestamp, value
- timeout_sleep_seconds: Sleep duration for timeout mode (must exceed client timeout)
"""
self._mode = mode
if rows is not None:
self._rows = rows
self._timeout_sleep_seconds = timeout_sleep_seconds
self._register_handler()
def _register_handler(self) -> None:
self._httpserver.expect_request(
STREAMSETS_RECORDED_PATH,
method='GET',
).respond_with_handler(self._handle_streamsets_recorded)
def _handle_streamsets_recorded(self, request: Request):
self.requests.append(request)
if self._mode == 'timeout':
time.sleep(self._timeout_sleep_seconds)
return self._json_response({'Items': []}, status=200)
if self._mode == 'error':
return self._json_response({'error': 'internal'}, status=500)
if self._mode == 'empty' or not self._rows:
return self._json_response({'Items': []}, status=200)
items_by_name: dict[str, list[dict[str, Any]]] = {}
for row in self._rows:
name = row['name']
items_by_name.setdefault(name, []).append(
{
'Timestamp': row['timestamp'],
'Value': row['value'],
'Good': True,
'Questionable': False,
}
)
items = [
{'Name': name, 'Items': points}
for name, points in items_by_name.items()
]
return self._json_response({'Items': items}, status=200)
@staticmethod
def _json_response(payload: dict[str, Any], *, status: int) -> Response:
return Response(
json.dumps(payload),
status=status,
mimetype='application/json',
)

View File

@@ -0,0 +1,24 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "24", "model_name": "Core Avg", "schedule_name": "core-avg", "workflow_name": "pi_web_api_scouter"}},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-avg",
"model_name": "Core Avg",
"model_id": "24",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag_avg", "value": 10.0, "tag": "w1"},
{"timestamp": "2024-01-01 12:01:00+0000", "name": "tag_avg", "value": 20.0, "tag": "w1"},
{"timestamp": "2024-01-01 12:02:00+0000", "name": "tag_avg", "value": 30.0, "tag": "w1"}
],
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag_avg": {"webid": "w1", "aggr_func": "avg", "data_range": [0, 1000], "frequency": 60000}
}
},
"expected_value": 20.0
}

View File

@@ -0,0 +1,24 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "28", "model_name": "Core Lts", "schedule_name": "core-lts", "workflow_name": "pi_web_api_scouter"}},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-lts",
"model_name": "Core Lts",
"model_id": "28",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag_lts", "value": 100.0, "tag": "w1"},
{"timestamp": "2024-01-01 12:01:00+0000", "name": "tag_lts", "value": 200.0, "tag": "w1"},
{"timestamp": "2024-01-01 12:02:00+0000", "name": "tag_lts", "value": 300.0, "tag": "w1"}
],
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag_lts": {"webid": "w1", "aggr_func": "lts", "data_range": [0, 1000], "frequency": 60000}
}
},
"expected_value": 300.0
}

View File

@@ -0,0 +1,24 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "26", "model_name": "Core Max", "schedule_name": "core-max", "workflow_name": "pi_web_api_scouter"}},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-max",
"model_name": "Core Max",
"model_id": "26",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag_max", "value": 5.0, "tag": "w1"},
{"timestamp": "2024-01-01 12:01:00+0000", "name": "tag_max", "value": 15.0, "tag": "w1"},
{"timestamp": "2024-01-01 12:02:00+0000", "name": "tag_max", "value": 10.0, "tag": "w1"}
],
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag_max": {"webid": "w1", "aggr_func": "max", "data_range": [0, 1000], "frequency": 60000}
}
},
"expected_value": 15.0
}

View File

@@ -0,0 +1,24 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "25", "model_name": "Core Mdn", "schedule_name": "core-mdn", "workflow_name": "pi_web_api_scouter"}},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-mdn",
"model_name": "Core Mdn",
"model_id": "25",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag_mdn", "value": 1.0, "tag": "w1"},
{"timestamp": "2024-01-01 12:01:00+0000", "name": "tag_mdn", "value": 9.0, "tag": "w1"},
{"timestamp": "2024-01-01 12:02:00+0000", "name": "tag_mdn", "value": 5.0, "tag": "w1"}
],
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag_mdn": {"webid": "w1", "aggr_func": "mdn", "data_range": [0, 1000], "frequency": 60000}
}
},
"expected_value": 5.0
}

View File

@@ -0,0 +1,24 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "27", "model_name": "Core Min", "schedule_name": "core-min", "workflow_name": "pi_web_api_scouter"}},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-min",
"model_name": "Core Min",
"model_id": "27",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag_min", "value": 50.0, "tag": "w1"},
{"timestamp": "2024-01-01 12:01:00+0000", "name": "tag_min", "value": 30.0, "tag": "w1"},
{"timestamp": "2024-01-01 12:02:00+0000", "name": "tag_min", "value": 40.0, "tag": "w1"}
],
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag_min": {"webid": "w1", "aggr_func": "min", "data_range": [0, 1000], "frequency": 60000}
}
},
"expected_value": 30.0
}

View File

@@ -0,0 +1,22 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "30", "model_name": "Core Debug", "schedule_name": "core-debug", "workflow_name": "subworkflow.core_scouter"}},
"workflow_name": "subworkflow.core_scouter",
"schedule_name": "core-debug",
"model_name": "Core Debug",
"model_id": "30",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag1", "value": 11.0, "tag": "webid1"}
],
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"debug_data_package": true,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
}
}
}

View File

@@ -0,0 +1,24 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "31", "model_name": "Core Empty Group", "schedule_name": "core-empty-group", "workflow_name": "pi_web_api_scouter"}},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-empty-group",
"model_name": "Core Empty Group",
"model_id": "31",
"data": {
"timestamp": [],
"name": [],
"value": [],
"tag": []
},
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
}
}
}

View File

@@ -0,0 +1,24 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "29", "model_name": "Core Fill Tags", "schedule_name": "core-fill", "workflow_name": "pi_web_api_scouter"}},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-fill",
"model_name": "Core Fill Tags",
"model_id": "29",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag1", "value": 10.0, "tag": "webid1"}
],
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": true,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000},
"tag2": {"webid": "webid2", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000},
"tag3": {"webid": "webid3", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
}
},
"expected_missing_tags": ["tag2", "tag3"]
}

View File

@@ -0,0 +1,28 @@
{
"workflow_input": {
"metadata": {
"metadata": {
"model_id": "20",
"model_name": "Core Happy",
"schedule_name": "core-happy",
"workflow_name": "pi_web_api_scouter"
}
},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-happy",
"model_name": "Core Happy",
"model_id": "20",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag1", "value": 10.5, "tag": "webid1"}
],
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
}
}
}

View File

@@ -0,0 +1,23 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "32", "model_name": "Core Bad Aggr", "schedule_name": "core-bad-aggr", "workflow_name": "pi_web_api_scouter"}},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-bad-aggr",
"model_name": "Core Bad Aggr",
"model_id": "32",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag_ok", "value": 10.0, "tag": "w1"},
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag_bad", "value": 20.0, "tag": "w2"}
],
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag_ok": {"webid": "w1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000},
"tag_bad": {"webid": "w2", "aggr_func": "bogus", "data_range": [0, 100], "frequency": 60000}
}
}
}

View File

@@ -0,0 +1,23 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "21", "model_name": "Core Null Discard", "schedule_name": "core-null-discard", "workflow_name": "pi_web_api_scouter"}},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-null-discard",
"model_name": "Core Null Discard",
"model_id": "21",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag1", "value": null, "tag": "webid1"},
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag2", "value": 50.0, "tag": "webid2"}
],
"trigger_laborious": false,
"filters": {"NULL_VALUES_FILTER": {"policy": "DISCARD"}},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000},
"tag2": {"webid": "webid2", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
}
}
}

View File

@@ -0,0 +1,23 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "22", "model_name": "Core Null Warn", "schedule_name": "core-null-warn", "workflow_name": "pi_web_api_scouter"}},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-null-warn",
"model_name": "Core Null Warn",
"model_id": "22",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag1", "value": null, "tag": "webid1"},
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag2", "value": 50.0, "tag": "webid2"}
],
"trigger_laborious": false,
"filters": {"NULL_VALUES_FILTER": {"policy": "WARN"}},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000},
"tag2": {"webid": "webid2", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
}
}
}

View File

@@ -0,0 +1,23 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "23", "model_name": "Core OOB Discard", "schedule_name": "core-oob-discard", "workflow_name": "pi_web_api_scouter"}},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-oob-discard",
"model_name": "Core OOB Discard",
"model_id": "23",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag1", "value": 150.0, "tag": "webid1"},
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag2", "value": 50.0, "tag": "webid2"}
],
"trigger_laborious": false,
"filters": {"OUT_OF_BOUNDS_FILTER": {"policy": "DISCARD"}},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000},
"tag2": {"webid": "webid2", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
}
}
}

View File

@@ -0,0 +1,21 @@
{
"workflow_input": {
"metadata": {"metadata": {"model_id": "33", "model_name": "Core PG Fail", "schedule_name": "core-pg-fail", "workflow_name": "pi_web_api_scouter"}},
"workflow_name": "pi_web_api_scouter",
"schedule_name": "core-pg-fail",
"model_name": "Core PG Fail",
"model_id": "33",
"data": [
{"timestamp": "2024-01-01 12:00:00+0000", "name": "tag1", "value": 10.0, "tag": "webid1"}
],
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
}
}
}

View File

@@ -0,0 +1,23 @@
{
"workflow_input": {
"model_id": "14",
"model_name": "PI Error",
"schedule_name": "pi-error",
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
},
"pi_web_api_query": {
"endpoint": "/streamsets/recorded",
"period": "*-1d",
"max_count": 1,
"api_timeout": 30
}
},
"pi_web_api_server": {"mode": "error"}
}

View File

@@ -0,0 +1,29 @@
{
"workflow_input": {
"model_id": "12",
"model_name": "PI Debug Package",
"schedule_name": "pi-debug",
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"debug_data_package": true,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
},
"pi_web_api_query": {
"endpoint": "/streamsets/recorded",
"period": "*-1d",
"max_count": 5,
"api_timeout": 30
}
},
"pi_web_api_server": {
"mode": "success",
"rows": [
{"name": "tag1", "timestamp": "2024-06-01T12:00:00Z", "value": 7.5}
]
}
}

View File

@@ -0,0 +1,23 @@
{
"workflow_input": {
"model_id": "13",
"model_name": "PI Empty",
"schedule_name": "pi-empty",
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
},
"pi_web_api_query": {
"endpoint": "/streamsets/recorded",
"period": "*-1d",
"max_count": 1,
"api_timeout": 30
}
},
"pi_web_api_server": {"mode": "empty"}
}

View File

@@ -0,0 +1,30 @@
{
"workflow_input": {
"model_id": "10",
"model_name": "PI Web API Happy",
"schedule_name": "pi-happy",
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000},
"tag2": {"webid": "webid2", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
},
"pi_web_api_query": {
"endpoint": "/streamsets/recorded",
"period": "*-1d",
"max_count": 10,
"api_timeout": 30
}
},
"pi_web_api_server": {
"mode": "success",
"rows": [
{"name": "tag1", "timestamp": "2024-06-01T12:00:00Z", "value": 10.5},
{"name": "tag2", "timestamp": "2024-06-01T12:00:00Z", "value": 20.3}
]
}
}

View File

@@ -0,0 +1,23 @@
{
"workflow_input": {
"model_id": "16",
"model_name": "PI Invalid Endpoint",
"schedule_name": "pi-invalid",
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
},
"pi_web_api_query": {
"endpoint": "/invalid/endpoint",
"period": "*-1d",
"max_count": 1,
"api_timeout": 30
}
},
"pi_web_api_server": {"mode": "success", "rows": []}
}

View File

@@ -0,0 +1,53 @@
{
"workflow_input": {
"model_id": "11",
"model_name": "PI Multi Tag",
"schedule_name": "pi-multi",
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag_avg": {"webid": "w1", "aggr_func": "avg", "data_range": [0, 1000], "frequency": 60000},
"tag_mdn": {"webid": "w2", "aggr_func": "mdn", "data_range": [0, 1000], "frequency": 60000},
"tag_max": {"webid": "w3", "aggr_func": "max", "data_range": [0, 1000], "frequency": 60000},
"tag_min": {"webid": "w4", "aggr_func": "min", "data_range": [0, 1000], "frequency": 60000},
"tag_lts": {"webid": "w5", "aggr_func": "lts", "data_range": [0, 1000], "frequency": 60000}
},
"pi_web_api_query": {
"endpoint": "/streamsets/recorded",
"period": "*-1d",
"max_count": 3,
"api_timeout": 30
}
},
"pi_web_api_server": {
"mode": "success",
"rows": [
{"name": "tag_avg", "timestamp": "2024-06-01T12:00:00Z", "value": 10},
{"name": "tag_avg", "timestamp": "2024-06-01T12:01:00Z", "value": 20},
{"name": "tag_avg", "timestamp": "2024-06-01T12:02:00Z", "value": 30},
{"name": "tag_mdn", "timestamp": "2024-06-01T12:00:00Z", "value": 1},
{"name": "tag_mdn", "timestamp": "2024-06-01T12:01:00Z", "value": 9},
{"name": "tag_mdn", "timestamp": "2024-06-01T12:02:00Z", "value": 5},
{"name": "tag_max", "timestamp": "2024-06-01T12:00:00Z", "value": 5},
{"name": "tag_max", "timestamp": "2024-06-01T12:01:00Z", "value": 15},
{"name": "tag_max", "timestamp": "2024-06-01T12:02:00Z", "value": 10},
{"name": "tag_min", "timestamp": "2024-06-01T12:00:00Z", "value": 50},
{"name": "tag_min", "timestamp": "2024-06-01T12:01:00Z", "value": 30},
{"name": "tag_min", "timestamp": "2024-06-01T12:02:00Z", "value": 40},
{"name": "tag_lts", "timestamp": "2024-06-01T12:00:00Z", "value": 100},
{"name": "tag_lts", "timestamp": "2024-06-01T12:01:00Z", "value": 200},
{"name": "tag_lts", "timestamp": "2024-06-01T12:02:00Z", "value": 300}
]
},
"expected_values": {
"tag_avg": 20.0,
"tag_mdn": 5.0,
"tag_max": 15.0,
"tag_min": 30.0,
"tag_lts": 300.0
}
}

View File

@@ -0,0 +1,23 @@
{
"workflow_input": {
"model_id": "15",
"model_name": "PI Timeout",
"schedule_name": "pi-timeout",
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {"webid": "webid1", "aggr_func": "avg", "data_range": [0, 100], "frequency": 60000}
},
"pi_web_api_query": {
"endpoint": "/streamsets/recorded",
"period": "*-1d",
"max_count": 1,
"api_timeout": 2
}
},
"pi_web_api_server": {"mode": "timeout", "timeout_sleep_seconds": 5}
}

View File

@@ -0,0 +1,23 @@
{
"workflow_input": {
"topic": "e2e-topic",
"model_id": "3",
"model_name": "Scouter Empty Mongo",
"schedule_name": "scouter-empty",
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {
"webid": "webid1",
"aggr_func": "avg",
"data_range": [0, 100],
"frequency": 60000
}
}
},
"raw_documents": []
}

View File

@@ -0,0 +1,38 @@
{
"workflow_input": {
"topic": "e2e-topic",
"model_id": "1",
"model_name": "Scouter E2E Model",
"schedule_name": "scouter-happy",
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {
"webid": "webid1",
"aggr_func": "avg",
"data_range": [0, 100],
"frequency": 60000
}
}
},
"raw_documents": [
{
"inserted_at": "2024-06-01 12:00:00.000000+0000",
"timestamp": "2024-06-01 12:00:00+0000",
"name": "tag1",
"value": 10.5,
"tag": "webid1"
},
{
"inserted_at": "2024-06-01 12:01:00.000000+0000",
"timestamp": "2024-06-01 12:01:00+0000",
"name": "tag1",
"value": 20.0,
"tag": "webid1"
}
]
}

View File

@@ -0,0 +1,43 @@
{
"workflow_input": {
"topic": "e2e-topic",
"model_id": "2",
"model_name": "Scouter Incremental",
"schedule_name": "scouter-incremental",
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {
"webid": "webid1",
"aggr_func": "avg",
"data_range": [0, 100],
"frequency": 60000
}
}
},
"redis_seed": {
"last_data_timestamp": "2024-06-01 10:00:00.000000+0000"
},
"raw_documents": [
{
"inserted_at": "2024-06-01 09:00:00.000000+0000",
"timestamp": "2024-06-01 09:00:00+0000",
"name": "tag1",
"value": 1.0,
"tag": "webid1"
},
{
"inserted_at": "2024-06-01 11:00:00.000000+0000",
"timestamp": "2024-06-01 11:00:00+0000",
"name": "tag1",
"value": 99.0,
"tag": "webid1"
}
],
"expected_newer_count": 1,
"expected_last_timestamp": "2024-06-01 11:00:00.000000+0000"
}

View File

@@ -0,0 +1,31 @@
{
"workflow_input": {
"topic": "e2e-topic",
"model_id": "4",
"model_name": "Scouter First Run",
"schedule_name": "scouter-first-run",
"trigger_laborious": false,
"filters": {},
"schema": "sientia_data",
"table_name": "laborious_data",
"retention_time": 3600,
"fill_missing_tags": false,
"model_tags": {
"tag1": {
"webid": "webid1",
"aggr_func": "lts",
"data_range": [0, 100],
"frequency": 60000
}
}
},
"raw_documents": [
{
"inserted_at": "2024-06-01 08:00:00.000000+0000",
"timestamp": "2024-06-01 08:00:00+0000",
"name": "tag1",
"value": 42.0,
"tag": "webid1"
}
]
}

View File

@@ -1,427 +1,138 @@
# Test Scenarios for PI Web API Scouter Workflow
# Scouter E2E scenario catalog
This document describes all possible test scenarios for the `pi_web_api_scouter` workflow and its child workflow `core_scouter`.
## Execution context
## Workflow Overview
The `pi_web_api_scouter` workflow:
1. Retrieves tag values from PI Web API
2. Delegates processing to `core_scouter` child workflow which:
- Applies data quality gates
- Aggregates data
- Groups and holds data in Redis
- Exports to PostgreSQL
- Writes metrics
- Optionally stores debug data package
- **MongoDB**, **Redis**, and **PostgreSQL** run as session-scoped testcontainers with autouse cleanup between tests.
- **PI Web API** is an in-process HTTP server (`e2e/pi_web_api_test_server.py`) speaking the wire format consumed by `PIWebAPIClient`.
- **Temporal** uses `WorkflowEnvironment.start_local()` and a single worker on `scouter-test-queue`.
- **Production code is not mocked** (except `Logger` and optional notification insert spy).
---
## 1. PI Web API Scouter - Main Workflow Scenarios
## 0. Harness smoke tests
### 1.1 Success Scenarios
Diagnostic-only checks under `e2e/test_harness_smoke.py`. They are not business scenarios; they exist to fail fast when the harness itself (Docker / containers / Temporal worker wiring) is broken, before the numbered suite runs.
#### Scenario 1.1.1: Happy Path - Complete Success
**Description**: Workflow completes successfully with valid data from PI Web API
### 0.0.1 Postgres schema ready (`test_postgres_schema_ready`)
**Input**:
- Valid `model_name`, `model_id`, `schedule_name`
- Valid `pi_web_api_query` with endpoint, period, max_count, api_timeout
- Valid `model_tags` with webids and configurations
- Valid filters, schema, table_name, retention_time
Confirms the autouse fixture executed `db_schema.sql` and `sientia_data.laborious_data` exists in the Postgres testcontainer.
**Expected Behavior**:
- `get_tag_values` returns non-empty list of records
- Workflow proceeds to `core_scouter`
- All activities execute successfully
- Data is stored in PostgreSQL
- Metrics are written
- Workflow completes without errors
### 0.0.2 Activities construct (`test_activities_construct`)
**Assertions**:
- PI Web API client called once with correct parameters
- Data exists in SQLite (PostgreSQL substitute)
- Data cached in Redis
- Metrics written
- No errors raised
Confirms the production `Activities` instance initializes against the Mongo/Redis/Postgres testcontainers without hanging (no `patch(...)` involved).
### 0.0.3 Temporal PI happy path (`test_temporal_pi_happy_path`)
End-to-end liveness check: `WorkflowEnvironment.start_local()` + worker + in-process PI server + `PIWebAPIScouter` complete without raising. Functional assertions for this flow live in scenario **2.1.1**.
---
#### Scenario 1.1.2: Success with Multiple Tags
**Description**: Workflow processes multiple tags successfully
## 1. Scouter workflow
**Input**:
- Multiple tags in `model_tags` (3+ tags)
- Each tag has valid webid, aggr_function, data_range, frequency
### 1.1.1 Happy path
**Expected Behavior**:
- All tags retrieved from PI Web API
- All tags processed through quality gates
- All tags aggregated correctly
- All tags stored in database
Seed `raw_<schedule>` with multiple documents, run `Scouter`, assert Postgres rows and Redis `last_data_timestamp:scouter:<schedule>`.
**Assertions**:
- Number of records matches number of tags
- All tags present in final data
- Aggregation applied per tag configuration
### 1.2.1 Incremental load
Pre-seed Redis timestamp; seed older and newer Mongo docs; assert only newer rows export and timestamp advances.
### 1.3.1 Empty Mongo early exit
Empty `raw_<schedule>`; workflow exits without Postgres rows or Redis timestamp key.
### 1.3.2 No Redis timestamp first run
No prior Redis key; all seeded Mongo docs load and timestamp is written after success.
---
#### Scenario 1.1.3: Success with Debug Data Package Enabled
**Description**: Workflow completes with `debug_data_package=True`
## 2. PIWebAPIScouter workflow
**Input**:
- All standard input
- `debug_data_package: True`
### 2.1.1 Happy path
**Expected Behavior**:
- Normal workflow execution
- `store_data_package` activity called
- Data package stored in Redis
PI server `success` mode with two tags; assert Postgres rows, Redis hold key, one HTTP request recorded.
**Assertions**:
- `store_data_package` called once
- Data package key exists in Redis
- Package contains both `data` and `held_data`
### 2.1.2 Multiple tags
Five tags with `avg` / `mdn` / `max` / `min` / `lts`; assert five distinct `variable` values and exact aggregated numbers in Postgres.
### 2.1.3 Debug data package
`debug_data_package=True`; assert `data_package_pi_web_api_scouter_*` Redis key with `data` and `held_data`.
### 2.2.1 Empty response early exit
Server `empty` mode; zero Postgres rows for `model_id`, one request recorded.
### 2.3.1 PI Web API connection error
Server `error` mode (HTTP 5xx); workflow fails; `PI_WEB_API_REQUEST_ERROR` notification in Mongo.
### 2.3.2 PI Web API timeout
Server `timeout` mode; workflow fails; `PI_WEB_API_REQUEST_ERROR` notification sent.
### 2.3.3 Invalid endpoint
Workflow uses `/invalid/endpoint` (404); workflow fails; `PI_WEB_API_REQUEST_ERROR` notification sent.
---
### 1.2 Early Exit Scenarios
## 3. CoreScouter subworkflow
#### Scenario 1.2.1: Empty Data from PI Web API
**Description**: PI Web API returns empty data
### 3.1.1 Complete processing success
**Input**:
- Valid configuration
- PI Web API returns empty DataFrame or empty list
Single tag, no filters; Postgres row and `held_data_*` Redis key; no `data_package_*` key.
**Expected Behavior**:
- `get_tag_values` returns empty list `[]`
- Workflow checks `if not data:` and returns early
- `core_scouter` is NOT called
- Workflow completes without error
### 3.1.2 Null values filter discard
**Assertions**:
- PI Web API called once
- `core_scouter` NOT called
- No data in PostgreSQL
- No data in Redis (except possibly from previous runs)
`NULL_VALUES_FILTER` DISCARD; one WARNING notification; only valid row in Postgres.
---
### 3.1.3 Null values filter warn
#### Scenario 1.2.2: None Returned from PI Web API
**Description**: PI Web API returns None
`NULL_VALUES_FILTER` WARN; notification sent; both rows in Postgres.
**Input**:
- Valid configuration
- PI Web API returns None
### 3.1.4 Out of bounds filter discard
**Expected Behavior**:
- `get_tag_values` returns None
- Workflow checks `if not data:` and returns early
- `core_scouter` is NOT called
`OUT_OF_BOUNDS_FILTER` DISCARD; in-range row only; WARNING notification.
**Assertions**:
- PI Web API called once
- `core_scouter` NOT called
- Workflow completes without error
### 3.1.5 Aggregation avg
---
Three points; Postgres `value` equals arithmetic mean (20.0).
### 1.3 Error Scenarios
### 3.1.6 Aggregation mdn
#### Scenario 1.3.1: PI Web API Connection Error
**Description**: PI Web API client raises connection error
Median equals 5.0 in Postgres.
**Input**:
- Valid configuration
- PI Web API client raises `PIMSRequestError` or connection exception
### 3.1.7 Aggregation max
**Expected Behavior**:
- `get_tag_values` catches exception
- Sends notification with `PI_WEB_API_REQUEST_ERROR`
- Raises exception (workflow fails after retries)
Maximum equals 15.0 in Postgres.
**Assertions**:
- Notification sent with correct error details
- Exception propagated to workflow
- Workflow fails (after retry policy exhausted)
- `core_scouter` NOT called
### 3.1.8 Aggregation min
---
Minimum equals 30.0 in Postgres.
#### Scenario 1.3.2: PI Web API Timeout
**Description**: PI Web API request times out
### 3.1.9 Aggregation lts
**Input**:
- Valid configuration
- `api_timeout` set to low value
- PI Web API takes longer than timeout
Last-by-timestamp value equals 300.0 in Postgres.
**Expected Behavior**:
- Request times out
- Exception raised
- Notification sent
- Workflow fails after retries
### 3.1.10 Fill missing tags
**Assertions**:
- Timeout exception caught
- Notification sent
- Workflow fails
`fill_missing_tags=True`; `held_data_*` contains missing tag keys with `None`.
---
### 3.1.11 Debug data package
#### Scenario 1.3.3: Invalid Endpoint
**Description**: Invalid PI Web API endpoint provided
`debug_data_package=True`; `data_package_*` Redis key decodes to dict with `data` and `held_data`.
**Input**:
- Invalid endpoint path in `pi_web_api_query`
### 3.2.1 Empty after grouping early exit
**Expected Behavior**:
- PI Web API client raises error
- Notification sent
- Workflow fails
Empty column-oriented `data`; no Postgres rows; no populated `held_data_*`.
**Assertions**:
- Error notification sent
- Workflow fails
### 3.3.1 Invalid aggregation function
---
`aggr_func= bogus`; `AGGREGATION_ISSUES` ERROR notification; bogus tag absent from Postgres.
## 2. CoreScouter - Child Workflow Scenarios
### 2.1 Success Scenarios
#### Scenario 2.1.1: Complete Processing Success
**Description**: All stages complete successfully
**Input**:
- Valid data from parent workflow
- Valid filters, model_tags, schema, table_name
- `fill_missing_tags: False`
- `debug_data_package: False`
**Expected Behavior**:
- `data_quality_gate` filters data
- `aggregate_data` aggregates by tag
- `group_and_hold_data` stores in Redis
- `export_data_to_postgres` writes to database
- `write_metrics` records metrics
- Workflow completes
**Assertions**:
- All activities called in correct order
- Data in PostgreSQL
- Data in Redis
- Metrics written
- `store_data_package` NOT called
---
#### Scenario 2.1.2: Success with Data Quality Filters
**Description**: Data quality filters applied successfully
**Input**:
- Data with some quality issues
- Filters configured with `NULL_VALUES_FILTER` or `OUT_OF_BOUNDS_FILTER`
- Policy set to `DISCARD` or `WARN`
**Expected Behavior**:
- Quality gate identifies issues
- Notification sent (WARNING level)
- If policy is `DISCARD`, bad rows removed
- Remaining data processed normally
**Assertions**:
- Quality issues detected
- Notification sent
- Bad data discarded if policy is `DISCARD`
- Good data processed
---
#### Scenario 2.1.3: Success with Different Aggregation Functions
**Description**: Different aggregation functions applied correctly
**Input**:
- Multiple tags with different `aggr_function`: `avg`, `mdn`, `max`, `min`, `lts`
- Time-series data with multiple points per tag
**Expected Behavior**:
- Each tag aggregated with its configured function
- Aggregated values correct for each function type
**Assertions**:
- `avg` calculates mean correctly
- `mdn` calculates median correctly
- `max` returns maximum value
- `min` returns minimum value
- `lts` returns latest value
---
#### Scenario 2.1.4: Success with Fill Missing Tags
**Description**: Missing tags filled with None
**Input**:
- `fill_missing_tags: True`
- Some tags missing from data
**Expected Behavior**:
- Missing tags added to `data_hold` with value `None`
- All expected tags present in final data
**Assertions**:
- Missing tags present with `None` value
- All model_tags represented in output
---
### 2.2 Early Exit Scenarios
#### Scenario 2.2.1: Empty Data After Grouping
**Description**: `group_and_hold_data` returns empty dict
**Input**:
- Data that results in empty `held_data` after grouping
**Expected Behavior**:
- `group_and_hold_data` returns `{}`
- Workflow checks `if held_data == {}:` and returns early
- `export_data_to_postgres` NOT called
- `write_metrics` NOT called
- `store_data_package` NOT called
**Assertions**:
- Early return after grouping
- No database export
- No metrics written
- Workflow completes without error
---
### 2.3 Error Scenarios
#### Scenario 2.3.1: Redis Connection Error
**Description**: Redis unavailable during `group_and_hold_data`
**Input**:
- Valid data
- Redis connection fails
**Expected Behavior**:
- `redis_repository.get()` or `redis_repository.set()` raises exception
- Notification sent with `REDIS_GET_ERROR` or `REDIS_SET_ERROR`
- Exception propagated (workflow fails after retries)
**Assertions**:
- Error notification sent
- Workflow fails
---
## 3. Activity-Specific Scenarios
> **Note**: Activity-specific scenarios are better suited for unit tests rather than e2e tests.
> These scenarios are covered indirectly through workflow e2e tests. For detailed activity testing,
> refer to the unit test suite in `tests/activities/`.
---
## 5. Test Data Requirements
### 5.1 Valid Test Data Structure
```python
{
'model_name': 'test_model',
'model_id': 'test_model_id',
'schedule_name': 'test_schedule',
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
'trigger_laborious': False,
'filters': {},
'schema': 'test_schema',
'table_name': 'test_table',
'retention_time': 3600,
'fill_missing_tags': False,
'debug_data_package': False,
'pi_web_api_query': {
'endpoint': '/streamsets/recorded',
'period': '*-1d',
'max_count': 10,
'api_timeout': 30,
},
}
```
### 5.2 Mock PI Web API Response
```python
DataFrame({
'timestamp': ['2024-01-01 12:00:00+0000', ...],
'name': ['tag1', 'tag2', ...],
'value': [10.5, 20.3, ...],
'tag': ['webid1', 'webid2', ...],
})
```
---
## 6. Test Implementation Notes
### 6.1 Test Organization
- Group tests by scenario category
- Use descriptive test names matching scenario IDs
- Share fixtures for common setup
- Use parametrized tests for similar scenarios
### 6.2 Assertions Checklist
For each scenario, verify:
- [ ] Correct activities called
- [ ] Correct parameters passed
- [ ] Expected data in storage (SQLite/Redis)
- [ ] Expected notifications sent
- [ ] Expected metrics written
- [ ] No unexpected errors
- [ ] Workflow state correct
### 6.3 Mock Configuration
- Mock PI Web API client responses
- Use fake Redis (fakeredis)
- Use fake MongoDB (mongomock)
- Use SQLite for PostgreSQL
- Mock notification handler
- Mock metrics controller
---
## 7. Priority Scenarios
### High Priority (Must Test)
1. Scenario 1.1.1: Happy Path
2. Scenario 1.2.1: Empty Data
3. Scenario 1.3.1: API Connection Error
4. Scenario 2.1.1: Complete Processing
5. Scenario 2.2.1: Empty After Grouping
6. Scenario 2.3.4: PostgreSQL Error
### Medium Priority (Should Test)
1. Scenario 1.1.3: Debug Package
2. Scenario 2.1.2: Quality Filters
3. Scenario 2.1.3: Different Aggregations
4. Scenario 3.3.5: Invalid Aggregation
### Low Priority (Nice to Have)
1. Scenario 4.1.2: Retry Success
2. Scenario 5.1.1: Large Dataset
3. Scenario 5.3.1: Concurrent Execution
### 3.3.2 Postgres export failure surfaces
Drop `value` column before run; workflow fails; ERROR notification in Mongo.

View File

@@ -1,96 +0,0 @@
"""
End-to-end tests for CoreScouter workflow - Early exit scenarios.
"""
from datetime import datetime
import pytest
from sqlalchemy import text
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from scouter.activities.activities import Activities
from scouter.workflow.sub_workflows.core_scouter import CoreScouter
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_2_2_1_empty_data_after_grouping(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
postgres_engine,
):
"""
Scenario 2.2.1: Empty Data After Grouping
group_and_hold_data returns empty dict, workflow exits early.
"""
client = temporal_test_env.client
# Prepare test data that will result in empty held_data
# This can happen if all data is filtered out or aggregation results in empty data
# Empty data must be a dict with empty lists for each column
test_data = {
'tag': [],
'name': [],
'value': [],
'timestamp': [],
}
# Prepare input data
input_data = {
'metadata': {
'metadata': {
'model_id': '1',
'model_name': 'Test Model',
'schedule_name': 'test-schedule',
'workflow_name': 'pi_web_api_scouter',
}
},
'workflow_name': 'pi_web_api_scouter',
'schedule_name': 'test-schedule',
'model_name': 'Test Model',
'model_id': '1',
'data': test_data,
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
}
# Start workflow
handle = await client.start_workflow(
CoreScouter.run,
input_data,
id=f'test-core-scouter-empty-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion (should complete without error)
await handle.result()
# Verify no data was exported to PostgreSQL (early exit)
schema_name = 'sientia_data'
table_name = 'laborious_data'
full_table_name = f"{schema_name}.{table_name}"
with postgres_engine.connect() as conn:
result = conn.execute(
text(f"SELECT COUNT(*) FROM {full_table_name} WHERE model_id = 1")
)
row_count = result.scalar()
# Should have 0 rows since export_data_to_postgres was not called
assert row_count == 0, f"Expected no data in PostgreSQL, got {row_count} rows"

View File

@@ -1,90 +0,0 @@
"""
End-to-end tests for CoreScouter workflow - Error scenarios.
"""
from datetime import datetime
from unittest.mock import AsyncMock
import pytest
from sqlalchemy import text
from temporalio.exceptions import ApplicationError, FailureError
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from scouter.activities.activities import Activities
from scouter.workflow.sub_workflows.core_scouter import CoreScouter
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_2_3_1_redis_connection_error(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
):
"""
Scenario 2.3.1: Redis Connection Error
Redis unavailable during group_and_hold_data, notification sent, workflow fails.
"""
client = temporal_test_env.client
# Mock Redis get to raise an error
original_get = test_activities.redis_repository.get
test_activities.redis_repository.get = AsyncMock(side_effect=Exception("Redis connection error"))
try:
test_data = [
{
'timestamp': '2024-01-01 12:00:00+0000',
'name': 'tag1',
'value': 10.5,
'tag': 'webid1',
},
]
input_data = {
'metadata': {
'metadata': {
'model_id': '1',
'model_name': 'Test Model',
'schedule_name': 'test-schedule',
'workflow_name': 'pi_web_api_scouter',
}
},
'workflow_name': 'pi_web_api_scouter',
'schedule_name': 'test-schedule',
'model_name': 'Test Model',
'model_id': '1',
'data': test_data,
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
}
# Start workflow
handle = await client.start_workflow(
CoreScouter.run,
input_data,
id=f'test-core-scouter-redis-error-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion - should fail
with pytest.raises((FailureError, ApplicationError, Exception)):
await handle.result()
finally:
# Restore original method
test_activities.redis_repository.get = original_get

View File

@@ -1,462 +0,0 @@
"""
End-to-end tests for CoreScouter workflow - Success scenarios.
"""
from datetime import datetime
import pandas as pd
import pytest
from sqlalchemy import inspect, text
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from scouter.activities.activities import Activities
from scouter.workflow.sub_workflows.core_scouter import CoreScouter
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_2_1_1_complete_processing_success(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
postgres_engine,
):
"""
Scenario 2.1.1: Complete Processing Success
All stages complete successfully.
"""
client = temporal_test_env.client
# Prepare test data (simulating data from parent workflow)
test_data = [
{
'timestamp': '2024-01-01 12:00:00+0000',
'name': 'tag1',
'value': 10.5,
'tag': 'webid1',
},
{
'timestamp': '2024-01-01 12:00:00+0000',
'name': 'tag2',
'value': 20.3,
'tag': 'webid2',
},
]
# Prepare input data for CoreScouter
input_data = {
'metadata': {
'metadata': {
'model_id': '1',
'model_name': 'Test Model',
'schedule_name': 'test-schedule',
'workflow_name': 'pi_web_api_scouter',
}
},
'workflow_name': 'pi_web_api_scouter',
'schedule_name': 'test-schedule',
'model_name': 'Test Model',
'model_id': '1',
'data': test_data,
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
'tag2': {
'webid': 'webid2',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
}
# Start workflow
handle = await client.start_workflow(
CoreScouter.run,
input_data,
id=f'test-core-scouter-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion
await handle.result()
# Verify data was stored in PostgreSQL
schema_name = 'sientia_data'
table_name = 'laborious_data'
full_table_name = f"{schema_name}.{table_name}"
with postgres_engine.connect() as conn:
result = conn.execute(
text(f"SELECT COUNT(*) FROM {full_table_name} WHERE model_id = 1")
)
row_count = result.scalar()
assert row_count > 0, f"Expected data in PostgreSQL, got {row_count} rows"
# Verify data was cached in Redis
keys = await test_activities.redis_repository.keys('*')
assert len(keys) > 0, "Expected data in Redis"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_2_1_2_success_with_data_quality_filters(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
postgres_engine,
):
"""
Scenario 2.1.2: Success with Data Quality Filters
Data quality filters applied successfully.
"""
client = temporal_test_env.client
# Prepare test data with some quality issues (null values, out of bounds)
test_data = [
{
'timestamp': '2024-01-01 12:00:00+0000',
'name': 'tag1',
'value': None, # Null value - should be filtered if DISCARD policy
'tag': 'webid1',
},
{
'timestamp': '2024-01-01 12:00:00+0000',
'name': 'tag2',
'value': 150.0, # Out of bounds (range is [0, 100])
'tag': 'webid2',
},
{
'timestamp': '2024-01-01 12:00:00+0000',
'name': 'tag3',
'value': 50.0, # Valid value
'tag': 'webid3',
},
]
# Prepare input data with filters
input_data = {
'metadata': {
'metadata': {
'model_id': '1',
'model_name': 'Test Model',
'schedule_name': 'test-schedule',
'workflow_name': 'pi_web_api_scouter',
}
},
'workflow_name': 'pi_web_api_scouter',
'schedule_name': 'test-schedule',
'model_name': 'Test Model',
'model_id': '1',
'data': test_data,
'trigger_laborious': False,
'filters': {
'NULL_VALUES_FILTER': {
'policy': 'DISCARD',
},
'OUT_OF_BOUNDS_FILTER': {
'policy': 'WARN',
},
},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
'tag2': {
'webid': 'webid2',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
'tag3': {
'webid': 'webid3',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
}
# Start workflow
handle = await client.start_workflow(
CoreScouter.run,
input_data,
id=f'test-core-scouter-quality-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion
await handle.result()
# Verify data was stored in PostgreSQL
# tag1 should be filtered out (null value with DISCARD policy)
# tag2 should be kept (out of bounds with WARN policy)
# tag3 should be kept (valid value)
schema_name = 'sientia_data'
table_name = 'laborious_data'
full_table_name = f"{schema_name}.{table_name}"
with postgres_engine.connect() as conn:
result = conn.execute(
text(f"SELECT COUNT(*) FROM {full_table_name} WHERE model_id = 1")
)
row_count = result.scalar()
# Should have at least tag2 and tag3 (tag1 filtered out)
assert row_count >= 2, f"Expected at least 2 records after filtering, got {row_count}"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_2_1_3_success_with_different_aggregation_functions(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
postgres_engine,
):
"""
Scenario 2.1.3: Success with Different Aggregation Functions
Different aggregation functions applied correctly.
"""
client = temporal_test_env.client
# Prepare test data with multiple points per tag for aggregation
test_data = [
{'timestamp': '2024-01-01 12:00:00+0000', 'name': 'tag_avg', 'value': 10.0, 'tag': 'webid_avg'},
{'timestamp': '2024-01-01 12:01:00+0000', 'name': 'tag_avg', 'value': 20.0, 'tag': 'webid_avg'},
{'timestamp': '2024-01-01 12:02:00+0000', 'name': 'tag_avg', 'value': 30.0, 'tag': 'webid_avg'},
{'timestamp': '2024-01-01 12:00:00+0000', 'name': 'tag_max', 'value': 5.0, 'tag': 'webid_max'},
{'timestamp': '2024-01-01 12:01:00+0000', 'name': 'tag_max', 'value': 15.0, 'tag': 'webid_max'},
{'timestamp': '2024-01-01 12:02:00+0000', 'name': 'tag_max', 'value': 10.0, 'tag': 'webid_max'},
{'timestamp': '2024-01-01 12:00:00+0000', 'name': 'tag_min', 'value': 50.0, 'tag': 'webid_min'},
{'timestamp': '2024-01-01 12:01:00+0000', 'name': 'tag_min', 'value': 30.0, 'tag': 'webid_min'},
{'timestamp': '2024-01-01 12:02:00+0000', 'name': 'tag_min', 'value': 40.0, 'tag': 'webid_min'},
{'timestamp': '2024-01-01 12:00:00+0000', 'name': 'tag_lts', 'value': 100.0, 'tag': 'webid_lts'},
{'timestamp': '2024-01-01 12:01:00+0000', 'name': 'tag_lts', 'value': 200.0, 'tag': 'webid_lts'},
{'timestamp': '2024-01-01 12:02:00+0000', 'name': 'tag_lts', 'value': 300.0, 'tag': 'webid_lts'},
]
# Prepare input data with different aggregation functions
input_data = {
'metadata': {
'metadata': {
'model_id': '1',
'model_name': 'Test Model',
'schedule_name': 'test-schedule',
'workflow_name': 'pi_web_api_scouter',
}
},
'workflow_name': 'pi_web_api_scouter',
'schedule_name': 'test-schedule',
'model_name': 'Test Model',
'model_id': '1',
'data': test_data,
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'model_tags': {
'tag_avg': {
'webid': 'webid_avg',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
'tag_max': {
'webid': 'webid_max',
'aggr_function': 'max',
'data_range': [0, 100],
'frequency': 60000,
},
'tag_min': {
'webid': 'webid_min',
'aggr_function': 'min',
'data_range': [0, 100],
'frequency': 60000,
},
'tag_lts': {
'webid': 'webid_lts',
'aggr_function': 'lts',
'data_range': [0, 100],
'frequency': 60000,
},
},
}
# Start workflow
handle = await client.start_workflow(
CoreScouter.run,
input_data,
id=f'test-core-scouter-aggr-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion
await handle.result()
# Verify data was stored in PostgreSQL
schema_name = 'sientia_data'
table_name = 'laborious_data'
full_table_name = f"{schema_name}.{table_name}"
with postgres_engine.connect() as conn:
result = conn.execute(
text(f"SELECT variable, value FROM {full_table_name} WHERE model_id = 1")
)
rows = result.fetchall()
# Should have 4 rows (one per tag)
assert len(rows) >= 4, f"Expected at least 4 records, got {len(rows)}"
# Verify aggregation was applied correctly
values_by_tag = {row[0]: float(row[1]) if row[1] is not None else None for row in rows}
# Verify all tags are present
assert 'tag_avg' in values_by_tag, "Expected tag_avg to be present"
assert 'tag_max' in values_by_tag, "Expected tag_max to be present"
assert 'tag_min' in values_by_tag, "Expected tag_min to be present"
assert 'tag_lts' in values_by_tag, "Expected tag_lts to be present"
# Note: Due to timestamp normalization in get_tag_values, aggregation may behave differently
# We verify that aggregation was applied (values exist) rather than exact values
# avg: should be between min and max of input values (10, 20, 30)
if 'tag_avg' in values_by_tag and values_by_tag['tag_avg'] is not None:
assert 10.0 <= values_by_tag['tag_avg'] <= 30.0, f"Expected avg between 10-30, got {values_by_tag['tag_avg']}"
# max: should be >= 15.0 (max of 5, 15, 10)
if 'tag_max' in values_by_tag and values_by_tag['tag_max'] is not None:
assert values_by_tag['tag_max'] >= 10.0, f"Expected max >= 10, got {values_by_tag['tag_max']}"
# min: should be <= 50.0 (min of 50, 30, 40)
if 'tag_min' in values_by_tag and values_by_tag['tag_min'] is not None:
assert values_by_tag['tag_min'] <= 50.0, f"Expected min <= 50, got {values_by_tag['tag_min']}"
# lts: should be one of the values (100, 200, 300)
if 'tag_lts' in values_by_tag and values_by_tag['tag_lts'] is not None:
assert values_by_tag['tag_lts'] in [100.0, 200.0, 300.0], f"Expected lts to be one of [100, 200, 300], got {values_by_tag['tag_lts']}"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_2_1_4_success_with_fill_missing_tags(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
postgres_engine,
):
"""
Scenario 2.1.4: Success with Fill Missing Tags
Missing tags filled with None.
"""
client = temporal_test_env.client
# Prepare test data with only some tags present
test_data = [
{
'timestamp': '2024-01-01 12:00:00+0000',
'name': 'tag1',
'value': 10.5,
'tag': 'webid1',
},
# tag2 and tag3 are missing
]
# Prepare input data with fill_missing_tags enabled
input_data = {
'metadata': {
'metadata': {
'model_id': '1',
'model_name': 'Test Model',
'schedule_name': 'test-schedule',
'workflow_name': 'pi_web_api_scouter',
}
},
'workflow_name': 'pi_web_api_scouter',
'schedule_name': 'test-schedule',
'model_name': 'Test Model',
'model_id': '1',
'data': test_data,
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': True, # Enable fill missing tags
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
'tag2': {
'webid': 'webid2',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
'tag3': {
'webid': 'webid3',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
}
# Start workflow
handle = await client.start_workflow(
CoreScouter.run,
input_data,
id=f'test-core-scouter-fill-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion
await handle.result()
# Verify all tags are present in Redis (including missing ones with None)
keys = await test_activities.redis_repository.keys('*')
assert len(keys) > 0, "Expected data in Redis"
# Verify data was stored in PostgreSQL
schema_name = 'sientia_data'
table_name = 'laborious_data'
full_table_name = f"{schema_name}.{table_name}"
with postgres_engine.connect() as conn:
result = conn.execute(
text(f"SELECT variable, value FROM {full_table_name} WHERE model_id = 1")
)
rows = result.fetchall()
# Should have all 3 tags (tag1 with value, tag2 and tag3 with None)
variables = [row[0] for row in rows]
assert 'tag1' in variables, "Expected tag1 to be present"
# Note: tags with None values might not be stored in PostgreSQL, so we just verify tag1 exists

51
e2e/test_harness_smoke.py Normal file
View File

@@ -0,0 +1,51 @@
"""Fast smoke checks for E2E fixture wiring."""
import pytest
from e2e.helpers import (
apply_pi_web_api_server_config,
load_scenario_input,
make_workflow_id,
start_and_await_workflow,
)
from e2e.pi_web_api_test_server import PIWebAPITestServer
from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
@pytest.mark.e2e
def test_postgres_schema_ready(postgres_engine):
"""Verify autouse schema setup created laborious_data."""
with postgres_engine.connect() as conn:
count = conn.exec_driver_sql(
'SELECT COUNT(*) FROM information_schema.tables '
"WHERE table_schema = 'sientia_data' AND table_name = 'laborious_data'"
).scalar()
assert count == 1
@pytest.mark.e2e
def test_activities_construct(test_activities):
"""Verify Activities initializes against testcontainers without hanging."""
assert test_activities.redis_repository is not None
assert test_activities.mongodb_repository is not None
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_temporal_pi_happy_path(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
pi_web_api_server: PIWebAPITestServer,
):
"""Minimal Temporal path: PIWebAPIScouter happy path completes."""
scenario = load_scenario_input('pi_web_api_scouter_happy_path')
apply_pi_web_api_server_config(pi_web_api_server, scenario.get('pi_web_api_server'))
await start_and_await_workflow(
temporal_env.client,
PIWebAPIScouter.run,
scenario['workflow_input'],
make_workflow_id('smoke-pi-happy'),
timeout=60.0,
)

View File

@@ -1,182 +0,0 @@
"""
End-to-end tests for PI Web API Scouter workflow - Early exit scenarios.
"""
from datetime import datetime
from unittest.mock import AsyncMock
import pandas as pd
import pytest
from sqlalchemy import text
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from scouter.activities.activities import Activities
from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_1_2_1_empty_data_from_pi_web_api(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
mock_pi_web_api_client,
postgres_engine,
):
"""
Scenario 1.2.1: Empty Data from PI Web API
PI Web API returns empty data, workflow should exit early without calling core_scouter.
"""
client = temporal_test_env.client
# Mock empty DataFrame response with proper datetime column
empty_df = pd.DataFrame({
'timestamp': pd.to_datetime([], utc=True),
'name': [],
'value': [],
'tag': [],
})
mock_pi_web_api_client.get_latest_values_df = AsyncMock(return_value=empty_df)
# Prepare test input with unique model_id to avoid conflicts
unique_id = int(datetime.now().timestamp())
input_data = {
'model_name': 'PI Web API Scouter Test Model',
'model_id': str(unique_id),
'schedule_name': 'pi-web-api-scouter-test-empty',
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'pi_web_api_query': {
'endpoint': '/streamsets/recorded',
'period': '*-1d',
'max_count': 10,
'api_timeout': 30,
},
}
# Start workflow
handle = await client.start_workflow(
PIWebAPIScouter.run,
input_data,
id=f'test-workflow-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion (should complete without error)
await handle.result()
# Verify PI Web API was called once
mock_pi_web_api_client.get_latest_values_df.assert_called_once()
# Verify no data was stored in PostgreSQL (core_scouter was not called)
schema_name = 'sientia_data'
table_name = 'laborious_data'
full_table_name = f"{schema_name}.{table_name}"
with postgres_engine.connect() as conn:
result = conn.execute(
text(f"SELECT COUNT(*) FROM {full_table_name} WHERE model_id = :model_id"),
{'model_id': unique_id}
)
row_count = result.scalar()
# Should have 0 rows since core_scouter was not called
assert row_count == 0, f"Expected no data in PostgreSQL, got {row_count} rows"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_1_2_2_none_returned_from_pi_web_api(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
mock_pi_web_api_client,
postgres_engine,
):
"""
Scenario 1.2.2: None Returned from PI Web API
PI Web API returns None, workflow should exit early without calling core_scouter.
"""
client = temporal_test_env.client
# Mock empty DataFrame response (None is not handled well, so use empty DataFrame instead)
empty_df = pd.DataFrame({
'timestamp': pd.to_datetime([], utc=True),
'name': [],
'value': [],
'tag': [],
})
mock_pi_web_api_client.get_latest_values_df = AsyncMock(return_value=empty_df)
# Prepare test input with unique model_id to avoid conflicts
unique_id = int(datetime.now().timestamp())
input_data = {
'model_name': 'PI Web API Scouter Test Model',
'model_id': str(unique_id),
'schedule_name': 'pi-web-api-scouter-test-none',
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'pi_web_api_query': {
'endpoint': '/streamsets/recorded',
'period': '*-1d',
'max_count': 10,
'api_timeout': 30,
},
}
# Start workflow
handle = await client.start_workflow(
PIWebAPIScouter.run,
input_data,
id=f'test-workflow-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion (should complete without error)
await handle.result()
# Verify PI Web API was called once
mock_pi_web_api_client.get_latest_values_df.assert_called_once()
# Verify no data was stored in PostgreSQL (core_scouter was not called)
schema_name = 'sientia_data'
table_name = 'laborious_data'
full_table_name = f"{schema_name}.{table_name}"
with postgres_engine.connect() as conn:
result = conn.execute(
text(f"SELECT COUNT(*) FROM {full_table_name} WHERE model_id = :model_id"),
{'model_id': unique_id}
)
row_count = result.scalar()
# Should have 0 rows since core_scouter was not called
assert row_count == 0, f"Expected no data in PostgreSQL, got {row_count} rows"

View File

@@ -1,206 +0,0 @@
"""
End-to-end tests for PI Web API Scouter workflow - Error scenarios.
"""
from datetime import datetime
from unittest.mock import AsyncMock
import pytest
from temporalio.exceptions import ApplicationError, FailureError
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from scouter.activities.activities import Activities
from sientia_do.repository.pi_web_api_client import PIMSRequestError
from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_1_3_1_pi_web_api_connection_error(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
mock_pi_web_api_client,
):
"""
Scenario 1.3.1: PI Web API Connection Error
PI Web API client raises connection error, notification sent, workflow fails.
"""
client = temporal_test_env.client
# Mock connection error
mock_pi_web_api_client.get_latest_values_df = AsyncMock(
side_effect=PIMSRequestError("Connection failed")
)
# Prepare test input
input_data = {
'model_name': 'PI Web API Scouter Test Model',
'model_id': '1',
'schedule_name': 'pi-web-api-scouter-test-error',
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'pi_web_api_query': {
'endpoint': '/streamsets/recorded',
'period': '*-1d',
'max_count': 10,
'api_timeout': 30,
},
}
# Start workflow
handle = await client.start_workflow(
PIWebAPIScouter.run,
input_data,
id=f'test-workflow-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion - should fail
with pytest.raises((FailureError, ApplicationError, Exception)):
await handle.result()
# Verify PI Web API was called
mock_pi_web_api_client.get_latest_values_df.assert_called()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_1_3_2_pi_web_api_timeout(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
mock_pi_web_api_client,
):
"""
Scenario 1.3.2: PI Web API Timeout
PI Web API request times out, notification sent, workflow fails.
"""
client = temporal_test_env.client
# Mock timeout error
mock_pi_web_api_client.get_latest_values_df = AsyncMock(
side_effect=TimeoutError("Request timed out")
)
# Prepare test input with low timeout
input_data = {
'model_name': 'PI Web API Scouter Test Model',
'model_id': '1',
'schedule_name': 'pi-web-api-scouter-test-timeout',
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'pi_web_api_query': {
'endpoint': '/streamsets/recorded',
'period': '*-1d',
'max_count': 10,
'api_timeout': 1, # Very short timeout
},
}
# Start workflow
handle = await client.start_workflow(
PIWebAPIScouter.run,
input_data,
id=f'test-workflow-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion - should fail
with pytest.raises((FailureError, ApplicationError, Exception)):
await handle.result()
# Verify PI Web API was called
mock_pi_web_api_client.get_latest_values_df.assert_called()
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_1_3_3_invalid_endpoint(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
mock_pi_web_api_client,
):
"""
Scenario 1.3.3: Invalid Endpoint
Invalid PI Web API endpoint provided, error raised, workflow fails.
"""
client = temporal_test_env.client
# Mock error for invalid endpoint
mock_pi_web_api_client.get_latest_values_df = AsyncMock(
side_effect=PIMSRequestError("HTTP 404 calling '/invalid/endpoint': Not found")
)
# Prepare test input with invalid endpoint
input_data = {
'model_name': 'PI Web API Scouter Test Model',
'model_id': '1',
'schedule_name': 'pi-web-api-scouter-test-invalid',
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'pi_web_api_query': {
'endpoint': '/invalid/endpoint',
'period': '*-1d',
'max_count': 10,
'api_timeout': 30,
},
}
# Start workflow
handle = await client.start_workflow(
PIWebAPIScouter.run,
input_data,
id=f'test-workflow-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion - should fail
with pytest.raises((FailureError, ApplicationError, Exception)):
await handle.result()
# Verify PI Web API was called
mock_pi_web_api_client.get_latest_values_df.assert_called()

View File

@@ -0,0 +1,222 @@
"""
End-to-end tests for the PIWebAPIScouter main workflow.
"""
import json
import pytest
from redis import Redis
from temporalio.client import WorkflowFailureError
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from e2e.conftest import E2E_DATABASE
from e2e.helpers import (
apply_pi_web_api_server_config,
count_laborious_rows,
count_notifications,
fetch_laborious_rows,
load_scenario_input,
make_workflow_id,
start_and_await_workflow,
)
from e2e.pi_web_api_test_server import PIWebAPITestServer
from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_2_1_1_happy_path(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
pi_web_api_server: PIWebAPITestServer,
postgres_engine,
redis_client: Redis,
):
scenario = load_scenario_input('pi_web_api_scouter_happy_path')
apply_pi_web_api_server_config(pi_web_api_server, scenario.get('pi_web_api_server'))
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
PIWebAPIScouter.run,
workflow_input,
make_workflow_id('pi-happy'),
)
model_id = workflow_input['model_id']
assert count_laborious_rows(postgres_engine, model_id) >= 1
assert len(redis_client.keys('held_data_*')) >= 1
assert len(pi_web_api_server.requests) == 1
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_2_1_2_multiple_tags(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
pi_web_api_server: PIWebAPITestServer,
postgres_engine,
):
scenario = load_scenario_input('pi_web_api_scouter_multiple_tags')
apply_pi_web_api_server_config(pi_web_api_server, scenario.get('pi_web_api_server'))
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
PIWebAPIScouter.run,
workflow_input,
make_workflow_id('pi-multi'),
)
rows = {
row['variable']: float(row['value'])
for row in fetch_laborious_rows(postgres_engine, workflow_input['model_id'])
}
for tag, expected in scenario['expected_values'].items():
assert tag in rows
assert rows[tag] == pytest.approx(expected)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_2_1_3_debug_data_package(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
pi_web_api_server: PIWebAPITestServer,
redis_client: Redis,
):
scenario = load_scenario_input('pi_web_api_scouter_debug_data_package')
apply_pi_web_api_server_config(pi_web_api_server, scenario.get('pi_web_api_server'))
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
PIWebAPIScouter.run,
workflow_input,
make_workflow_id('pi-debug'),
)
keys = redis_client.keys('data_package_pi_web_api_scouter_*')
assert len(keys) >= 1
payload = json.loads(redis_client.get(keys[0]))
assert 'data' in payload and 'held_data' in payload
@pytest.mark.e2e
@pytest.mark.asyncio
@pytest.mark.xfail(
strict=True,
reason='Empty PI DataFrame lacks timestamp column in get_tag_values; tracked in fix-pi-empty-response-handling',
)
async def test_scenario_2_2_1_empty_response_early_exit(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
pi_web_api_server: PIWebAPITestServer,
postgres_engine,
):
scenario = load_scenario_input('pi_web_api_scouter_empty_response')
apply_pi_web_api_server_config(pi_web_api_server, scenario.get('pi_web_api_server'))
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
PIWebAPIScouter.run,
workflow_input,
make_workflow_id('pi-empty'),
)
assert count_laborious_rows(postgres_engine, workflow_input['model_id']) == 0
assert len(pi_web_api_server.requests) == 1
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_2_3_1_pi_web_api_connection_error(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
pi_web_api_server: PIWebAPITestServer,
mongo_uri: str,
):
scenario = load_scenario_input('pi_web_api_scouter_connection_error')
apply_pi_web_api_server_config(pi_web_api_server, scenario.get('pi_web_api_server'))
workflow_input = scenario['workflow_input']
with pytest.raises(WorkflowFailureError):
await start_and_await_workflow(
temporal_env.client,
PIWebAPIScouter.run,
workflow_input,
make_workflow_id('pi-conn-error'),
)
assert (
count_notifications(
mongo_uri,
E2E_DATABASE,
notification_id='PI_WEB_API_REQUEST_ERROR',
level='ERROR',
)
>= 1
)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_2_3_2_pi_web_api_timeout(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
pi_web_api_server: PIWebAPITestServer,
mongo_uri: str,
):
scenario = load_scenario_input('pi_web_api_scouter_timeout')
apply_pi_web_api_server_config(pi_web_api_server, scenario.get('pi_web_api_server'))
workflow_input = scenario['workflow_input']
with pytest.raises(WorkflowFailureError):
await start_and_await_workflow(
temporal_env.client,
PIWebAPIScouter.run,
workflow_input,
make_workflow_id('pi-timeout'),
timeout=180.0,
)
assert (
count_notifications(
mongo_uri,
E2E_DATABASE,
notification_id='PI_WEB_API_REQUEST_ERROR',
)
>= 1
)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_2_3_3_invalid_endpoint(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
pi_web_api_server: PIWebAPITestServer,
mongo_uri: str,
):
scenario = load_scenario_input('pi_web_api_scouter_invalid_endpoint')
apply_pi_web_api_server_config(pi_web_api_server, scenario.get('pi_web_api_server'))
workflow_input = scenario['workflow_input']
with pytest.raises(WorkflowFailureError):
await start_and_await_workflow(
temporal_env.client,
PIWebAPIScouter.run,
workflow_input,
make_workflow_id('pi-invalid-endpoint'),
)
assert (
count_notifications(
mongo_uri,
E2E_DATABASE,
notification_id='PI_WEB_API_REQUEST_ERROR',
)
>= 1
)

View File

@@ -1,298 +0,0 @@
"""
End-to-end tests for PI Web API Scouter workflow - Success scenarios.
"""
from datetime import datetime
from unittest.mock import AsyncMock
import pandas as pd
import pytest
from sqlalchemy import inspect, text
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from scouter.activities.activities import Activities
from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_1_1_1_happy_path_complete_success(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
mock_pi_web_api_client,
postgres_engine,
):
"""
Scenario 1.1.1: Happy Path - Complete Success
Workflow completes successfully with valid data from PI Web API.
"""
client = temporal_test_env.client
# Prepare test input
input_data = {
'model_name': 'PI Web API Scouter Test Model',
'model_id': '1',
'schedule_name': 'pi-web-api-scouter-test',
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
'tag2': {
'webid': 'webid2',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'pi_web_api_query': {
'endpoint': '/streamsets/recorded',
'period': '*-1d',
'max_count': 10,
'api_timeout': 30,
},
}
# Start workflow
handle = await client.start_workflow(
PIWebAPIScouter.run,
input_data,
id=f'test-workflow-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion
await handle.result()
# Verify PI Web API was called once with correct parameters
mock_pi_web_api_client.get_latest_values_df.assert_called_once()
# Verify data was stored in PostgreSQL
inspector = inspect(postgres_engine)
schema_name = 'sientia_data'
table_name = 'laborious_data'
full_table_name = f"{schema_name}.{table_name}"
table_exists = inspector.has_table(table_name, schema=schema_name)
assert table_exists, f"Expected table {full_table_name} to exist in PostgreSQL"
# Verify data was inserted
with postgres_engine.connect() as conn:
result = conn.execute(text(f"SELECT COUNT(*) FROM {full_table_name}"))
row_count = result.scalar()
assert row_count > 0, f"Expected data in PostgreSQL table {full_table_name}, got {row_count} rows"
# Verify data was cached in Redis
keys = await test_activities.redis_repository.keys('*')
assert len(keys) > 0, "Expected data in Redis"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_1_1_2_success_with_multiple_tags(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
mock_pi_web_api_client,
postgres_engine,
):
"""
Scenario 1.1.2: Success with Multiple Tags
Workflow processes multiple tags successfully.
"""
client = temporal_test_env.client
# Create mock DataFrame with 5 tags
mock_df = pd.DataFrame({
'timestamp': [
'2024-01-01 12:00:00+0000',
'2024-01-01 12:01:00+0000',
'2024-01-01 12:02:00+0000',
'2024-01-01 12:03:00+0000',
'2024-01-01 12:04:00+0000',
],
'name': ['tag1', 'tag2', 'tag3', 'tag4', 'tag5'],
'value': [10.5, 20.3, 30.7, 40.1, 50.9],
'tag': ['webid1', 'webid2', 'webid3', 'webid4', 'webid5'],
})
mock_df['timestamp'] = pd.to_datetime(mock_df['timestamp'], utc=True).dt.floor('s')
mock_pi_web_api_client.get_latest_values_df = AsyncMock(return_value=mock_df)
# Prepare test input with 5 tags
input_data = {
'model_name': 'PI Web API Scouter Test Model',
'model_id': '1',
'schedule_name': 'pi-web-api-scouter-test-multiple',
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
'tag2': {
'webid': 'webid2',
'aggr_function': 'max',
'data_range': [0, 100],
'frequency': 60000,
},
'tag3': {
'webid': 'webid3',
'aggr_function': 'min',
'data_range': [0, 100],
'frequency': 60000,
},
'tag4': {
'webid': 'webid4',
'aggr_function': 'mdn',
'data_range': [0, 100],
'frequency': 60000,
},
'tag5': {
'webid': 'webid5',
'aggr_function': 'lts',
'data_range': [0, 100],
'frequency': 60000,
},
},
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'pi_web_api_query': {
'endpoint': '/streamsets/recorded',
'period': '*-1d',
'max_count': 10,
'api_timeout': 30,
},
}
# Start workflow
handle = await client.start_workflow(
PIWebAPIScouter.run,
input_data,
id=f'test-workflow-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion
await handle.result()
# Verify all tags were retrieved
mock_pi_web_api_client.get_latest_values_df.assert_called_once()
# Verify data was stored in PostgreSQL
schema_name = 'sientia_data'
table_name = 'laborious_data'
full_table_name = f"{schema_name}.{table_name}"
with postgres_engine.connect() as conn:
result = conn.execute(
text(f"SELECT COUNT(*) FROM {full_table_name} WHERE model_id = 1")
)
row_count = result.scalar()
# Should have at least 5 rows (one per tag)
assert row_count >= 5, f"Expected at least 5 records, got {row_count}"
# Verify all tags are present in the database
with postgres_engine.connect() as conn:
result = conn.execute(
text(f"SELECT DISTINCT variable FROM {full_table_name} WHERE model_id = 1")
)
variables = [row[0] for row in result.fetchall()]
expected_tags = ['tag1', 'tag2', 'tag3', 'tag4', 'tag5']
for tag in expected_tags:
assert tag in variables, f"Expected tag {tag} to be in database"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_1_1_3_success_with_debug_data_package(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
mock_pi_web_api_client,
postgres_engine,
):
"""
Scenario 1.1.3: Success with Debug Data Package Enabled
Workflow completes with debug_data_package=True.
"""
client = temporal_test_env.client
# Generate unique model_id to avoid conflicts with other tests
unique_id = int(datetime.now().timestamp() * 1000) % 1000000
# Prepare test input with debug_data_package enabled
input_data = {
'model_name': 'PI Web API Scouter Test Model',
'model_id': str(unique_id),
'schedule_name': 'pi-web-api-scouter-test-debug',
'model_tags': {
'tag1': {
'webid': 'webid1',
'aggr_function': 'avg',
'data_range': [0, 100],
'frequency': 60000,
},
},
'trigger_laborious': False,
'filters': {},
'schema': 'sientia_data',
'table_name': 'laborious_data',
'retention_time': 3600,
'fill_missing_tags': False,
'debug_data_package': True,
'pi_web_api_query': {
'endpoint': '/streamsets/recorded',
'period': '*-1d',
'max_count': 10,
'api_timeout': 30,
},
}
# Start workflow
handle = await client.start_workflow(
PIWebAPIScouter.run,
input_data,
id=f'test-workflow-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion
await handle.result()
# Verify PI Web API was called
mock_pi_web_api_client.get_latest_values_df.assert_called_once()
# Verify data package was stored in Redis
# The key format is: data_package_{workflow_name}_{schedule_name}_{timestamp}
keys = await test_activities.redis_repository.keys('data_package_*')
assert len(keys) > 0, f"Expected data package key in Redis, found keys: {keys}"
# Verify the data package contains both 'data' and 'held_data'
if keys:
package_key = keys[0]
package_data = await test_activities.redis_repository.get(package_key)
assert package_data is not None, "Expected data package to exist"
# The package should be a dict with 'data' and 'held_data' keys
assert isinstance(package_data, dict), "Expected data package to be a dict"

View File

@@ -0,0 +1,154 @@
"""
End-to-end tests for the Scouter main workflow (Mongo load path).
"""
import json
import pytest
from redis import Redis
from e2e.conftest import E2E_DATABASE
from e2e.helpers import (
count_laborious_rows,
load_scenario_input,
make_workflow_id,
seed_last_data_timestamp,
seed_raw_collection,
start_and_await_workflow,
)
from scouter.workflow.scouter import Scouter
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_1_1_happy_path(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
mongo_uri: str,
redis_client: Redis,
postgres_engine,
):
scenario = load_scenario_input('scouter_happy_path')
workflow_input = scenario['workflow_input']
seed_raw_collection(
mongo_uri,
E2E_DATABASE,
workflow_input['schedule_name'],
scenario['raw_documents'],
)
await start_and_await_workflow(
temporal_env.client,
Scouter.run,
workflow_input,
make_workflow_id('scouter-happy'),
)
model_id = workflow_input['model_id']
assert count_laborious_rows(postgres_engine, model_id) >= 1
key = f"last_data_timestamp:scouter:{workflow_input['schedule_name']}"
assert redis_client.get(key) is not None
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_2_1_incremental_load(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
mongo_uri: str,
redis_client: Redis,
postgres_engine,
):
scenario = load_scenario_input('scouter_incremental_load')
workflow_input = scenario['workflow_input']
redis_seed = scenario['redis_seed']
seed_last_data_timestamp(
redis_client,
'scouter',
workflow_input['schedule_name'],
redis_seed['last_data_timestamp'],
)
seed_raw_collection(
mongo_uri,
E2E_DATABASE,
workflow_input['schedule_name'],
scenario['raw_documents'],
)
await start_and_await_workflow(
temporal_env.client,
Scouter.run,
workflow_input,
make_workflow_id('scouter-incremental'),
)
model_id = workflow_input['model_id']
assert count_laborious_rows(postgres_engine, model_id) == scenario['expected_newer_count']
stored = json.loads(
redis_client.get(f"last_data_timestamp:scouter:{workflow_input['schedule_name']}")
)
assert stored == scenario['expected_last_timestamp']
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_3_1_empty_mongo_early_exit(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
mongo_uri: str,
redis_client: Redis,
postgres_engine,
):
scenario = load_scenario_input('scouter_empty_mongo')
workflow_input = scenario['workflow_input']
seed_raw_collection(
mongo_uri,
E2E_DATABASE,
workflow_input['schedule_name'],
scenario['raw_documents'],
)
await start_and_await_workflow(
temporal_env.client,
Scouter.run,
workflow_input,
make_workflow_id('scouter-empty'),
)
assert count_laborious_rows(postgres_engine, workflow_input['model_id']) == 0
key = f"last_data_timestamp:scouter:{workflow_input['schedule_name']}"
assert redis_client.get(key) is None
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_3_2_no_redis_timestamp_first_run(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
mongo_uri: str,
redis_client: Redis,
postgres_engine,
):
scenario = load_scenario_input('scouter_no_redis_timestamp_first_run')
workflow_input = scenario['workflow_input']
seed_raw_collection(
mongo_uri,
E2E_DATABASE,
workflow_input['schedule_name'],
scenario['raw_documents'],
)
await start_and_await_workflow(
temporal_env.client,
Scouter.run,
workflow_input,
make_workflow_id('scouter-first-run'),
)
assert count_laborious_rows(postgres_engine, workflow_input['model_id']) >= 1
key = f"last_data_timestamp:scouter:{workflow_input['schedule_name']}"
assert redis_client.get(key) is not None

View File

@@ -0,0 +1,360 @@
"""
End-to-end tests for the CoreScouter subworkflow.
"""
import json
import pytest
from redis import Redis
from sqlalchemy import text
from temporalio.client import WorkflowFailureError
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from e2e.conftest import E2E_DATABASE
from e2e.helpers import (
count_laborious_rows,
count_notifications,
fetch_laborious_rows,
load_scenario_input,
make_workflow_id,
start_and_await_workflow,
)
from scouter.activities.activities import Activities
from scouter.workflow.sub_workflows.core_scouter import CoreScouter
def _held_data_blob(test_activities: Activities, workflow_input: dict) -> dict | None:
"""
Read held_data Redis payload via the production RedisRepository.
Return:
Decoded held-data dict or None
"""
key = (
f"held_data_{workflow_input['workflow_name']}_{workflow_input['schedule_name']}"
)
metadata = workflow_input['metadata']['metadata']
return test_activities.redis_repository.get(key, metadata=metadata)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_1_1_complete_processing_success(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
postgres_engine,
redis_client: Redis,
):
scenario = load_scenario_input('core_scouter_happy_path')
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
CoreScouter.run,
workflow_input,
make_workflow_id('core-happy'),
)
assert count_laborious_rows(postgres_engine, workflow_input['model_id']) >= 1
assert _held_data_blob(test_activities, workflow_input) is not None
assert len(redis_client.keys('data_package_*')) == 0
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_1_2_null_values_filter_discard(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
mongo_uri: str,
postgres_engine,
):
scenario = load_scenario_input('core_scouter_null_values_filter_discard')
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
CoreScouter.run,
workflow_input,
make_workflow_id('core-null-discard'),
)
assert (
count_notifications(
mongo_uri,
E2E_DATABASE,
notification_id='DATA_QUALITY_GATE_ISSUES__NULL_VALUES_FILTER',
level='WARNING',
)
== 1
)
rows = fetch_laborious_rows(postgres_engine, workflow_input['model_id'])
assert len(rows) == 1
assert rows[0]['variable'] == 'tag2'
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_1_3_null_values_filter_warn(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
mongo_uri: str,
postgres_engine,
):
scenario = load_scenario_input('core_scouter_null_values_filter_warn')
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
CoreScouter.run,
workflow_input,
make_workflow_id('core-null-warn'),
)
assert (
count_notifications(
mongo_uri,
E2E_DATABASE,
notification_id='DATA_QUALITY_GATE_ISSUES__NULL_VALUES_FILTER',
level='WARNING',
)
== 1
)
assert count_laborious_rows(postgres_engine, workflow_input['model_id']) == 2
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_1_4_out_of_bounds_filter_discard(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
mongo_uri: str,
postgres_engine,
):
scenario = load_scenario_input('core_scouter_out_of_bounds_filter_discard')
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
CoreScouter.run,
workflow_input,
make_workflow_id('core-oob-discard'),
)
assert (
count_notifications(
mongo_uri,
E2E_DATABASE,
notification_id='DATA_QUALITY_GATE_ISSUES__OUT_OF_BOUNDS_FILTER',
level='WARNING',
)
== 1
)
rows = fetch_laborious_rows(postgres_engine, workflow_input['model_id'])
assert len(rows) == 1
assert rows[0]['variable'] == 'tag2'
async def _run_aggregation_scenario(
temporal_env: WorkflowEnvironment,
postgres_engine,
slug: str,
workflow_id_prefix: str,
) -> None:
scenario = load_scenario_input(slug)
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
CoreScouter.run,
workflow_input,
make_workflow_id(workflow_id_prefix),
)
rows = fetch_laborious_rows(postgres_engine, workflow_input['model_id'])
tag_name = next(iter(workflow_input['model_tags']))
value = next(row['value'] for row in rows if row['variable'] == tag_name)
assert float(value) == pytest.approx(scenario['expected_value'])
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_1_5_aggregation_avg(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
postgres_engine,
):
await _run_aggregation_scenario(
temporal_env, postgres_engine, 'core_scouter_aggregation_avg', 'core-avg'
)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_1_6_aggregation_mdn(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
postgres_engine,
):
await _run_aggregation_scenario(
temporal_env, postgres_engine, 'core_scouter_aggregation_mdn', 'core-mdn'
)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_1_7_aggregation_max(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
postgres_engine,
):
await _run_aggregation_scenario(
temporal_env, postgres_engine, 'core_scouter_aggregation_max', 'core-max'
)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_1_8_aggregation_min(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
postgres_engine,
):
await _run_aggregation_scenario(
temporal_env, postgres_engine, 'core_scouter_aggregation_min', 'core-min'
)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_1_9_aggregation_lts(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
postgres_engine,
):
await _run_aggregation_scenario(
temporal_env, postgres_engine, 'core_scouter_aggregation_lts', 'core-lts'
)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_1_10_fill_missing_tags(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
):
scenario = load_scenario_input('core_scouter_fill_missing_tags')
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
CoreScouter.run,
workflow_input,
make_workflow_id('core-fill-tags'),
)
held = _held_data_blob(test_activities, workflow_input)
assert held is not None
for tag in scenario['expected_missing_tags']:
assert tag in held
assert held[tag] is None
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_1_11_debug_data_package(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
redis_client: Redis,
):
scenario = load_scenario_input('core_scouter_debug_data_package')
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
CoreScouter.run,
workflow_input,
make_workflow_id('core-debug-pkg'),
)
keys = redis_client.keys('data_package_*')
assert len(keys) >= 1
payload = json.loads(redis_client.get(keys[0]))
assert 'data' in payload and 'held_data' in payload
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_2_1_empty_after_grouping_early_exit(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
postgres_engine,
):
scenario = load_scenario_input('core_scouter_empty_after_grouping')
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
CoreScouter.run,
workflow_input,
make_workflow_id('core-empty-group'),
)
assert count_laborious_rows(postgres_engine, workflow_input['model_id']) == 0
assert _held_data_blob(test_activities, workflow_input) is None
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_3_1_invalid_aggregation_function(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
mongo_uri: str,
postgres_engine,
):
scenario = load_scenario_input('core_scouter_invalid_aggregation')
workflow_input = scenario['workflow_input']
await start_and_await_workflow(
temporal_env.client,
CoreScouter.run,
workflow_input,
make_workflow_id('core-bad-aggr'),
)
variables = {row['variable'] for row in fetch_laborious_rows(postgres_engine, workflow_input['model_id'])}
assert 'tag_bad' not in variables
assert 'tag_ok' in variables
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_3_3_2_postgres_export_failure_surfaces(
temporal_env: WorkflowEnvironment,
temporal_worker: Worker,
mongo_uri: str,
postgres_engine,
):
scenario = load_scenario_input('core_scouter_postgres_export_failure')
workflow_input = scenario['workflow_input']
with postgres_engine.begin() as conn:
conn.execute(text('ALTER TABLE sientia_data.laborious_data DROP COLUMN value'))
with pytest.raises(WorkflowFailureError):
await start_and_await_workflow(
temporal_env.client,
CoreScouter.run,
workflow_input,
make_workflow_id('core-pg-fail'),
)
assert (
count_notifications(
mongo_uri,
E2E_DATABASE,
notification_id='ERROR_EXPORTING_DATA_TO_POSTGRES',
)
>= 1
)