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

SIENTIAPDE-1646

Browse Source 
Refactor ModelMetrics to utilize DriftAnalysis for drift detection

- Replaced ModelAnalysis with DriftAnalysis in the ModelMetrics class to enhance drift detection capabilities.
- Updated method signatures and documentation to reflect the changes in target_name and return values.
- Adjusted data handling to ensure compatibility with the new analysis methods and improved clarity in the drift metrics dataframe preparation.
This commit is contained in:
vitor-aignosi
2026-05-08 16:39:12 -03:00
parent c9ac25ea80
commit ee9c71cbbb
5 changed files with 209 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
e2e/conftest.py
View File

@@ -1,7 +1,5 @@
"""Pytest configuration and fixtures for E2E tests."""
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from unittest.mock import MagicMock
@@ -16,7 +14,6 @@ from testcontainers.postgres import PostgresContainer
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from e2e.opc_test_server import OpcE2ETestServer
from laborious.activities.activities import Activities
from laborious.workflows.drift import Drift
from laborious.workflows.minimal_retrain import MinimalRetrain
@@ -464,140 +461,3 @@ async def temporal_worker_minimal_retrain(temporal_test_env, test_activities):
yield worker
def _connect_activities_to_opc_server(activities: Activities, server_url: str) -> None:
"""
Initialize OPC repositories and block until the E2E server session is ready.
Runs synchronously (typically via ``asyncio.to_thread``) so the asyncua test
server event loop is not blocked during ``Client.connect()``.
Args:
activities (Activities): Worker activities under test.
server_url (str): ``opc.tcp://`` URL from ``OpcE2ETestServer``.
"""
activities.init_opc()
repo = activities.opc_repository['1']
deadline = time.monotonic() + 30.0
while time.monotonic() < deadline:
if repo._session_ready.is_set():
return
connected, _ = repo.connect()
if connected:
return
time.sleep(0.5)
raise RuntimeError(f'Could not connect OpcRepository to OPC E2E server at {server_url}')
def _build_e2e_opc_config(server_url: str) -> dict[str, dict]:
"""
OPC server config for E2E Activities pointing at an in-process asyncua server.
Args:
server_url (str): ``opc.tcp://`` endpoint from ``OpcE2ETestServer``.
Return:
dict: ``opc_config`` payload for ``Activities`` (server id ``1``).
"""
return {
'1': {
'id': '1',
'server_name': 'e2e_opcua',
'url': server_url,
'server_uri': server_url,
'cert_path': None,
'private_key_path': None,
'server_cert_path': None,
'reconnection_interval': 0,
}
}
@pytest_asyncio.fixture
async def opc_e2e_server():
"""In-process asyncua server with writable prediction/confidence nodes."""
server = OpcE2ETestServer()
await server.start()
await asyncio.sleep(0.5)
try:
yield server
finally:
await server.stop()
@pytest_asyncio.fixture(scope='function')
async def test_activities_real_opc(
postgres_container,
minio_container,
mock_logger,
notification_handler,
metrics_controller,
mlflow_repository_stub,
plugin_store_stub,
pi_web_api_client_stub,
opc_e2e_server: OpcE2ETestServer,
):
"""Activities with real OpcRepository connected to the in-process OPC UA server."""
minio_client = minio_container.get_client()
if not minio_client.bucket_exists('test-bucket'):
minio_client.make_bucket('test-bucket')
minio_port = minio_container.get_exposed_port(9000)
activities = Activities(
postgres_config={
'host': 'localhost',
'port': int(postgres_container.get_exposed_port(5432)),
'user': postgres_container.username,
'password': postgres_container.password,
'dbname': postgres_container.dbname,
'min_connections': 1,
'max_connections': 5,
},
plugin_store=plugin_store_stub,
minio_config={
'endpoint_url': f'localhost:{minio_port}',
'access_key': 'minioadmin',
'secret_key': 'minioadmin',
'default_bucket': 'test-bucket',
'retention_hours': 24,
'secure': False,
},
opc_config=_build_e2e_opc_config(opc_e2e_server.url),
pi_web_api_config={
'base_url': 'http://localhost:8080',
'auth_type': 'bearer',
'auth_token': 'test_token',
},
logger=mock_logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
mlflow_repository=mlflow_repository_stub,
)
activities.pi_web_api_client = pi_web_api_client_stub
await asyncio.to_thread(_connect_activities_to_opc_server, activities, opc_e2e_server.url)
try:
yield activities
finally:
await asyncio.to_thread(_teardown_real_opc_activities, activities)
def _teardown_real_opc_activities(activities: Activities) -> None:
"""Disconnect OPC sessions and shut down activities (sync, for asyncio.to_thread)."""
for opc_repo in activities.opc_repository.values():
opc_repo.disconnect()
activities.shutdown()
@pytest_asyncio.fixture(scope='function')
async def temporal_worker_real_opc(temporal_test_env, test_activities_real_opc):
"""Temporal worker using real OpcRepository against the in-process OPC UA server."""
with ThreadPoolExecutor(max_workers=32) as activity_executor:
async with Worker(
temporal_test_env.client,
task_queue='test-queue',
workflows=[PredictionsBatch, PredictionProcess, FormatAndExportPrediction],
activities=_worker_activity_list(test_activities_real_opc),
activity_executor=activity_executor,
) as worker:
yield worker

81
e2e/scenarios.md
View File

@@ -10,27 +10,6 @@ It is a functional reference of scenario behavior, inputs, and expected outcomes
- Tests run under `e2e/` and are marked with `@pytest.mark.integration`.
- PostgreSQL and MinIO are provisioned with testcontainers.
- `test_minio_offload.py` uses real MinIO I/O; other scenario suites may use stubs/mocks for optional outputs.
- Real OPC UA scenarios use `@pytest.mark.opc` and an in-process asyncua server (`e2e/test_opc_real_server.py`).
### Local validation
Use the existing project virtualenv and the shared `validate` script for unit/quality gates; run E2E separately (Docker required).
```bash
source ./venv/bin/activate
# Auto-fix + static checks (no pytest)
validate --fix --project-name=laborious
# Full unit + quality gate
validate --project-name=laborious
# E2E (integration)
pytest e2e/ --override-ini testpaths=e2e -m integration
# E2E (real OPC server only)
pytest e2e/test_opc_real_server.py --override-ini testpaths=e2e -m opc
```
---
@@ -106,15 +85,11 @@ Source: `e2e/test_predictions_batch_prediction_process.py`
- Workflow exits without export.
#### 2.1.3 REPEAT with history
**Summary**: Prior prediction is reused; outcome depends on batch vs history timestamp.
**Summary**: Prior prediction is reused.
**Description**:
- Input gate returns `REPEAT`.
- `repeat_last_prediction` inserts a row using ``last_timestamp`` from the batch payload (max timestamp in `laborious_data` for the query), not the copied row’s timestamp.
**Tests**:
- **Collision**: batch `last_timestamp` equals the historical prediction row’s `timestamp` → Postgres `unique_model_id_timestamp` violation; workflow fails; still one row.
- **Distinct batch time**: laborious rows are stamped later than the historical prediction → second row inserted; same prediction fields as the first (see `assert_repeat`).
- `repeat_last_prediction` path is executed using existing historical row.
#### 2.1.4 REPEAT without history
**Summary**: Repeat requested but no previous prediction exists.
@@ -137,8 +112,6 @@ Source: `e2e/test_predictions_batch_prediction_process.py`
#### 2.2.3 REPEAT on transform response error
**Summary**: Transform response error triggers repeat-last-prediction path.
**Tests**: Same timestamp collision vs distinct batch timestamp as §2.1.3 (`*_fails` / `*_inserts_second_row`).
#### 2.2.4 STOP on transform content NaN
**Summary**: Content gate (`NAN_VALUES`) blocks on all-NaN transform payload.
@@ -153,8 +126,6 @@ Source: `e2e/test_predictions_batch_prediction_process.py`
#### 2.3.3 REPEAT on predict response error
**Summary**: Predict response error routes to repeat-last-prediction.
**Tests**: Same timestamp collision vs distinct batch timestamp as §2.1.3 (`*_fails` / `*_inserts_second_row`).
### 2.4.1 Priority Conflict Resolution
**Summary**: Deterministic selection when multiple filters produce different flags.
@@ -216,30 +187,6 @@ Source: `e2e/test_predictions_batch_format_export.py`
- Workflow completes.
- Prediction persisted with PI error confidence and descriptive comment.
#### 3.2.4 OPC session / channel error (confidence 14)
**Summary**: Tier-1 `BadSessionIdInvalid` (or equivalent session error) degrades the prediction without failing the workflow.
**Sources**:
- Mock: `e2e/test_predictions_batch_format_export.py::test_scenario_3_2_4_opc_session_bad_mock`
- Real server: `e2e/test_opc_real_server.py::test_scenario_3_2_4_opc_session_bad_real_server` (`@pytest.mark.opc`)
**Expected Outcome**:
- Workflow completes.
- `prediction_confidence` is 14.
- Comments contain `OPC UA session/channel error: BadSessionIdInvalid`.
#### 3.2.5 OPC write blocked during reconnect (confidence 14)
**Summary**: While reconnect holds the repository connection lock, writes fail fast with `reconnect_in_progress`.
**Sources**:
- Mock: `e2e/test_predictions_batch_format_export.py::test_scenario_3_2_5_opc_reconnect_in_progress_mock`
- Real server: `e2e/test_opc_real_server.py::test_scenario_3_2_5_opc_write_blocked_during_reconnect_real_server` (`@pytest.mark.opc`)
**Expected Outcome**:
- Workflow completes.
- `prediction_confidence` is 14.
- Comments contain `OPC UA reconnect in progress`.
### 3.3.1 Combined Optional Outputs (PI + OPC)
**Summary**: Both external output channels are enabled together.
@@ -349,33 +296,15 @@ first 30% of target rows as reference.
- The workflow surfaces the `ValueError` ("Invalid chunk period: ...").
- No rows are persisted.
#### D.4.3a Insufficient drift metrics while target has rows
**Summary**: Laborious raises when the merged drift table is empty but the
target window is non-empty (`Insufficient drift data:` + notification
`MODEL_METRICS_DRIFT_INSUFFICIENT_DATA`).
**Description**:
- The e2e patches `ModelMetrics.get_drift_metrics` to return an empty
DataFrame, simulating a ``sientia_model`` path that emits no rows.
**Expected Outcome**:
- Workflow fails; no rows in `sientia_data.drift_metrics`.
#### D.4.3b `chunk_period='s'` preserves seconds in `chunk_start_date`
**Summary**: Target data includes sub-minute spacing across several minutes;
the activity uses `chunk_period='s'`.
#### D.4.3 `chunk_period='s'` preserves seconds in `chunk_start_date`
**Summary**: Target data spans two minutes with samples at second-30
boundaries; the activity is configured with `chunk_period='s'`.
**Expected Outcome**:
- At least one persisted `chunk_start_date` carries `seconds=30`, proving
that the analyzer chunked at sub-minute granularity and the ISO-text
serialization preserved the boundary.
**Note**: The e2e patches `DriftAnalysis._chunk_dataframe` to **skip empty**
`pd.Grouper(freq='s')` buckets. The stock implementation iterates every
second between min/max timestamps, producing empty chunks and NaT rows that
`calculate_drift` filters away entirely. The durable fix belongs in
`sientia_model`.
---
## 6. Simple Metrics Workflow Scenarios