fix: raise on 404 in ArgoClient.get_workflow_templates (#3239)

metaflow/plugins/argo/argo_client.py

@@ -109,7 +109,7 @@ def get_workflow_templates(self, page_size=100):
                 error_body = json.loads(e.body) if e.body else {}
                 error_message = error_body.get("message", e.reason)
                 if e.status == 404:
-                    return None
+                    raise ArgoClientException(error_message)
                 elif e.status == 410 and error_body.get("reason") == "Expired":
                     new_token = error_body.get("metadata", {}).get("continue")
                     if new_token:

test/unit/test_argo_client.py

@@ -0,0 +1,44 @@
+import pytest
+from kubernetes.client.rest import ApiException
+
+from metaflow.plugins.argo.argo_client import ArgoClient, ArgoClientException
+
+
+@pytest.fixture
+def argo_client(mocker):
+    mock_custom_objects_api = mocker.Mock()
+    mock_k8s = mocker.Mock()
+    mock_k8s.CustomObjectsApi.return_value = mock_custom_objects_api
+    mock_k8s.rest.ApiException = ApiException
+
+    mock_kubernetes_client = mocker.Mock()
+    mock_kubernetes_client.get.return_value = mock_k8s
+
+    mocker.patch(
+        "metaflow.plugins.argo.argo_client.KubernetesClient",
+        return_value=mock_kubernetes_client,
+    )
+
+    client = ArgoClient(namespace="test-ns")
+    return client, mock_custom_objects_api
+
+
+def test_get_workflow_templates_404_raises(argo_client):
+    client, mock_api = argo_client
+    api_error = ApiException(status=404, reason="Not Found")
+    api_error.body = '{"message": "workflowtemplates.argoproj.io not found"}'
+    mock_api.list_namespaced_custom_object.side_effect = api_error
+
+    with pytest.raises(ArgoClientException):
+        list(client.get_workflow_templates())
+
+
+def test_get_workflow_templates_empty_200_does_not_raise(argo_client):
+    client, mock_api = argo_client
+    mock_api.list_namespaced_custom_object.return_value = {
+        "items": [],
+        "metadata": {},
+    }
+
+    result = list(client.get_workflow_templates())
+    assert result == []