b17eaf972c
Update requirements-dev.txt to add E2E testing dependencies: fakeredis and mongomock for in-memory testing, and include testcontainers for PostgreSQL support.
107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
"""
|
|
End-to-end tests for PI Web API Scouter workflow.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
|
|
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_pi_web_api_scouter_e2e(
|
|
temporal_test_env: WorkflowEnvironment,
|
|
temporal_worker: Worker,
|
|
test_activities: Activities,
|
|
mock_pi_web_api_client,
|
|
postgres_engine,
|
|
):
|
|
"""
|
|
End-to-end test for PI Web API Scouter workflow.
|
|
|
|
This test:
|
|
1. Starts the workflow with test data
|
|
2. Verifies PI Web API is called
|
|
3. Verifies data flows through CoreScouter
|
|
4. Verifies data is stored in PostgreSQL (schema: sientia_data, table: laborious_data)
|
|
5. Verifies data is cached in Redis
|
|
"""
|
|
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
|
|
mock_pi_web_api_client.get_latest_values_df.assert_called_once()
|
|
|
|
# Verify data was stored in PostgreSQL
|
|
inspector = inspect(postgres_engine)
|
|
|
|
# Schema and table are created by the setup_postgres_schema_and_table fixture
|
|
schema_name = 'sientia_data'
|
|
table_name = 'laborious_data'
|
|
full_table_name = f"{schema_name}.{table_name}"
|
|
|
|
# Check if table exists in the schema
|
|
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"
|