Compare commits
4 Commits
6334be2edf
...
0.6.8
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
da27e2ba91 | ||
| 3f42cdea74 | |||
|
|
d963cd3e8a | ||
|
|
0d3ff73557 |
@@ -242,7 +242,12 @@ class MongoDB(SientiaMonitoring):
|
||||
for pipeline in updated_pipelines
|
||||
if pipeline['success']
|
||||
]
|
||||
data_filter = {'$or': argument} if argument else {}
|
||||
if not argument:
|
||||
# Every update failed: stamp nothing so they are retried next tick (same guard as
|
||||
# create_pipelines_timestamps). An empty filter would match every tracked schedule.
|
||||
self.info('No successful pipeline updates to timestamp', metadata=metadata)
|
||||
return
|
||||
data_filter = {'$or': argument}
|
||||
|
||||
try:
|
||||
self.mongo_db_repository.update_many(
|
||||
@@ -343,7 +348,13 @@ class MongoDB(SientiaMonitoring):
|
||||
for pipeline in deleted_pipelines
|
||||
if pipeline['success']
|
||||
]
|
||||
data_filter = {'$or': argument} if argument else {}
|
||||
if not argument:
|
||||
# Every delete failed: remove nothing (same guard as create_pipelines_timestamps).
|
||||
# An empty filter would wipe orchestrated_schedules and make the next tick delete
|
||||
# and recreate every orchestrated schedule.
|
||||
self.info('No successful pipeline deletions to untrack', metadata=metadata)
|
||||
return
|
||||
data_filter = {'$or': argument}
|
||||
|
||||
try:
|
||||
self.mongo_db_repository.delete_many('orchestrated_schedules', data_filter, metadata)
|
||||
|
||||
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "orchestrator"
|
||||
version = "0.6.7"
|
||||
version = "0.6.8"
|
||||
description = "Sientia DataOps Orchestrator - ML Model Orchestration System"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.11"
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
Reference in New Issue
Block a user