added unit tests for validate_subscriptions

gso/schedules/validate_subscriptions.py

@@ -13,7 +13,12 @@ logger = structlog.get_logger(__name__)
 @celery.task
 @scheduler(name="Subscriptions Validator", minute="10", hour="0")
 def validate_subscriptions() -> None:
-    for subscription in get_insync_subscriptions():
+    subscriptions = get_insync_subscriptions()
+    if not subscriptions:
+        logger.info("No subscriptions to validate")
+        return
+
+    for subscription in subscriptions:
         validation_workflow = None
 
         for workflow in subscription.product.workflows:

test/conftest.py

@@ -142,15 +142,14 @@ def configuration_data() -> dict:
 
 @pytest.fixture(scope="session", autouse=True)
 def data_config_filename(configuration_data) -> str:
-    file_name = os.path.join(tempfile.gettempdir(), os.urandom(24).hex())
-    open(file_name, "x").close()
-    with open(file_name, "wb") as f:
-        f.write(json.dumps(configuration_data).encode("utf-8"))
-        f.flush()
-
+    """Create a temporary file with configuration data and set an environment variable to its path."""
+    with tempfile.NamedTemporaryFile(mode="w+", delete=False) as f:
+        json.dump(configuration_data, f, ensure_ascii=False)
         os.environ["OSS_PARAMS_FILENAME"] = f.name
 
-        yield f.name
+    yield f.name
+    del os.environ["OSS_PARAMS_FILENAME"]
+    os.remove(f.name)
 
 
 @pytest.fixture(scope="session")

test/schedules/test_scheduling.py

-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 import pytest
+from orchestrator.targets import Target
 
 from gso.schedules.scheduling import scheduler
 
 
+@pytest.fixture(scope="module")
+def validate_subscriptions():
+    from gso.schedules.validate_subscriptions import validate_subscriptions as vs
+
+    return vs
+
+
+@pytest.fixture
+def mock_get_insync_subscriptions():
+    with patch("gso.schedules.validate_subscriptions.get_insync_subscriptions") as mock:
+        yield mock
+
+
+@pytest.fixture
+def mock_get_execution_context():
+    with patch("gso.schedules.validate_subscriptions.get_execution_context") as mock:
+        mock.return_value = {"validate": MagicMock()}
+        yield mock
+
+
+@pytest.fixture
+def mock_logger():
+    with patch("gso.schedules.validate_subscriptions.logger") as mock:
+        yield mock
+
+
 @pytest.fixture
 def mock_celery():
     with patch("gso.schedules.scheduling.current_app") as mock_app:
@@ -35,3 +62,46 @@ def test_scheduled_task_still_works():
 
     result = mock_task()
     assert result == "task result"
+
+
+def test_no_subscriptions(mock_get_insync_subscriptions, mock_logger, validate_subscriptions):
+    mock_get_insync_subscriptions.return_value = []
+    validate_subscriptions()
+    mock_logger.info.assert_called_once_with("No subscriptions to validate")
+
+
+def test_subscriptions_without_system_target_workflow(
+    mock_get_insync_subscriptions, mock_logger, validate_subscriptions
+):
+    mock_get_insync_subscriptions.return_value = [MagicMock(product=MagicMock(workflows=[]))]
+    validate_subscriptions()
+    mock_logger.warning.assert_called_once()
+
+
+def test_subscription_status_not_usable(
+    mock_get_insync_subscriptions, mock_get_execution_context, validate_subscriptions
+):
+    subscription_mock = MagicMock()
+    subscription_mock.product.workflows = [MagicMock(target=Target.SYSTEM, name="workflow_name")]
+    subscription_mock.status = "Not Usable Status"
+
+    mock_get_insync_subscriptions.return_value = [subscription_mock]
+    validate_subscriptions()
+
+    validate_func = mock_get_execution_context()["validate"]
+    validate_func.assert_not_called()
+
+
+def test_valid_subscriptions_for_validation(
+    mock_get_insync_subscriptions, mock_get_execution_context, validate_subscriptions
+):
+    subscription_mock = MagicMock()
+    mocked_workflow = MagicMock(target=Target.SYSTEM, name="workflow_name")
+    subscription_mock.product.workflows = [mocked_workflow]
+    subscription_mock.status = "active"
+    mock_get_insync_subscriptions.return_value = [subscription_mock]
+    validate_subscriptions()
+    validate_func = mock_get_execution_context()["validate"]
+    validate_func.assert_called_once_with(
+        mocked_workflow.name, json=[{"subscription_id": str(subscription_mock.subscription_id)}]
+    )