SIENTIAPDE-1974
SIENTIAPDE-1974: merge main (0.8.2) into the tag database feature Brings release/SIENTIAPDE-1974 (slot v2) up to date with main (R1). Conflicts: both sides kept in orchestrator_functions.py imports, test imports and README; version follows main (bump is automated on merge). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -1333,3 +1333,96 @@ def test_filter_notification_reports_ignore_models_id_explicit_none_does_not_rai
|
||||
response = formatters.filter_notification_reports(input_data)
|
||||
|
||||
assert response['test_group_1']['notifications'] == input_data['notification_package']
|
||||
|
||||
|
||||
def _schedule_config_with_frequency_alert(frequency_alert):
|
||||
return {
|
||||
'metadata': {},
|
||||
'current_schedule_config': {'laborious': {}},
|
||||
'schedule_config': {
|
||||
'laborious': {
|
||||
'm1-simple_metrics-60m': {
|
||||
'workflow_type': 'simple_metrics',
|
||||
'model_id': 'm1',
|
||||
'model_name': 'm1',
|
||||
'frequency': '1h',
|
||||
'frequency_alert': frequency_alert,
|
||||
'updated_at': '2026-10-07',
|
||||
}
|
||||
},
|
||||
'scouter': {},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_create_schedule_config_warns_on_frequency_alert_below_floor(formatters):
|
||||
result = formatters.create_schedule_config(_schedule_config_with_frequency_alert('30s'))
|
||||
|
||||
# The configured value is kept.
|
||||
assert result['to_create']['laborious']['m1-simple_metrics-60m']['frequency_alert'] == '30s'
|
||||
formatters.send_notification.assert_called_once()
|
||||
kwargs = formatters.send_notification.call_args.kwargs
|
||||
assert kwargs['notification_id'] == 'SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR'
|
||||
assert kwargs['level'] == NotificationLevel.WARNING
|
||||
assert kwargs['metadata']['schedule_name'] == 'm1-simple_metrics-60m'
|
||||
|
||||
|
||||
def test_create_schedule_config_warns_on_frequency_alert_below_floor_on_update(formatters):
|
||||
input_data = _schedule_config_with_frequency_alert('30s')
|
||||
input_data['current_schedule_config'] = {'laborious': {'m1-simple_metrics-60m': '2026-10-06'}}
|
||||
|
||||
result = formatters.create_schedule_config(input_data)
|
||||
|
||||
assert 'm1-simple_metrics-60m' in result['to_update']['laborious']
|
||||
formatters.send_notification.assert_called_once()
|
||||
kwargs = formatters.send_notification.call_args.kwargs
|
||||
assert kwargs['notification_id'] == 'SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR'
|
||||
|
||||
|
||||
def test_create_schedule_config_no_warning_at_or_above_floor(formatters):
|
||||
formatters.create_schedule_config(_schedule_config_with_frequency_alert('5m'))
|
||||
|
||||
formatters.send_notification.assert_not_called()
|
||||
|
||||
|
||||
def _drift_schedule_config(chunk_period_configured):
|
||||
return {
|
||||
'metadata': {},
|
||||
'current_schedule_config': {'laborious': {}},
|
||||
'schedule_config': {
|
||||
'laborious': {
|
||||
'm1-drift-1m': {
|
||||
'workflow_type': 'drift',
|
||||
'model_id': 'm1',
|
||||
'model_name': 'm1',
|
||||
'frequency': '1m',
|
||||
'chunk_period': 'min',
|
||||
'chunk_period_configured': chunk_period_configured,
|
||||
'updated_at': '2026-10-27',
|
||||
}
|
||||
},
|
||||
'scouter': {},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_create_schedule_config_warns_on_invalid_chunk_period(formatters):
|
||||
result = formatters.create_schedule_config(_drift_schedule_config('hour'))
|
||||
|
||||
assert 'm1-drift-1m' in result['to_create']['laborious']
|
||||
formatters.send_notification.assert_called_once()
|
||||
kwargs = formatters.send_notification.call_args.kwargs
|
||||
assert kwargs['notification_id'] == 'DRIFT_CHUNK_PERIOD_INVALID'
|
||||
assert kwargs['level'] == NotificationLevel.WARNING
|
||||
assert kwargs['metadata']['schedule_name'] == 'm1-drift-1m'
|
||||
assert kwargs['metadata']['model_id'] == 'm1'
|
||||
assert kwargs['metadata']['model_name'] == 'm1'
|
||||
|
||||
|
||||
def test_create_schedule_config_no_chunk_period_warning_when_unchanged(formatters):
|
||||
input_data = _drift_schedule_config('hour')
|
||||
input_data['current_schedule_config'] = {'laborious': {'m1-drift-1m': '2026-10-27'}}
|
||||
|
||||
formatters.create_schedule_config(input_data)
|
||||
|
||||
formatters.send_notification.assert_not_called()
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from datetime import UTC, datetime
|
||||
from unittest.mock import ANY, AsyncMock, MagicMock, patch
|
||||
|
||||
from pytest import fixture
|
||||
from pytest import fixture, mark
|
||||
from sientia_do.notifications.models import NotificationLevel
|
||||
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ
|
||||
|
||||
@@ -265,6 +265,41 @@ def test_update_pipelines_timestamps_success(now_mock, mongo_db):
|
||||
)
|
||||
|
||||
|
||||
@patch('orchestrator.activities.mongo_db.now')
|
||||
@mark.parametrize(
|
||||
'pipelines', [[], [{'schedule_name': 'test1', 'namespace': 'test1', 'success': False}]]
|
||||
)
|
||||
def test_update_pipelines_timestamps_all_failed_stamps_nothing(now_mock, mongo_db, pipelines):
|
||||
input_data = {'updated_pipelines': pipelines}
|
||||
mongo_db.mongo_db_repository.update_many = MagicMock()
|
||||
|
||||
mongo_db.update_pipelines_timestamps(input_data)
|
||||
|
||||
# An empty filter would match and stamp every tracked schedule, so the failed one
|
||||
# would never be retried.
|
||||
mongo_db.mongo_db_repository.update_many.assert_not_called()
|
||||
|
||||
|
||||
@patch('orchestrator.activities.mongo_db.now')
|
||||
def test_update_pipelines_timestamps_mixed_stamps_only_successes(now_mock, mongo_db):
|
||||
input_data = {
|
||||
'updated_pipelines': [
|
||||
{'schedule_name': 'test1', 'namespace': 'test1', 'success': True},
|
||||
{'schedule_name': 'test2', 'namespace': 'test2', 'success': False},
|
||||
]
|
||||
}
|
||||
mongo_db.mongo_db_repository.update_many = MagicMock()
|
||||
|
||||
mongo_db.update_pipelines_timestamps(input_data)
|
||||
|
||||
mongo_db.mongo_db_repository.update_many.assert_called_once_with(
|
||||
'orchestrated_schedules',
|
||||
{'$or': [{'schedule_name': 'test1', 'namespace': 'test1'}]},
|
||||
{'$set': {'updated_at': now_mock.return_value}},
|
||||
{},
|
||||
)
|
||||
|
||||
|
||||
@patch('orchestrator.activities.mongo_db.now')
|
||||
def test_update_pipelines_timestamps_failure(now_mock, mongo_db):
|
||||
input_data = {
|
||||
@@ -362,6 +397,38 @@ def test_delete_pipelines_timestamps_success(now_mock, mongo_db):
|
||||
)
|
||||
|
||||
|
||||
@mark.parametrize(
|
||||
'pipelines', [[], [{'schedule_name': 'test1', 'namespace': 'test1', 'success': False}]]
|
||||
)
|
||||
def test_delete_pipelines_timestamps_all_failed_deletes_nothing(mongo_db, pipelines):
|
||||
input_data = {'deleted_pipelines': pipelines}
|
||||
mongo_db.mongo_db_repository.delete_many = MagicMock()
|
||||
|
||||
mongo_db.delete_pipelines_timestamps(input_data)
|
||||
|
||||
# An empty filter would wipe orchestrated_schedules; the next tick's normalize_schedules
|
||||
# would then delete and recreate every orchestrated Temporal schedule.
|
||||
mongo_db.mongo_db_repository.delete_many.assert_not_called()
|
||||
|
||||
|
||||
def test_delete_pipelines_timestamps_mixed_deletes_only_successes(mongo_db):
|
||||
input_data = {
|
||||
'deleted_pipelines': [
|
||||
{'schedule_name': 'test1', 'namespace': 'test1', 'success': True},
|
||||
{'schedule_name': 'test2', 'namespace': 'test2', 'success': False},
|
||||
]
|
||||
}
|
||||
mongo_db.mongo_db_repository.delete_many = MagicMock()
|
||||
|
||||
mongo_db.delete_pipelines_timestamps(input_data)
|
||||
|
||||
mongo_db.mongo_db_repository.delete_many.assert_called_once_with(
|
||||
'orchestrated_schedules',
|
||||
{'$or': [{'schedule_name': 'test1', 'namespace': 'test1'}]},
|
||||
{},
|
||||
)
|
||||
|
||||
|
||||
@patch('orchestrator.activities.mongo_db.now')
|
||||
def test_delete_pipelines_timestamps_failure(now_mock, mongo_db):
|
||||
input_data = {
|
||||
|
||||
@@ -3,6 +3,7 @@ from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
|
||||
|
||||
from pytest import fixture, mark
|
||||
from sientia_do.notifications.models import NotificationLevel
|
||||
from temporalio.client import ScheduleOverlapPolicy
|
||||
|
||||
from orchestrator.activities.temporal_manager import TemporalManager
|
||||
from orchestrator.utils.converters import parse_frequency
|
||||
@@ -684,3 +685,41 @@ async def test_create_schedules_tenant_runtime_queue(
|
||||
task_timeout=timedelta(seconds=300),
|
||||
typed_search_attributes=_mock_typed_search_attributes.return_value,
|
||||
)
|
||||
|
||||
|
||||
@mark.asyncio
|
||||
@patch('orchestrator.activities.temporal_manager.Schedule')
|
||||
@patch('orchestrator.activities.temporal_manager.ScheduleActionStartWorkflow')
|
||||
@patch('orchestrator.activities.temporal_manager.ScheduleIntervalSpec')
|
||||
@patch('orchestrator.activities.temporal_manager.ScheduleSpec')
|
||||
@patch('orchestrator.activities.temporal_manager.TypedSearchAttributes')
|
||||
@patch('orchestrator.activities.temporal_manager.SearchAttributePair')
|
||||
async def test_create_schedules_drift_keeps_default_skip_overlap(
|
||||
_mock_search_attribute_pair,
|
||||
_mock_typed_search_attributes,
|
||||
_mock_schedule_spec,
|
||||
_mock_schedule_interval_spec,
|
||||
_mock_schedule_action_start_workflow,
|
||||
mock_schedule,
|
||||
temporal_manager,
|
||||
):
|
||||
# The laborious DriftAnalysis cache assumes no two drift runs of a model overlap.
|
||||
input_data = {
|
||||
'schedules': {
|
||||
'laborious': {
|
||||
'm1-drift-1m': {
|
||||
'model_id': 'm1',
|
||||
'model_name': 'm1',
|
||||
'workflow_type': 'drift',
|
||||
'frequency': '1m',
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
temporal_manager.temporal_clients['laborious'].create_schedule = AsyncMock()
|
||||
|
||||
await temporal_manager.create_schedules(input_data)
|
||||
|
||||
mock_schedule.assert_called_once()
|
||||
policy = mock_schedule.call_args.kwargs.get('policy')
|
||||
assert policy is None or policy.overlap == ScheduleOverlapPolicy.SKIP
|
||||
|
||||
@@ -1,11 +1,15 @@
|
||||
from unittest.mock import call, patch
|
||||
|
||||
from pytest import mark
|
||||
|
||||
from orchestrator.utils.orchestrator_functions import (
|
||||
base_scouter,
|
||||
build_tag_config,
|
||||
chunk_period_warning,
|
||||
common_config,
|
||||
drift,
|
||||
filter_by_tag_ids,
|
||||
frequency_alert_warning,
|
||||
gather_read_tags,
|
||||
index_connector_tags,
|
||||
index_connectors,
|
||||
@@ -81,6 +85,7 @@ def test_drift():
|
||||
'source_table_name': 'laborious_data',
|
||||
'target_table_name': 'drift_metrics',
|
||||
'interval': 120,
|
||||
'chunk_period': 'min',
|
||||
'drift_metrics': ['kolmogorov_smirnov', 'jensen_shannon'],
|
||||
'execution_timeout_seconds': 300,
|
||||
'task_timeout_seconds': 300,
|
||||
@@ -89,6 +94,79 @@ def test_drift():
|
||||
assert result == expected
|
||||
|
||||
|
||||
def _drift_config(**fields):
|
||||
return {
|
||||
'workflow_type': 'drift',
|
||||
'schedule_name': 'm1-drift',
|
||||
'model_id': 'm1',
|
||||
'model': {'name': 'm1'},
|
||||
**fields,
|
||||
}
|
||||
|
||||
|
||||
def test_drift_default_window_is_frequency():
|
||||
result = drift(_drift_config())
|
||||
|
||||
assert result['frequency'] == '1m'
|
||||
assert result['interval'] == 1
|
||||
|
||||
|
||||
@mark.parametrize(
|
||||
('frequency', 'interval'),
|
||||
[('1h', 60), ('30s', 1), ('90s', 2)],
|
||||
)
|
||||
def test_drift_window_derived_from_frequency(frequency, interval):
|
||||
assert drift(_drift_config(frequency=frequency))['interval'] == interval
|
||||
|
||||
|
||||
def test_drift_explicit_interval_minutes_wins():
|
||||
result = drift(_drift_config(frequency='1m', interval_minutes=60))
|
||||
|
||||
assert result['interval'] == 60
|
||||
|
||||
|
||||
def test_drift_invalid_frequency_falls_back_to_legacy_window():
|
||||
result = drift(_drift_config(frequency='abc'))
|
||||
|
||||
assert result['frequency'] == 'abc'
|
||||
assert result['interval'] == 60
|
||||
|
||||
|
||||
def test_drift_chunk_period_defaults_to_min():
|
||||
result = drift(_drift_config())
|
||||
|
||||
assert result['chunk_period'] == 'min'
|
||||
assert 'chunk_period_configured' not in result
|
||||
|
||||
|
||||
def test_drift_chunk_period_seconds_kept():
|
||||
result = drift(_drift_config(chunk_period='s'))
|
||||
|
||||
assert result['chunk_period'] == 's'
|
||||
assert 'chunk_period_configured' not in result
|
||||
|
||||
|
||||
@mark.parametrize('chunk_period', ['hour', 5])
|
||||
def test_drift_invalid_chunk_period_falls_back_to_min(chunk_period):
|
||||
result = drift(_drift_config(chunk_period=chunk_period))
|
||||
|
||||
assert result['chunk_period'] == 'min'
|
||||
assert result['chunk_period_configured'] == chunk_period
|
||||
|
||||
|
||||
@mark.parametrize('chunk_period', [None, 'min', 's'])
|
||||
def test_chunk_period_warning_none_when_absent_or_valid(chunk_period):
|
||||
assert chunk_period_warning(chunk_period) is None
|
||||
|
||||
|
||||
@mark.parametrize('chunk_period', ['hour', 5])
|
||||
def test_chunk_period_warning_message_when_invalid(chunk_period):
|
||||
warning = chunk_period_warning(chunk_period)
|
||||
|
||||
assert warning is not None
|
||||
assert repr(chunk_period) in warning
|
||||
|
||||
|
||||
def test_simple_metrics():
|
||||
config = {
|
||||
'workflow_type': 'simple_metrics',
|
||||
@@ -115,6 +193,7 @@ def test_simple_metrics():
|
||||
'target_table_name': 'simple_metrics',
|
||||
'interval_minutes': 120,
|
||||
'metrics': ['rmse', 'mse'],
|
||||
'frequency_alert': None,
|
||||
'execution_timeout_seconds': 300,
|
||||
'task_timeout_seconds': 300,
|
||||
'runtime': 'legacy',
|
||||
@@ -122,6 +201,61 @@ def test_simple_metrics():
|
||||
assert result == expected
|
||||
|
||||
|
||||
def _simple_metrics_config(**fields):
|
||||
return {
|
||||
'workflow_type': 'simple_metrics',
|
||||
'schedule_name': 'm1-simple_metrics-60m',
|
||||
'model_id': 'm1',
|
||||
'model': {'name': 'm1'},
|
||||
'frequency': '1h',
|
||||
**fields,
|
||||
}
|
||||
|
||||
|
||||
def test_simple_metrics_escalated_uses_frequency_alert():
|
||||
config = _simple_metrics_config(
|
||||
frequency_alert='5m', adaptive_state={'escalated': True, 'healthy_streak': 0}
|
||||
)
|
||||
result = simple_metrics(config)
|
||||
assert result['frequency'] == '5m'
|
||||
assert result['frequency_alert'] == '5m'
|
||||
|
||||
|
||||
def test_simple_metrics_not_escalated_keeps_frequency():
|
||||
config = _simple_metrics_config(
|
||||
frequency_alert='5m', adaptive_state={'escalated': False, 'healthy_streak': 2}
|
||||
)
|
||||
assert simple_metrics(config)['frequency'] == '1h'
|
||||
|
||||
|
||||
def test_simple_metrics_escalated_without_frequency_alert_keeps_frequency():
|
||||
# Removing frequency_alert turns adaptive mode off even if a stale state says escalated.
|
||||
config = _simple_metrics_config(adaptive_state={'escalated': True, 'healthy_streak': 0})
|
||||
assert simple_metrics(config)['frequency'] == '1h'
|
||||
|
||||
|
||||
def test_simple_metrics_frequency_alert_without_state_keeps_frequency():
|
||||
assert simple_metrics(_simple_metrics_config(frequency_alert='5m'))['frequency'] == '1h'
|
||||
|
||||
|
||||
def test_simple_metrics_escalated_unparseable_frequency_alert_keeps_frequency():
|
||||
# An unparseable frequency would make the Temporal schedule create/update fail.
|
||||
config = _simple_metrics_config(frequency_alert='5min', adaptive_state={'escalated': True})
|
||||
assert simple_metrics(config)['frequency'] == '1h'
|
||||
|
||||
|
||||
def test_simple_metrics_escalated_below_floor_frequency_alert_is_applied():
|
||||
# Decision 7: below the floor only warns, the value is kept.
|
||||
config = _simple_metrics_config(frequency_alert='30s', adaptive_state={'escalated': True})
|
||||
assert simple_metrics(config)['frequency'] == '30s'
|
||||
|
||||
|
||||
@mark.parametrize('adaptive_state', ['yes', {'escalated': 'false'}])
|
||||
def test_simple_metrics_malformed_adaptive_state_keeps_frequency(adaptive_state):
|
||||
config = _simple_metrics_config(frequency_alert='5m', adaptive_state=adaptive_state)
|
||||
assert simple_metrics(config)['frequency'] == '1h'
|
||||
|
||||
|
||||
def test_minimal_retrain():
|
||||
config = {
|
||||
'workflow_type': 'minimal_retrain',
|
||||
@@ -1080,3 +1214,19 @@ def test_pi_web_api_scouter_with_no_timeout():
|
||||
'runtime': 'legacy',
|
||||
}
|
||||
assert result == expected
|
||||
|
||||
|
||||
@mark.parametrize(
|
||||
'frequency_alert, warns',
|
||||
[
|
||||
(None, False),
|
||||
('5m', False),
|
||||
('1m', False), # the floor itself is allowed
|
||||
('60s', False),
|
||||
('30s', True),
|
||||
('5min', True), # unparseable
|
||||
(5, True), # not a string: must warn, never raise
|
||||
],
|
||||
)
|
||||
def test_frequency_alert_warning(frequency_alert, warns):
|
||||
assert (frequency_alert_warning(frequency_alert) is not None) is warns
|
||||
|
||||
Reference in New Issue
Block a user