"""Coverage for animetix.tasks_views — Cloud Tasks / Workflow / Eventarc endpoints. These are plain ``csrf_exempt`` Django views (not DRF). They are exercised through the Django test ``Client`` via ``reverse()``. Every external collaborator is mocked: * ``run_task_view`` -> registered task funcs (via tasks_registry), cache * ``poll_workflow_view`` -> GCPWorkflowsClient, cache * ``eventarc_gcs_upload_view`` -> enqueue_task, cache Production-only OIDC branches are reached with ``override_settings`` flipping ``IS_PRODUCTION`` and the relevant audience URLs, while ``google.oauth2.id_token.verify_oauth2_token`` is patched to accept/reject. """ import json from unittest.mock import MagicMock, patch import pytest from animetix import tasks_views as tv from django.core.cache import cache from django.test import Client, RequestFactory, override_settings from django.urls import reverse @pytest.fixture def client(): return Client() @pytest.fixture def rf(): return RequestFactory() def _post_run_task(rf, payload=None, raw=None, **extra): """Drive ``run_task_view`` directly. The project routes ``/api/tasks/run/`` to a RedirectView *before* the real view in ``urls.py``, so dispatching through the URL resolver yields a 302. We therefore call the view function directly with a RequestFactory request. """ body = raw if raw is not None else json.dumps(payload or {}) request = rf.post( "/api/tasks/run/", data=body, content_type="application/json", **extra ) return tv.run_task_view(request) @pytest.fixture(autouse=True) def _clear_cache(): cache.clear() yield cache.clear() # --------------------------------------------------------------------------- # # run_task_view # --------------------------------------------------------------------------- # def _body(response): return json.loads(response.content) @pytest.mark.django_db def test_run_task_method_not_allowed(rf): response = tv.run_task_view(rf.get("/api/tasks/run/")) assert response.status_code == 405 assert _body(response)["error"] == "Method not allowed" @pytest.mark.django_db def test_run_task_invalid_json(rf): response = _post_run_task(rf, raw="not-json") assert response.status_code == 400 assert _body(response)["error"] == "Invalid JSON payload" @pytest.mark.django_db def test_run_task_missing_fields(rf): response = _post_run_task(rf, {"task_id": "abc"}) assert response.status_code == 400 assert "required" in _body(response)["error"] @pytest.mark.django_db def test_run_task_unregistered(rf): response = _post_run_task(rf, {"task_id": "id1", "task_name": "does_not_exist"}) assert response.status_code == 400 assert "not registered" in _body(response)["error"] # Failure result is persisted in the cache for later polling. cached = cache.get("task_result:id1") assert cached["state"] == "FAILURE" assert "not registered" in cached["result"]["error"] @pytest.mark.django_db def test_run_task_success(rf): task = MagicMock(return_value={"ok": 42}) with patch.object(tv, "get_registered_task", return_value=task): response = _post_run_task( rf, { "task_id": "id2", "task_name": "my_task", "args": [1, 2], "kwargs": {"x": "y"}, }, ) assert response.status_code == 200 assert _body(response) == {"status": "success", "task_id": "id2"} task.assert_called_once_with(1, 2, x="y") cached = cache.get("task_result:id2") assert cached["state"] == "SUCCESS" assert cached["result"] == {"ok": 42} @pytest.mark.django_db def test_run_task_execution_error(rf): task = MagicMock(side_effect=RuntimeError("boom")) with patch.object(tv, "get_registered_task", return_value=task): response = _post_run_task(rf, {"task_id": "id3", "task_name": "my_task"}) # 500 so Cloud Tasks retries. assert response.status_code == 500 assert _body(response)["error"] == "boom" cached = cache.get("task_result:id3") assert cached["state"] == "FAILURE" assert cached["result"]["error"] == "boom" @pytest.mark.django_db @override_settings(IS_PRODUCTION=True, GCP_TASKS_WORKER_URL="https://worker") def test_run_task_prod_missing_auth(rf): response = _post_run_task(rf, {"task_id": "id", "task_name": "t"}) assert response.status_code == 401 @pytest.mark.django_db @override_settings(IS_PRODUCTION=True, GCP_TASKS_WORKER_URL="https://worker") def test_run_task_prod_bad_token(rf): with patch( "google.oauth2.id_token.verify_oauth2_token", side_effect=ValueError("bad token"), ): response = _post_run_task( rf, {"task_id": "id", "task_name": "t"}, HTTP_AUTHORIZATION="Bearer faketoken", ) assert response.status_code == 403 assert _body(response)["error"] == "Invalid OIDC token" @pytest.mark.django_db @override_settings(IS_PRODUCTION=True, GCP_TASKS_WORKER_URL="https://worker") def test_run_task_prod_valid_token(rf): task = MagicMock(return_value={"done": True}) with ( patch( "google.oauth2.id_token.verify_oauth2_token", return_value={"sub": "svc"} ), patch.object(tv, "get_registered_task", return_value=task), ): response = _post_run_task( rf, {"task_id": "p1", "task_name": "t"}, HTTP_AUTHORIZATION="Bearer goodtoken", ) assert response.status_code == 200 assert cache.get("task_result:p1")["state"] == "SUCCESS" # --------------------------------------------------------------------------- # # poll_workflow_view # --------------------------------------------------------------------------- # @pytest.mark.django_db def test_poll_workflow_method_not_allowed(client): response = client.get(reverse("poll-workflow")) assert response.status_code == 405 @pytest.mark.django_db def test_poll_workflow_invalid_json(client): response = client.post( reverse("poll-workflow"), data="{bad", content_type="application/json" ) assert response.status_code == 400 assert response.json()["error"] == "Invalid payload" @pytest.mark.django_db def test_poll_workflow_missing_fields(client): response = client.post( reverse("poll-workflow"), data=json.dumps({"task_id": "t"}), content_type="application/json", ) assert response.status_code == 400 assert "Missing" in response.json()["error"] @pytest.mark.django_db def test_poll_workflow_active(client): wf = MagicMock() wf.get_execution_status.return_value = {"state": "ACTIVE"} with patch.object(tv, "GCPWorkflowsClient", return_value=wf): response = client.post( reverse("poll-workflow"), data=json.dumps({"execution_name": "exec/1", "task_id": "t1"}), content_type="application/json", ) # 503 makes Cloud Tasks retry. assert response.status_code == 503 wf.get_execution_status.assert_called_once_with("exec/1") @pytest.mark.django_db def test_poll_workflow_succeeded(client): wf = MagicMock() wf.get_execution_status.return_value = { "state": "SUCCEEDED", "result": {"translated_text": "hola", "audio_url": "http://a"}, } with patch.object(tv, "GCPWorkflowsClient", return_value=wf): response = client.post( reverse("poll-workflow"), data=json.dumps({"execution_name": "exec/2", "task_id": "t2"}), content_type="application/json", ) assert response.status_code == 200 assert response.json()["status"] == "completed" cached = cache.get("task_result:t2") assert cached["status"] == "success" assert cached["result"]["translated_text"] == "hola" assert cached["result"]["audio_url"] == "http://a" @pytest.mark.django_db def test_poll_workflow_failed(client): wf = MagicMock() wf.get_execution_status.return_value = {"state": "FAILED", "error": "nope"} with patch.object(tv, "GCPWorkflowsClient", return_value=wf): response = client.post( reverse("poll-workflow"), data=json.dumps({"execution_name": "exec/3", "task_id": "t3"}), content_type="application/json", ) assert response.status_code == 200 assert response.json()["status"] == "failed" cached = cache.get("task_result:t3") assert cached["status"] == "failed" assert cached["error"] == "nope" @pytest.mark.django_db @override_settings(IS_PRODUCTION=True, GCP_WORKFLOW_POLL_URL="https://poll") def test_poll_workflow_prod_missing_auth(client): response = client.post( reverse("poll-workflow"), data=json.dumps({"execution_name": "e", "task_id": "t"}), content_type="application/json", ) assert response.status_code == 401 assert response.json()["error"] == "Unauthorized" @pytest.mark.django_db @override_settings(IS_PRODUCTION=True, GCP_WORKFLOW_POLL_URL="https://poll") def test_poll_workflow_prod_bad_token(client): with patch( "animetix.tasks_views.id_token.verify_oauth2_token", side_effect=ValueError("bad"), ): response = client.post( reverse("poll-workflow"), data=json.dumps({"execution_name": "e", "task_id": "t"}), content_type="application/json", HTTP_AUTHORIZATION="Bearer x", ) assert response.status_code == 403 assert response.json()["error"] == "Forbidden" @pytest.mark.django_db @override_settings(IS_PRODUCTION=True, GCP_WORKFLOW_POLL_URL="https://poll") def test_poll_workflow_prod_valid_token(client): wf = MagicMock() wf.get_execution_status.return_value = {"state": "ACTIVE"} with ( patch("animetix.tasks_views.id_token.verify_oauth2_token", return_value={}), patch.object(tv, "GCPWorkflowsClient", return_value=wf), ): response = client.post( reverse("poll-workflow"), data=json.dumps({"execution_name": "e", "task_id": "t"}), content_type="application/json", HTTP_AUTHORIZATION="Bearer good", ) assert response.status_code == 503 # --------------------------------------------------------------------------- # # eventarc_gcs_upload_view # --------------------------------------------------------------------------- # @pytest.mark.django_db def test_eventarc_method_not_allowed(client): response = client.get(reverse("eventarc-gcs-upload")) assert response.status_code == 405 @pytest.mark.django_db def test_eventarc_invalid_json(client): response = client.post( reverse("eventarc-gcs-upload"), data="{bad", content_type="application/json" ) assert response.status_code == 400 assert "Invalid JSON body" in response.json()["error"] @pytest.mark.django_db def test_eventarc_missing_fields(client): response = client.post( reverse("eventarc-gcs-upload"), data=json.dumps({"bucket": "b"}), content_type="application/json", ) assert response.status_code == 400 assert "Missing bucket or name" in response.json()["error"] @pytest.mark.django_db def test_eventarc_non_manga_no_enqueue(client): with patch.object(tv, "enqueue_task") as enqueue: response = client.post( reverse("eventarc-gcs-upload"), data=json.dumps({"bucket": "b", "name": "docs/readme.txt"}), content_type="application/json", ) assert response.status_code == 200 body = response.json() assert body["status"] == "event processed" assert body["bucket"] == "b" enqueue.assert_not_called() @pytest.mark.django_db def test_eventarc_manga_enqueues(client): with patch.object(tv, "enqueue_task") as enqueue: response = client.post( reverse("eventarc-gcs-upload"), data=json.dumps({"bucket": "b", "name": "raw-manga/page1.PNG"}), content_type="application/json", ) assert response.status_code == 200 assert response.json()["status"] == "event processed" enqueue.assert_called_once_with( "process_gcs_upload_task", bucket="b", name="raw-manga/page1.PNG" ) @pytest.mark.django_db def test_eventarc_manga_enqueue_failure(client): with patch.object(tv, "enqueue_task", side_effect=RuntimeError("queue down")): response = client.post( reverse("eventarc-gcs-upload"), data=json.dumps({"bucket": "b", "name": "manga/p.jpeg"}), content_type="application/json", ) assert response.status_code == 500 assert "Failed to enqueue task" in response.json()["error"] @pytest.mark.django_db @override_settings(IS_PRODUCTION=True, EVENTARC_RECEIVER_URL="https://recv") def test_eventarc_prod_missing_auth(client): response = client.post( reverse("eventarc-gcs-upload"), data=json.dumps({"bucket": "b", "name": "n"}), content_type="application/json", ) assert response.status_code == 401 @pytest.mark.django_db @override_settings(IS_PRODUCTION=True, EVENTARC_RECEIVER_URL="https://recv") def test_eventarc_prod_bad_token(client): with patch( "animetix.tasks_views.id_token.verify_oauth2_token", side_effect=ValueError("bad"), ): response = client.post( reverse("eventarc-gcs-upload"), data=json.dumps({"bucket": "b", "name": "n"}), content_type="application/json", HTTP_AUTHORIZATION="Bearer x", ) assert response.status_code == 403 assert response.json()["error"] == "Invalid OIDC token" @pytest.mark.django_db @override_settings(IS_PRODUCTION=True, EVENTARC_RECEIVER_URL="https://recv") def test_eventarc_prod_valid_token(client): with ( patch("animetix.tasks_views.id_token.verify_oauth2_token", return_value={}), patch.object(tv, "enqueue_task") as enqueue, ): response = client.post( reverse("eventarc-gcs-upload"), data=json.dumps({"bucket": "b", "name": "manga/p.webp"}), content_type="application/json", HTTP_AUTHORIZATION="Bearer good", ) assert response.status_code == 200 enqueue.assert_called_once()