e1b94b1da4
- interval_minutes is untyped Mongo config: a string or null now warns and skips the adaptive state instead of failing every run (TypeError). - A hand-edited healthy_streak (null, text, inf) restarts the count and keeps escalated, instead of raising on every run. - README: step 4 runs on every schedule, so a MongoDB failure fails the run.
358 lines
13 KiB
Python
358 lines
13 KiB
Python
from datetime import datetime
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
from pytest import fixture, mark, raises
|
|
from sientia_do.observability.metrics_controller import MetricsController
|
|
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
|
|
|
|
from laborious.activities.mongo import MongoActivities
|
|
|
|
|
|
@fixture
|
|
def mongo() -> MagicMock:
|
|
return MagicMock()
|
|
|
|
|
|
@fixture
|
|
def prediction_metadata(mongo: MagicMock):
|
|
activity = MongoActivities(
|
|
mongo_repository=mongo,
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metrics_controller=MagicMock(spec=MetricsController),
|
|
)
|
|
activity.error = MagicMock()
|
|
return activity
|
|
|
|
|
|
metadata = {
|
|
'metadata': {
|
|
'model_id': 'test_model',
|
|
'model_name': 'test_model',
|
|
'workflow_name': 'test_workflow',
|
|
'schema_name': 'test_schedule',
|
|
},
|
|
}
|
|
|
|
|
|
def test_write_prediction_metadata_no_documents_skips_mongo(prediction_metadata, mongo: MagicMock):
|
|
# Act
|
|
result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': []})
|
|
|
|
# Assert
|
|
assert result is None
|
|
mongo.insert.assert_not_called()
|
|
|
|
|
|
def test_write_prediction_metadata_missing_documents_key_skips_mongo(
|
|
prediction_metadata, mongo: MagicMock
|
|
):
|
|
# Act
|
|
result = prediction_metadata.write_prediction_metadata({**metadata})
|
|
|
|
# Assert
|
|
assert result is None
|
|
mongo.insert.assert_not_called()
|
|
|
|
|
|
def test_write_prediction_metadata_documents_with_no_filters_skips_mongo(
|
|
prediction_metadata, mongo: MagicMock
|
|
):
|
|
# Arrange: a falsy document (None) and one with an empty `filters` list — neither contributes
|
|
# anything to write.
|
|
documents = [None, {'filters': []}]
|
|
|
|
# Act
|
|
result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': documents})
|
|
|
|
# Assert
|
|
assert result is None
|
|
mongo.insert.assert_not_called()
|
|
|
|
|
|
def test_write_prediction_metadata_not_configured_returns_none():
|
|
# Arrange
|
|
activity = MongoActivities(
|
|
mongo_repository=None,
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metrics_controller=MagicMock(spec=MetricsController),
|
|
)
|
|
activity.error = MagicMock()
|
|
document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
|
|
|
|
# Act
|
|
result = activity.write_prediction_metadata({**metadata, 'documents': [document]})
|
|
|
|
# Assert
|
|
assert result is None
|
|
activity.error.assert_not_called()
|
|
|
|
|
|
def test_write_prediction_metadata_success_generates_id_and_inserts(
|
|
prediction_metadata, mongo: MagicMock
|
|
):
|
|
# Arrange
|
|
document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 3}]}
|
|
|
|
# Act
|
|
result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': [document]})
|
|
|
|
# Assert: a plain 24-hex-char string id (matches predictions.metadata_id's varchar(24)),
|
|
# not a BSON ObjectId — `insert(...)` never hands one back, so it's generated client-side.
|
|
assert result is not None
|
|
assert len(result) == 24
|
|
int(result, 16) # raises ValueError if not hex
|
|
|
|
collection, inserted_document, inserted_metadata = mongo.insert.call_args.args
|
|
assert collection == MongoActivities._PREDICTION_METADATA_COLLECTION
|
|
assert inserted_document == {'filters': document['filters'], '_id': result}
|
|
assert inserted_metadata == metadata['metadata']
|
|
|
|
|
|
def test_write_prediction_metadata_combines_multiple_documents(
|
|
prediction_metadata, mongo: MagicMock
|
|
):
|
|
# Arrange: two documents (e.g. from two different sources contributing metadata for the same
|
|
# prediction cycle) — their `filters` entries are flattened into one combined document.
|
|
documents = [
|
|
{'filters': [{'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0}]},
|
|
{
|
|
'filters': [
|
|
{'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}
|
|
]
|
|
},
|
|
]
|
|
|
|
# Act
|
|
result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': documents})
|
|
|
|
# Assert
|
|
assert result is not None
|
|
_, inserted_document, _ = mongo.insert.call_args.args
|
|
assert inserted_document['filters'] == [
|
|
{'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0},
|
|
{'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2},
|
|
]
|
|
|
|
|
|
def test_write_prediction_metadata_insert_failure_swallowed(prediction_metadata, mongo: MagicMock):
|
|
# Arrange
|
|
mongo.insert.side_effect = RuntimeError('connection reset')
|
|
document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
|
|
|
|
# Act
|
|
result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': [document]})
|
|
|
|
# Assert: the prediction write must never be blocked or failed by this
|
|
assert result is None
|
|
prediction_metadata.error.assert_called_once()
|
|
|
|
|
|
@patch.object(SientiaMonitoring, 'shutdown')
|
|
def test_close_disposes_mongo_repository(mock_shutdown):
|
|
"""``MongoActivities.close`` should close the Mongo client and clear the reference."""
|
|
mongo = MagicMock()
|
|
activity = MongoActivities(
|
|
mongo_repository=mongo,
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metrics_controller=MagicMock(spec=MetricsController),
|
|
)
|
|
activity.close()
|
|
mongo.close.assert_called_once()
|
|
assert activity.mongo_repository is None
|
|
# Other instances collected by GC during the test also hit the patched shutdown.
|
|
assert mock_shutdown.call_args_list.count(call(activity)) == 1
|
|
|
|
|
|
@patch.object(SientiaMonitoring, 'shutdown')
|
|
def test_close_without_mongo_repository(mock_shutdown):
|
|
"""When no Mongo repository is configured, ``close`` only shuts down monitoring."""
|
|
activity = MongoActivities(
|
|
mongo_repository=None,
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metrics_controller=MagicMock(spec=MetricsController),
|
|
)
|
|
activity.close()
|
|
# Other instances collected by GC during the test also hit the patched shutdown.
|
|
assert mock_shutdown.call_args_list.count(call(activity)) == 1
|
|
|
|
|
|
_SM_METADATA = {**metadata['metadata'], 'schedule_name': 'm1-simple_metrics-60m'}
|
|
_BREACH_INPUT = {
|
|
'metadata': _SM_METADATA,
|
|
'simple_metrics': [{'metric': 'rmse', 'value': 2.0}],
|
|
'thresholds': {'rmse_max': 1.0},
|
|
'interval_minutes': 60,
|
|
}
|
|
_HEALTHY_INPUT = {**_BREACH_INPUT, 'simple_metrics': [{'metric': 'rmse', 'value': 0.5}]}
|
|
_EMPTY_INPUT = {**_BREACH_INPUT, 'simple_metrics': []}
|
|
|
|
|
|
@fixture
|
|
def mongo_activities(mongo: MagicMock):
|
|
activity = MongoActivities(
|
|
mongo_repository=mongo,
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metrics_controller=MagicMock(spec=MetricsController),
|
|
)
|
|
activity.info = MagicMock()
|
|
activity.warning = MagicMock()
|
|
return activity
|
|
|
|
|
|
def _pipeline(**fields) -> dict:
|
|
return {'schedule_name': 'm1-simple_metrics-60m', 'frequency': '1h', **fields}
|
|
|
|
|
|
def _set_of(mongo: MagicMock) -> dict:
|
|
collection, query, update, _metadata = mongo.update_one.call_args.args
|
|
assert collection == 'pipelines'
|
|
assert query == {'schedule_name': 'm1-simple_metrics-60m'}
|
|
return update['$set']
|
|
|
|
|
|
def test_adaptive_state_no_mongo_is_noop():
|
|
activity = MongoActivities(
|
|
mongo_repository=None,
|
|
logger=MagicMock(),
|
|
notification_handler=MagicMock(),
|
|
metrics_controller=MagicMock(spec=MetricsController),
|
|
)
|
|
assert activity.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None
|
|
|
|
|
|
def test_adaptive_state_missing_pipeline_doc_is_noop(mongo_activities, mongo):
|
|
mongo.find_one.return_value = None
|
|
assert mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None
|
|
mongo.update_one.assert_not_called()
|
|
# Likely misconfig (other MONGODB_DATABASE_NAME or schedule_name mismatch): not silent.
|
|
mongo_activities.warning.assert_called_once()
|
|
assert 'm1-simple_metrics-60m' in mongo_activities.warning.call_args.args[0]
|
|
|
|
|
|
def test_adaptive_state_without_frequency_alert_is_noop(mongo_activities, mongo):
|
|
mongo.find_one.return_value = _pipeline()
|
|
assert mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None
|
|
mongo.update_one.assert_not_called()
|
|
mongo_activities.warning.assert_not_called()
|
|
|
|
|
|
def test_adaptive_state_breach_escalates_and_bumps_updated_at(mongo_activities, mongo):
|
|
mongo.find_one.return_value = _pipeline(frequency_alert='5m')
|
|
|
|
result = mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT)
|
|
|
|
assert result == {'escalated': True, 'healthy_streak': 0}
|
|
update = _set_of(mongo)
|
|
assert update['adaptive_state'] == {'escalated': True, 'healthy_streak': 0}
|
|
# BSON datetime, never a string: the orchestrator calls .replace(tzinfo=...) on it.
|
|
assert isinstance(update['updated_at'], datetime)
|
|
assert update['updated_at'].tzinfo is not None
|
|
|
|
|
|
def test_adaptive_state_healthy_run_increments_streak_without_updated_at(mongo_activities, mongo):
|
|
# 60 min window / 5m = N 12, so one healthy run only moves the streak.
|
|
mongo.find_one.return_value = _pipeline(
|
|
frequency_alert='5m', adaptive_state={'escalated': True, 'healthy_streak': 0}
|
|
)
|
|
|
|
result = mongo_activities.update_simple_metrics_adaptive_state(_HEALTHY_INPUT)
|
|
|
|
assert result == {'escalated': True, 'healthy_streak': 1}
|
|
assert _set_of(mongo) == {'adaptive_state': {'escalated': True, 'healthy_streak': 1}}
|
|
|
|
|
|
def test_adaptive_state_recovers_after_window_renewed(mongo_activities, mongo):
|
|
# 60 min window / 30m = N 2: second consecutive healthy run recovers.
|
|
mongo.find_one.return_value = _pipeline(
|
|
frequency_alert='30m', adaptive_state={'escalated': True, 'healthy_streak': 1}
|
|
)
|
|
|
|
result = mongo_activities.update_simple_metrics_adaptive_state(_HEALTHY_INPUT)
|
|
|
|
assert result == {'escalated': False, 'healthy_streak': 0}
|
|
update = _set_of(mongo)
|
|
assert update['adaptive_state'] == {'escalated': False, 'healthy_streak': 0}
|
|
assert isinstance(update['updated_at'], datetime)
|
|
|
|
|
|
def test_adaptive_state_empty_window_counts_as_healthy_run(mongo_activities, mongo):
|
|
# Spec decision 8: an empty/under-sampled window never escalates but counts toward recovery.
|
|
mongo.find_one.return_value = _pipeline(
|
|
frequency_alert='5m', adaptive_state={'escalated': True, 'healthy_streak': 3}
|
|
)
|
|
|
|
result = mongo_activities.update_simple_metrics_adaptive_state(_EMPTY_INPUT)
|
|
|
|
assert result == {'escalated': True, 'healthy_streak': 4}
|
|
|
|
|
|
def test_adaptive_state_empty_window_never_escalates(mongo_activities, mongo):
|
|
mongo.find_one.return_value = _pipeline(frequency_alert='5m')
|
|
|
|
result = mongo_activities.update_simple_metrics_adaptive_state(_EMPTY_INPUT)
|
|
|
|
assert result == {'escalated': False, 'healthy_streak': 0}
|
|
mongo.update_one.assert_not_called()
|
|
|
|
|
|
def test_adaptive_state_unchanged_does_not_write(mongo_activities, mongo):
|
|
mongo.find_one.return_value = _pipeline(
|
|
frequency_alert='5m', adaptive_state={'escalated': False, 'healthy_streak': 0}
|
|
)
|
|
|
|
result = mongo_activities.update_simple_metrics_adaptive_state(_HEALTHY_INPUT)
|
|
|
|
assert result == {'escalated': False, 'healthy_streak': 0}
|
|
mongo.update_one.assert_not_called()
|
|
|
|
|
|
@mark.parametrize('frequency_alert', ['5min', '0m', 5])
|
|
def test_adaptive_state_invalid_frequency_alert_is_skipped(
|
|
mongo_activities, mongo, frequency_alert
|
|
):
|
|
mongo.find_one.return_value = _pipeline(frequency_alert=frequency_alert)
|
|
|
|
assert mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None
|
|
mongo.update_one.assert_not_called()
|
|
mongo_activities.warning.assert_called_once()
|
|
|
|
|
|
@mark.parametrize('interval_minutes', ['60', None])
|
|
def test_adaptive_state_invalid_interval_minutes_is_skipped(
|
|
mongo_activities, mongo, interval_minutes
|
|
):
|
|
# Untyped Mongo config: "60" * 60 is a string, None * 60 a TypeError.
|
|
mongo.find_one.return_value = _pipeline(frequency_alert='5m')
|
|
|
|
result = mongo_activities.update_simple_metrics_adaptive_state(
|
|
{**_BREACH_INPUT, 'interval_minutes': interval_minutes}
|
|
)
|
|
|
|
assert result is None
|
|
mongo.update_one.assert_not_called()
|
|
mongo_activities.warning.assert_called_once()
|
|
|
|
|
|
@mark.parametrize('adaptive_state', ['garbage', ['escalated']])
|
|
def test_adaptive_state_malformed_state_is_not_escalated(mongo_activities, mongo, adaptive_state):
|
|
mongo.find_one.return_value = _pipeline(frequency_alert='5m', adaptive_state=adaptive_state)
|
|
|
|
result = mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT)
|
|
|
|
assert result == {'escalated': True, 'healthy_streak': 0}
|
|
mongo.update_one.assert_called_once()
|
|
|
|
|
|
@mark.parametrize('failing_call', ['find_one', 'update_one'])
|
|
def test_adaptive_state_mongo_error_propagates(mongo_activities, mongo, failing_call):
|
|
mongo.find_one.return_value = _pipeline(frequency_alert='5m')
|
|
getattr(mongo, failing_call).side_effect = RuntimeError('mongo down')
|
|
|
|
with raises(RuntimeError, match='mongo down'):
|
|
mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT)
|