gitea_admin/sientia-dataops-scouter_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1478

Browse Source 
Update pytest_asyncio fixture scopes in conftest.py for improved test isolation and add asyncio_default_fixture_loop_scope in pyproject.toml. Remove outdated scenarios from scenarios.md and delete unused test files for cleaner codebase.
This commit is contained in:
vitor-aignosi
2026-01-08 16:13:59 -03:00
parent cf7f9fb63c
commit 7958afbadf
9 changed files with 14 additions and 1085 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
e2e/conftest.py
View File

@@ -30,7 +30,7 @@ TEST_MONGODB_CONNECTION_STRING = 'mongodb://localhost:27017'
TEST_DATABASE_NAME = 'test_db'
@pytest.fixture(scope='session')
@pytest_asyncio.fixture(scope='session')
def postgres_container():
"""
Create a PostgreSQL container using testcontainers.
@@ -44,7 +44,7 @@ def postgres_container():
postgres.stop()
@pytest.fixture
@pytest_asyncio.fixture
def postgres_engine(postgres_container):
"""
Create SQLAlchemy engine for PostgreSQL test database.
@@ -81,7 +81,6 @@ def _create_schema_and_table(engine):
# Create table WITHOUT partitioning (simpler for tests)
# Same structure as production, but without PARTITION BY RANGE
# Use UNIQUE constraint directly since table is not partitioned
create_table_sql = f"""
CREATE TABLE IF NOT EXISTS {schema_name}.{table_name} (
id SERIAL NOT NULL,
@@ -90,8 +89,7 @@ def _create_schema_and_table(engine):
value numeric NULL,
"timestamp" timestamptz NOT NULL,
created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL,
PRIMARY KEY (id, created_at),
UNIQUE (model_id, timestamp, variable)
PRIMARY KEY (id, created_at)
);
"""
@@ -99,7 +97,7 @@ def _create_schema_and_table(engine):
# Transaction is automatically committed when exiting the 'with' block
@pytest.fixture(autouse=True)
@pytest_asyncio.fixture(autouse=True)
def setup_postgres_schema_and_table(postgres_engine):
"""
Automatically create necessary schema and table before each test.
@@ -108,15 +106,14 @@ def setup_postgres_schema_and_table(postgres_engine):
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 with a UNIQUE constraint
directly in the table definition, which is simpler and avoids issues
Note: For tests, we use a non-partitioned table which is simpler and avoids issues
with pandas to_sql recognizing partitioned tables.
"""
_create_schema_and_table(postgres_engine)
yield
@pytest.fixture
@pytest_asyncio.fixture
def mock_logger():
"""Mock logger for testing."""
logger = MagicMock(spec=Logger)
@@ -128,7 +125,7 @@ def mock_logger():
return logger
@pytest.fixture
@pytest_asyncio.fixture
def mock_mongo_client():
"""
Mock MongoDB client to avoid real connections.
@@ -153,7 +150,7 @@ def mock_mongo_client():
return mock_client
@pytest.fixture
@pytest_asyncio.fixture
def notification_handler(mock_logger, mock_mongo_client):
"""
Create a real NotificationHandler instance with mocked MongoDB client.
@@ -173,7 +170,7 @@ def notification_handler(mock_logger, mock_mongo_client):
handler.shutdown()
@pytest.fixture
@pytest_asyncio.fixture
def metrics_controller(mock_logger):
"""
Create a real MetricsController instance.
@@ -186,7 +183,7 @@ def metrics_controller(mock_logger):
# MetricsController might have cleanup, but it's optional
@pytest.fixture
@pytest_asyncio.fixture
def mock_pi_web_api_client():
"""Mock PI Web API client."""
mock_client = MagicMock()
@@ -212,8 +209,8 @@ def mock_pi_web_api_client():
return mock_client
@pytest_asyncio.fixture
async def test_activities(
@pytest_asyncio.fixture(scope='function')
def test_activities(
postgres_engine,
postgres_container,
mock_logger,

19
e2e/scenarios.md
View File

@@ -299,25 +299,6 @@ The `pi_web_api_scouter` workflow:
---
#### Scenario 2.2.2: Zero Affected Rows After Export
**Description**: PostgreSQL export returns zero affected rows
**Input**:
- Data that results in `affected_rows: 0` from export
**Expected Behavior**:
- `export_data_to_postgres` returns `{'affected_rows': 0}`
- Workflow checks `if data_exported.get('affected_rows', 0) <= 0:` and returns early
- `write_metrics` NOT called
- `store_data_package` NOT called
**Assertions**:
- Early return after export
- No metrics written
- Workflow completes without error
---
### 2.3 Error Scenarios
#### Scenario 2.3.1: Redis Connection Error

97
e2e/test_core_scouter_early_exit.py
View File

@@ -94,100 +94,3 @@ async def test_scenario_2_2_1_empty_data_after_grouping(
# 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"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_2_2_2_zero_affected_rows_after_export(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
postgres_engine,
):
"""
Scenario 2.2.2: Zero Affected Rows After Export
PostgreSQL export returns zero affected rows, workflow exits early.
Note: This scenario is hard to test directly in e2e because we'd need to
simulate a conflict or other condition that results in 0 affected rows.
We'll test by inserting duplicate data first, then running the workflow again.
"""
client = temporal_test_env.client
# First, insert some data directly to create a conflict scenario
test_data = [
{
'timestamp': '2024-01-01 12:00:00+0000',
'name': 'tag1',
'value': 10.5,
'tag': 'webid1',
},
]
# Insert data directly into PostgreSQL to create duplicates
schema_name = 'sientia_data'
table_name = 'laborious_data'
full_table_name = f"{schema_name}.{table_name}"
with postgres_engine.connect() as conn:
conn.execute(
text(f"""
INSERT INTO {full_table_name} (model_id, variable, value, timestamp)
VALUES (1, 'tag1', 10.5, '2024-01-01 12:00:00+00:00')
ON CONFLICT (model_id, timestamp, variable) DO NOTHING
""")
)
conn.commit()
# Prepare input data with the same data (will result in conflict)
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-zero-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion (should complete without error)
await handle.result()
# Verify the count didn't increase (conflict handled, 0 affected rows)
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 still have 1 row (the original one, duplicate was ignored)
assert row_count == 1, f"Expected 1 row (duplicate ignored), got {row_count}"

99
e2e/test_core_scouter_errors.py
View File

@@ -88,102 +88,3 @@ async def test_scenario_2_3_1_redis_connection_error(
# Restore original method
test_activities.redis_repository.get = original_get
@pytest.mark.asyncio
@pytest.mark.integration
async def test_scenario_2_3_2_postgresql_unique_constraint_violation(
temporal_test_env: WorkflowEnvironment,
temporal_worker: Worker,
test_activities: Activities,
postgres_engine,
):
"""
Scenario 2.3.2: PostgreSQL Unique Constraint Violation
Duplicate data violates unique constraint, handled gracefully with ON CONFLICT DO NOTHING.
"""
client = temporal_test_env.client
# Generate unique model_id to avoid conflicts with other tests
unique_id = int(datetime.now().timestamp() * 1000) % 1000000
# First, insert data directly to create a duplicate
schema_name = 'sientia_data'
table_name = 'laborious_data'
full_table_name = f"{schema_name}.{table_name}"
with postgres_engine.connect() as conn:
conn.execute(
text(f"""
INSERT INTO {full_table_name} (model_id, variable, value, timestamp)
VALUES (:model_id, 'tag1', 10.5, '2024-01-01 12:00:00+00:00')
ON CONFLICT (model_id, timestamp, variable) DO NOTHING
"""),
{'model_id': unique_id}
)
conn.commit()
# Prepare the same data to trigger conflict
test_data = [
{
'timestamp': '2024-01-01 12:00:00+0000',
'name': 'tag1',
'value': 10.5,
'tag': 'webid1',
},
]
input_data = {
'metadata': {
'metadata': {
'model_id': str(unique_id),
'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': str(unique_id),
'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-conflict-{datetime.now().timestamp()}',
task_queue='test-queue',
)
# Wait for workflow completion - should complete without error
# (conflict is handled gracefully with ON CONFLICT DO NOTHING)
await handle.result()
# Verify no exception was raised and workflow completed
# The duplicate should be ignored (0 affected rows), but workflow should complete
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 still have 1 row (duplicate was ignored)
assert row_count == 1, f"Expected 1 row (duplicate ignored), got {row_count}"

106
e2e/test_pi_web_api_scouter.py
View File

@@ -1,106 +0,0 @@
"""
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"

2
e2e/test_pi_web_api_scouter_errors.py
View File

@@ -11,7 +11,7 @@ from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from scouter.activities.activities import Activities
from scouter.utils.clients.pi_web_api_client import PIMSRequestError
from sientia_do.repository.pi_web_api_client import PIMSRequestError
from scouter.workflow.pi_web_api_scouter import PIWebAPIScouter