UN-3636 [DEV] Cover six worker-free critical paths with integration tests

backend/api_v2/tests/test_deployment_auth.py

@@ -0,0 +1,109 @@
+"""Critical path ``api-deployment-auth``: the public deployment endpoint
+rejects unauthenticated callers before any work is dispatched.
+
+``POST /deployment/api/<org>/<api_name>/`` is the only unauthenticated surface
+in the product — it is guarded by the ``@DeploymentHelper.validate_api_key``
+decorator rather than DRF permission classes, so a regression here silently
+opens the endpoint. Every rejection must land before ``execute_workflow``, i.e.
+before a ``WorkflowExecution`` row is written or a Celery task is queued;
+the mock below asserts exactly that. Needs a live DB (integration tier).
+"""
+
+from __future__ import annotations
+
+import uuid
+from unittest.mock import patch
+
+import pytest
+from account_v2.models import Organization
+from django.core.files.uploadedfile import SimpleUploadedFile
+from django.test import TestCase
+from rest_framework.test import APIRequestFactory
+from utils.user_context import UserContext
+from workflow_manager.workflow_v2.models.workflow import Workflow
+
+from api_v2.api_deployment_views import DeploymentExecution
+from api_v2.models import APIDeployment, APIKey
+
+ORG_ID = "org-auth"
+
+
+class APIDeploymentAuthTest(TestCase):
+    def setUp(self) -> None:
+        self.org = Organization.objects.create(
+            name=ORG_ID, display_name="Org Auth", organization_id=ORG_ID
+        )
+        UserContext.set_organization_identifier(ORG_ID)
+        workflow = Workflow.objects.create(workflow_name="wf-auth", is_active=True)
+
+        self.api = APIDeployment.objects.create(api_name="live-api", workflow=workflow)
+        self.key = APIKey.objects.create(api=self.api)
+        self.inactive_key = APIKey.objects.create(api=self.api, is_active=False)
+
+        self.inactive_api = APIDeployment.objects.create(
+            api_name="dead-api", workflow=workflow, is_active=False
+        )
+        self.other_api = APIDeployment.objects.create(
+            api_name="other-api", workflow=workflow
+        )
+        self.other_key = APIKey.objects.create(api=self.other_api)
+
+        self.view = DeploymentExecution.as_view()
+        self.factory = APIRequestFactory()
+
+    def _post(self, api_name: str, auth: str | None, org: str = ORG_ID):
+        headers = {"HTTP_AUTHORIZATION": auth} if auth is not None else {}
+        payload = {"files": SimpleUploadedFile("doc.txt", b"hello")}
+        request = self.factory.post(
+            f"/deployment/api/{org}/{api_name}/", payload, format="multipart", **headers
+        )
+        return self.view(request, org_name=org, api_name=api_name)
+
+    @pytest.mark.critical_path("api-deployment-auth")
+    @patch("api_v2.api_deployment_views.DeploymentHelper.execute_workflow")
+    def test_bad_credentials_rejected_before_dispatch(self, execute_workflow) -> None:
+        cases = [
+            ("missing header", "live-api", None, ORG_ID, 403),
+            ("no bearer prefix", "live-api", f"Token {self.key.api_key}", ORG_ID, 403),
+            ("empty bearer", "live-api", "Bearer ", ORG_ID, 403),
+            ("unknown key", "live-api", f"Bearer {uuid.uuid4()}", ORG_ID, 401),
+            ("not a uuid", "live-api", "Bearer not-a-uuid", ORG_ID, 401),
+            (
+                "inactive key",
+                "live-api",
+                f"Bearer {self.inactive_key.api_key}",
+                ORG_ID,
+                401,
+            ),
+            (
+                "key of another api",
+                "live-api",
+                f"Bearer {self.other_key.api_key}",
+                ORG_ID,
+                401,
+            ),
+            ("unknown api", "ghost-api", f"Bearer {self.key.api_key}", ORG_ID, 404),
+            ("inactive api", "dead-api", f"Bearer {self.key.api_key}", ORG_ID, 404),
+            ("wrong org", "live-api", f"Bearer {self.key.api_key}", "no-such-org", 404),
+        ]
+        for label, api_name, auth, org, expected in cases:
+            with self.subTest(label):
+                response = self._post(api_name, auth, org)
+                assert response.status_code == expected, response.data
+
+        execute_workflow.assert_not_called()
+
+    @pytest.mark.critical_path("api-deployment-auth")
+    @patch("api_v2.api_deployment_views.APIDeploymentRateLimiter.check_and_acquire")
+    @patch("api_v2.api_deployment_views.DeploymentHelper.execute_workflow")
+    def test_valid_key_reaches_execution(self, execute_workflow, rate_limit) -> None:
+        """Guard the inverse of the rejection cases: a guard that rejected
+        everything would pass them all.
+        """
+        rate_limit.return_value = (True, {})
+        execute_workflow.return_value = {"execution_status": "COMPLETED"}
+
+        response = self._post("live-api", f"Bearer {self.key.api_key}")
+
+        assert response.status_code == 200, response.data
+        execute_workflow.assert_called_once()

backend/api_v2/tests/test_deployment_provision.py

@@ -0,0 +1,88 @@
+"""Critical path ``api-deployment-provision``: deploying a workflow as an API
+mints a usable key and a resolvable endpoint.
+
+Provisioning is the only place an API key is created for a deployment, and the
+endpoint string it derives is what callers POST to. Both are returned exactly
+once — at create time — so a regression here is unrecoverable for the caller.
+Fully synchronous. Needs a live DB (integration tier).
+"""
+
+from __future__ import annotations
+
+import secrets
+
+import pytest
+from account_v2.models import Organization, User
+from django.test import TestCase
+from rest_framework import status
+from rest_framework.test import APIRequestFactory, force_authenticate
+from tenant_account_v2.models import OrganizationMember
+from utils.user_context import UserContext
+from workflow_manager.endpoint_v2.models import WorkflowEndpoint
+from workflow_manager.workflow_v2.models.workflow import Workflow
+
+from api_v2.api_deployment_views import APIDeploymentViewSet
+from api_v2.api_key_views import APIKeyViewSet
+from api_v2.models import APIDeployment, APIKey
+
+ORG_ID = "org-provision"
+
+
+class APIDeploymentProvisionTest(TestCase):
+    def setUp(self) -> None:
+        self.org = Organization.objects.create(
+            name=ORG_ID, display_name="Org Provision", organization_id=ORG_ID
+        )
+        UserContext.set_organization_identifier(ORG_ID)
+        self.user = User.objects.create_user(
+            username="deployer@example.com",
+            email="deployer@example.com",
+            password=secrets.token_urlsafe(),
+        )
+        OrganizationMember.objects.create(
+            organization=self.org, user=self.user, role="user"
+        )
+        self.workflow = Workflow.objects.create(workflow_name="wf-deploy", is_active=True)
+        for endpoint_type in WorkflowEndpoint.EndpointType:
+            WorkflowEndpoint.objects.create(
+                workflow=self.workflow,
+                endpoint_type=endpoint_type,
+                connection_type=WorkflowEndpoint.ConnectionType.API,
+            )
+        self.factory = APIRequestFactory()
+
+    @pytest.mark.critical_path("api-deployment-provision")
+    def test_deploy_mints_key_and_endpoint(self) -> None:
+        create = APIDeploymentViewSet.as_view({"post": "create"})
+        request = self.factory.post(
+            "/api/v1/api/deployment/",
+            {
+                "workflow": str(self.workflow.id),
+                "display_name": "my api",
+                "api_name": "my-api",
+                "description": "provisioned in a test",
+            },
+            format="json",
+        )
+        force_authenticate(request, user=self.user)
+        response = create(request)
+
+        assert response.status_code == status.HTTP_201_CREATED, response.data
+        assert response.data["api_endpoint"] == f"deployment/api/{ORG_ID}/my-api/"
+
+        api = APIDeployment.objects.get(api_name="my-api")
+        assert api.is_active
+        assert api.organization_id == self.org.id
+
+        # The key is returned once, at create time, and never again.
+        key = APIKey.objects.get(api=api)
+        assert str(key.api_key) == response.data["api_key"]
+        assert key.is_active
+
+        list_keys = APIKeyViewSet.as_view({"get": "api_keys"})
+        list_request = self.factory.get(f"/api/v1/api/keys/api/{api.id}/")
+        force_authenticate(list_request, user=self.user)
+        listed = list_keys(list_request, api_id=str(api.id))
+
+        assert listed.status_code == status.HTTP_200_OK, listed.data
+        assert [k["id"] for k in listed.data] == [str(key.id)]

backend/connector_v2/tests/test_connector_register.py

@@ -0,0 +1,123 @@
+"""Critical path ``connector-register-test``: credentials are validated against
+the live external system, and a registered connector persists them encrypted.
+
+``POST /test_connectors/`` is the only place the platform proves it can actually
+reach a customer's storage before a workflow depends on it; a version that
+always answers "valid" would surface as a run-time failure deep inside an
+execution. Runs against the rig's MinIO. Needs a live DB (integration tier).
+"""
+
+from __future__ import annotations
+
+import os
+import secrets
+
+import pytest
+from account_v2.models import Organization, User
+from django.db import connection
+from django.test import TestCase
+from rest_framework import status
+from rest_framework.test import APIRequestFactory, force_authenticate
+from tenant_account_v2.models import OrganizationMember
+from utils.user_context import UserContext
+
+from connector_v2.models import ConnectorInstance
+
+MINIO_CONNECTOR_ID = "minio|c799f6e3-2b57-434e-aaac-b5daa415da19"
+
+pytestmark = pytest.mark.skipif(
+    not all(
+        os.environ.get(var)
+        for var in (
+            "MINIO_ACCESS_KEY_ID",
+            "MINIO_SECRET_ACCESS_KEY",
+            "MINIO_ENDPOINT_URL",
+        )
+    ),
+    reason="needs a live MinIO (provisioned by the rig for this group)",
+)
+
+
+def _credentials(secret: str | None = None) -> dict:
+    return {
+        "connectorName": "rig-minio",
+        "key": os.environ["MINIO_ACCESS_KEY_ID"],
+        "secret": secret or os.environ["MINIO_SECRET_ACCESS_KEY"],
+        "endpoint_url": os.environ["MINIO_ENDPOINT_URL"],
+        "region_name": "",
+        "path": "/",
+    }
+
+
+class ConnectorRegisterTest(TestCase):
+    def setUp(self) -> None:
+        self.org = Organization.objects.create(
+            name="org-conn", display_name="Org Conn", organization_id="org-conn"
+        )
+        UserContext.set_organization_identifier(self.org.organization_id)
+        self.user = User.objects.create_user(
+            username="connector@example.com",
+            email="connector@example.com",
+            password=secrets.token_urlsafe(),
+        )
+        OrganizationMember.objects.create(
+            organization=self.org, user=self.user, role="user"
+        )
+        self.factory = APIRequestFactory()
+
+    def _test_connector(self, credentials: dict):
+        from connector_processor.views import ConnectorViewSet
+
+        view = ConnectorViewSet.as_view({"post": "test"})
+        request = self.factory.post(
+            "/api/v1/test_connectors/",
+            {"connector_id": MINIO_CONNECTOR_ID, "connector_metadata": credentials},
+            format="json",
+        )
+        force_authenticate(request, user=self.user)
+        return view(request)
+
+    @pytest.mark.critical_path("connector-register-test")
+    def test_good_credentials_validate_bad_ones_do_not(self) -> None:
+        valid = self._test_connector(_credentials())
+        assert valid.status_code == status.HTTP_200_OK, valid.data
+        assert valid.data["is_valid"] is True
+
+        invalid = self._test_connector(_credentials(secret="wrong-secret"))
+        assert invalid.status_code >= status.HTTP_400_BAD_REQUEST, invalid.data
+        assert "is_valid" not in invalid.data
+
+    @pytest.mark.critical_path("connector-register-test")
+    def test_register_persists_credentials_encrypted(self) -> None:
+        from connector_v2.views import ConnectorInstanceViewSet
+
+        view = ConnectorInstanceViewSet.as_view({"post": "create"})
+        request = self.factory.post(
+            "/api/v1/connector/",
+            {
+                "connector_id": MINIO_CONNECTOR_ID,
+                "connector_name": "rig-minio",
+                "connector_metadata": _credentials(),
+            },
+            format="json",
+        )
+        force_authenticate(request, user=self.user)
+        response = view(request)
+
+        assert response.status_code == status.HTTP_201_CREATED, response.data
+        instance = ConnectorInstance.objects.get(id=response.data["id"])
+        assert instance.organization_id == self.org.id
+        # Decrypts back to the input on read...
+        assert instance.connector_metadata["key"] == os.environ["MINIO_ACCESS_KEY_ID"]
+
+        # ...but the raw column must be ciphertext. Read it past the field's
+        # decrypting descriptor, else a regression to plaintext still passes.
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT connector_metadata FROM connector_instance WHERE id = %s",
+                [str(instance.id)],
+            )
+            raw = bytes(cursor.fetchone()[0])
+        secret = os.environ["MINIO_SECRET_ACCESS_KEY"].encode()
+        assert secret not in raw
+        assert os.environ["MINIO_ACCESS_KEY_ID"].encode() not in raw