"""Idempotency-Key behavior on the /api/upload and /api/remote routes.""" import io import json from contextlib import contextmanager from unittest.mock import MagicMock, patch import pytest from flask import Flask @pytest.fixture def app(): return Flask(__name__) @contextmanager def _patch_db(conn): @contextmanager def _yield(): yield conn with patch( "application.api.user.sources.upload.db_session", _yield ), patch( "application.api.user.sources.upload.db_readonly", _yield ): yield def _apply_async_mock(): """Mock for ``ingest.apply_async``; ``task.id`` mirrors the predetermined id.""" def _side_effect(*args, **kwargs): return MagicMock(id=kwargs.get("task_id") or "auto-task-id") m = MagicMock(side_effect=_side_effect) return m class TestUploadIdempotency: def test_no_header_enqueues_normally(self, app, pg_conn): from application.api.user.sources.upload import UploadFile fake_storage = MagicMock() apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.sources.upload.ingest.apply_async", apply_mock, ), app.test_request_context( "/api/upload", method="POST", data={ "user": "alice", "name": "j", "file": (io.BytesIO(b"content"), "doc.txt"), }, content_type="multipart/form-data", ): from flask import request request.decoded_token = {"sub": "alice"} response = UploadFile().post() assert response.status_code == 200 assert apply_mock.call_count == 1 # No key → no predetermined id was passed. assert "task_id" not in apply_mock.call_args.kwargs def test_header_first_post_records_row(self, app, pg_conn): from sqlalchemy import text from application.api.user.sources.upload import UploadFile fake_storage = MagicMock() apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.sources.upload.ingest.apply_async", apply_mock, ), app.test_request_context( "/api/upload", method="POST", data={ "user": "alice", "name": "j", "file": (io.BytesIO(b"content"), "doc.txt"), }, content_type="multipart/form-data", headers={"Idempotency-Key": "up-key-1"}, ): from flask import request request.decoded_token = {"sub": "alice"} response = UploadFile().post() assert response.status_code == 200 assert apply_mock.call_count == 1 predetermined_id = apply_mock.call_args.kwargs["task_id"] assert response.json["task_id"] == predetermined_id # The dedup row is keyed on the *scoped* form ``"{user}:{key}"`` # so two users sending the same raw header don't collapse. row = pg_conn.execute( text( "SELECT task_id, task_name, status FROM task_dedup " "WHERE idempotency_key = :k" ), {"k": "alice:up-key-1"}, ).fetchone() assert row is not None assert row[0] == predetermined_id assert row[1] == "ingest" assert row[2] == "pending" def test_header_forwards_idempotency_key_to_delay(self, app, pg_conn): """The Celery task body needs the key so ``with_idempotency`` can record terminal status and ``_derive_source_id`` can pick it up. """ from application.api.user.sources.upload import UploadFile fake_storage = MagicMock() apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.sources.upload.ingest.apply_async", apply_mock, ), app.test_request_context( "/api/upload", method="POST", data={ "user": "alice", "name": "j", "file": (io.BytesIO(b"c"), "doc.txt"), }, content_type="multipart/form-data", headers={"Idempotency-Key": "up-fwd"}, ): from flask import request request.decoded_token = {"sub": "alice"} UploadFile().post() # Worker sees the scoped form so its ``with_idempotency`` row # and ``_derive_source_id`` are also user-distinct. assert ( apply_mock.call_args.kwargs["kwargs"]["idempotency_key"] == "alice:up-fwd" ) def test_same_header_second_post_returns_cached(self, app, pg_conn): from application.api.user.sources.upload import UploadFile fake_storage = MagicMock() apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.sources.upload.ingest.apply_async", apply_mock, ): with app.test_request_context( "/api/upload", method="POST", data={ "user": "alice", "name": "j", "file": (io.BytesIO(b"content"), "doc.txt"), }, content_type="multipart/form-data", headers={"Idempotency-Key": "up-rep"}, ): from flask import request request.decoded_token = {"sub": "alice"} first = UploadFile().post() with app.test_request_context( "/api/upload", method="POST", data={ "user": "alice", "name": "j", "file": (io.BytesIO(b"content"), "doc.txt"), }, content_type="multipart/form-data", headers={"Idempotency-Key": "up-rep"}, ): from flask import request request.decoded_token = {"sub": "alice"} second = UploadFile().post() assert first.status_code == 200 assert second.status_code == 200 assert first.json == second.json assert apply_mock.call_count == 1 def test_same_key_different_users_does_not_collide(self, app, pg_conn): """Cross-user collision regression: two users sending the same raw ``Idempotency-Key`` must each get their own dedup row, both requests enqueue, and the responses carry distinct ``task_id``s. (Pre-fix, the second user's request was silently deduplicated against the first user's row.) """ from sqlalchemy import text as sql_text from application.api.user.sources.upload import UploadFile fake_storage = MagicMock() apply_mock = _apply_async_mock() def _fire(user): with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.sources.upload.ingest.apply_async", apply_mock, ), app.test_request_context( "/api/upload", method="POST", data={ "user": user, "name": "j", "file": (io.BytesIO(b"c"), "doc.txt"), }, content_type="multipart/form-data", headers={"Idempotency-Key": "shared-key"}, ): from flask import request request.decoded_token = {"sub": user} return UploadFile().post() first = _fire("alice") second = _fire("bob") assert first.status_code == 200 assert second.status_code == 200 assert first.json["task_id"] != second.json["task_id"] assert apply_mock.call_count == 2 rows = pg_conn.execute( sql_text( "SELECT idempotency_key FROM task_dedup " "WHERE idempotency_key LIKE :pat ORDER BY idempotency_key" ), {"pat": "%:shared-key"}, ).fetchall() assert {r[0] for r in rows} == {"alice:shared-key", "bob:shared-key"} def test_concurrent_same_key_only_one_apply_async(self, app, pg_engine): """Race test (M3): N parallel POSTs with same key → only ONE apply_async. Uses ``pg_engine`` (not ``pg_conn``) so each thread can check out its own DB connection — sharing a single Connection across threads serializes at the driver level and defeats the race. """ from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from application.api.user.sources.upload import UploadFile fake_storage = MagicMock() apply_mock = _apply_async_mock() @contextmanager def _engine_session(): with pg_engine.begin() as conn: yield conn @contextmanager def _engine_readonly(): with pg_engine.connect() as conn: yield conn def fire(idx): # Patches sit OUTSIDE the threads (see below); only the # per-thread Flask request context is set up inside. with app.test_request_context( "/api/upload", method="POST", data={ "user": "alice", "name": "j", "file": (io.BytesIO(b"x"), "doc.txt"), }, content_type="multipart/form-data", headers={"Idempotency-Key": "up-race"}, ): from flask import request request.decoded_token = {"sub": "alice"} return UploadFile().post() # ``unittest.mock.patch`` is not thread-safe — concurrent # ``__enter__`` calls race on saving/restoring the module # attribute and can leave threads pointing at the real # function instead of the mock. Set up patches once, share # across threads. with patch( "application.api.user.sources.upload.db_session", _engine_session, ), patch( "application.api.user.sources.upload.db_readonly", _engine_readonly, ), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.sources.upload.ingest.apply_async", apply_mock, ), ThreadPoolExecutor(max_workers=8) as ex: responses = list(ex.map(fire, range(8))) assert all(r.status_code == 200 for r in responses) # Only one writer wins the claim, so only one apply_async is fired. assert apply_mock.call_count == 1 # All 8 responses share the same task_id (winner's predetermined id). ids = {r.json["task_id"] for r in responses} assert len(ids) == 1 assert "deduplicated" not in ids def test_empty_header_treated_as_absent(self, app, pg_conn): from sqlalchemy import text from application.api.user.sources.upload import UploadFile fake_storage = MagicMock() apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.sources.upload.ingest.apply_async", apply_mock, ), app.test_request_context( "/api/upload", method="POST", data={ "user": "alice", "name": "j", "file": (io.BytesIO(b"content"), "doc.txt"), }, content_type="multipart/form-data", headers={"Idempotency-Key": ""}, ): from flask import request request.decoded_token = {"sub": "alice"} response = UploadFile().post() assert response.status_code == 200 assert apply_mock.call_count == 1 count = pg_conn.execute( text("SELECT count(*) FROM task_dedup") ).scalar() assert count == 0 def test_oversized_header_rejected_with_400(self, app, pg_conn): from application.api.user.sources.upload import UploadFile fake_storage = MagicMock() oversized = "x" * 257 with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.sources.upload.ingest.apply_async", ) as mock_apply, app.test_request_context( "/api/upload", method="POST", data={ "user": "alice", "name": "j", "file": (io.BytesIO(b"content"), "doc.txt"), }, content_type="multipart/form-data", headers={"Idempotency-Key": oversized}, ): from flask import request request.decoded_token = {"sub": "alice"} response = UploadFile().post() assert response.status_code == 400 assert mock_apply.call_count == 0 def test_stale_dedup_row_does_not_block_new_work(self, app, pg_conn): """Regression for the TTL fail-shut bug: a >24h-old dedup row must not silently drop a new upload. Pre-fix, the second POST returned ``task_id="deduplicated"`` and never enqueued. """ from sqlalchemy import text from application.api.user.sources.upload import UploadFile fake_storage = MagicMock() apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.sources.upload.ingest.apply_async", apply_mock, ): with app.test_request_context( "/api/upload", method="POST", data={ "user": "alice", "name": "j", "file": (io.BytesIO(b"content"), "doc.txt"), }, content_type="multipart/form-data", headers={"Idempotency-Key": "stale-up"}, ): from flask import request request.decoded_token = {"sub": "alice"} first = UploadFile().post() assert first.status_code == 200 first_task_id = first.json["task_id"] assert first_task_id != "deduplicated" # Backdate the row past TTL. pg_conn.execute( text( "UPDATE task_dedup SET created_at = " "clock_timestamp() - make_interval(hours => 25) " "WHERE idempotency_key = :k" ), {"k": "alice:stale-up"}, ) with app.test_request_context( "/api/upload", method="POST", data={ "user": "alice", "name": "j", "file": (io.BytesIO(b"content"), "doc.txt"), }, content_type="multipart/form-data", headers={"Idempotency-Key": "stale-up"}, ): from flask import request request.decoded_token = {"sub": "alice"} second = UploadFile().post() assert second.status_code == 200 assert second.json["task_id"] != "deduplicated" assert second.json["task_id"] != first_task_id assert apply_mock.call_count == 2 class TestRemoteIdempotency: def test_no_header_enqueues_normally(self, app, pg_conn): from application.api.user.sources.upload import UploadRemote apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.ingest_remote.apply_async", apply_mock, ), app.test_request_context( "/api/remote", method="POST", data={ "user": "u", "source": "github", "name": "g", "data": json.dumps({"repo_url": "https://github.com/x/y"}), }, content_type="multipart/form-data", ): from flask import request request.decoded_token = {"sub": "u"} response = UploadRemote().post() assert response.status_code == 200 assert apply_mock.call_count == 1 def test_header_first_post_records_row(self, app, pg_conn): from sqlalchemy import text from application.api.user.sources.upload import UploadRemote apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.ingest_remote.apply_async", apply_mock, ), app.test_request_context( "/api/remote", method="POST", data={ "user": "u", "source": "github", "name": "g", "data": json.dumps({"repo_url": "https://github.com/x/y"}), }, content_type="multipart/form-data", headers={"Idempotency-Key": "rem-key-1"}, ): from flask import request request.decoded_token = {"sub": "u"} response = UploadRemote().post() assert response.status_code == 200 assert apply_mock.call_count == 1 predetermined_id = apply_mock.call_args.kwargs["task_id"] # Scoped key: ``"{user}:{key}"``. row = pg_conn.execute( text("SELECT task_id, task_name FROM task_dedup WHERE idempotency_key = :k"), {"k": "u:rem-key-1"}, ).fetchone() assert row is not None assert row[0] == predetermined_id assert row[1] == "ingest_remote" def test_same_header_second_post_returns_cached(self, app, pg_conn): from application.api.user.sources.upload import UploadRemote apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.ingest_remote.apply_async", apply_mock, ): with app.test_request_context( "/api/remote", method="POST", data={ "user": "u", "source": "github", "name": "g", "data": json.dumps({"repo_url": "https://github.com/x/y"}), }, content_type="multipart/form-data", headers={"Idempotency-Key": "rem-rep"}, ): from flask import request request.decoded_token = {"sub": "u"} first = UploadRemote().post() with app.test_request_context( "/api/remote", method="POST", data={ "user": "u", "source": "github", "name": "g", "data": json.dumps({"repo_url": "https://github.com/x/y"}), }, content_type="multipart/form-data", headers={"Idempotency-Key": "rem-rep"}, ): from flask import request request.decoded_token = {"sub": "u"} second = UploadRemote().post() assert first.status_code == 200 assert second.status_code == 200 assert first.json == second.json assert apply_mock.call_count == 1 def test_oversized_header_rejected_with_400(self, app, pg_conn): from application.api.user.sources.upload import UploadRemote oversized = "x" * 257 with _patch_db(pg_conn), patch( "application.api.user.sources.upload.ingest_remote.apply_async", ) as mock_apply, app.test_request_context( "/api/remote", method="POST", data={ "user": "u", "source": "github", "name": "g", "data": json.dumps({"repo_url": "https://github.com/x/y"}), }, content_type="multipart/form-data", headers={"Idempotency-Key": oversized}, ): from flask import request request.decoded_token = {"sub": "u"} response = UploadRemote().post() assert response.status_code == 400 assert mock_apply.call_count == 0 def test_no_header_returns_source_id_matching_worker_kwarg( self, app, pg_conn, ): """Regression: without an ``Idempotency-Key``, the route must still return a ``source_id`` AND pass that same id to the worker as ``source_id`` so SSE envelopes line up with what the frontend already has. Previously the route omitted ``source_id`` entirely on the no-key path and the worker minted its own random uuid, breaking push correlation for the default upload flow. """ from application.api.user.sources.upload import UploadRemote apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.ingest_remote.apply_async", apply_mock, ), app.test_request_context( "/api/remote", method="POST", data={ "user": "u", "source": "github", "name": "g", "data": json.dumps({"repo_url": "https://github.com/x/y"}), }, content_type="multipart/form-data", ): from flask import request request.decoded_token = {"sub": "u"} response = UploadRemote().post() assert response.status_code == 200 assert "source_id" in response.json assert ( apply_mock.call_args.kwargs["kwargs"]["source_id"] == response.json["source_id"] ) def test_no_header_connector_returns_source_id_matching_worker_kwarg( self, app, pg_conn, ): """Same regression as above for the connector branch (``ingest_connector_task``). The connector path took the no-key gap independently of the plain remote path.""" from application.api.user.sources.upload import UploadRemote apply_mock = _apply_async_mock() # Pick any registered connector — the route only branches on # ``ConnectorCreator.get_supported_connectors()``. from application.parser.connectors.connector_creator import ( ConnectorCreator, ) supported = ConnectorCreator.get_supported_connectors() if not supported: pytest.skip("no connectors registered in this build") connector_source = next(iter(supported)) with _patch_db(pg_conn), patch( "application.api.user.sources.upload.ingest_connector_task.apply_async", apply_mock, ), app.test_request_context( "/api/remote", method="POST", data={ "user": "u", "source": connector_source, "name": "g", "data": json.dumps({ "session_token": "tok", "file_ids": ["f1"], }), }, content_type="multipart/form-data", ): from flask import request request.decoded_token = {"sub": "u"} response = UploadRemote().post() assert response.status_code == 200 assert "source_id" in response.json assert ( apply_mock.call_args.kwargs["kwargs"]["source_id"] == response.json["source_id"] ) def _seed_source(pg_conn, user="u", **kw): from application.storage.db.repositories.sources import SourcesRepository return SourcesRepository(pg_conn).create("manage-src", user_id=user, **kw) class TestManageSourceFilesIdempotency: """Same-key dedup contract for ``ManageSourceFiles.post``: a duplicate POST must not enqueue a second ``reingest_source_task``. The worker decorator only deduplicates *post-completion*, so the HTTP handler is the only place that can serialize concurrent in-flight requests. """ def _add_request(self, app, src_id, user, key=None): kwargs = dict( data={ "source_id": str(src_id), "operation": "add", "file": (io.BytesIO(b"content"), "new.txt"), }, content_type="multipart/form-data", ) if key is not None: kwargs["headers"] = {"Idempotency-Key": key} return app.test_request_context( "/api/manage_source_files", method="POST", **kwargs, ) def test_no_header_enqueues_normally_no_claim_row(self, app, pg_conn): from sqlalchemy import text from application.api.user.sources.upload import ManageSourceFiles user = "alice-mgr-noh" src = _seed_source(pg_conn, user=user, file_path="/data") fake_storage = MagicMock() apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.tasks.reingest_source_task.apply_async", apply_mock, ), self._add_request(app, src["id"], user): from flask import request request.decoded_token = {"sub": user} response = ManageSourceFiles().post() assert response.status_code == 200 assert apply_mock.call_count == 1 # No key → predetermined task_id is None, Celery generates one. assert apply_mock.call_args.kwargs["task_id"] is None n = pg_conn.execute( text("SELECT count(*) FROM task_dedup") ).scalar() assert n == 0 def test_header_records_dedup_row_with_predetermined_id(self, app, pg_conn): from sqlalchemy import text from application.api.user.sources.upload import ManageSourceFiles user = "alice-mgr-rec" src = _seed_source(pg_conn, user=user, file_path="/data") fake_storage = MagicMock() apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.tasks.reingest_source_task.apply_async", apply_mock, ), self._add_request(app, src["id"], user, key="mgr-key-1"): from flask import request request.decoded_token = {"sub": user} response = ManageSourceFiles().post() assert response.status_code == 200 assert apply_mock.call_count == 1 predetermined_id = apply_mock.call_args.kwargs["task_id"] assert predetermined_id is not None assert response.json["reingest_task_id"] == predetermined_id row = pg_conn.execute( text( "SELECT task_id, task_name, status FROM task_dedup " "WHERE idempotency_key = :k" ), {"k": f"{user}:mgr-key-1"}, ).fetchone() assert row is not None assert row[0] == predetermined_id assert row[1] == "reingest_source_task" assert row[2] == "pending" def test_same_key_second_post_returns_cached(self, app, pg_conn): from application.api.user.sources.upload import ManageSourceFiles user = "alice-mgr-rep" src = _seed_source(pg_conn, user=user, file_path="/data") fake_storage = MagicMock() apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.tasks.reingest_source_task.apply_async", apply_mock, ): with self._add_request(app, src["id"], user, key="mgr-rep"): from flask import request request.decoded_token = {"sub": user} first = ManageSourceFiles().post() with self._add_request(app, src["id"], user, key="mgr-rep"): from flask import request request.decoded_token = {"sub": user} second = ManageSourceFiles().post() assert first.status_code == 200 assert second.status_code == 200 # Loser short-circuits before any storage mutation. assert apply_mock.call_count == 1 # Loser's response carries the winner's task_id, not the # original 200-with-added_files payload. # ``manage_source_files`` aliases ``task_id`` -> # ``reingest_task_id`` in the cached payload so the dedup # response shape matches the fresh-request response (FileTree # keys reingest correlation on ``reingest_task_id`` / # ``source_id``). assert second.json["reingest_task_id"] == first.json["reingest_task_id"] # Cached ``source_id`` must equal the real source row id (not # the helper's uuid5-of-key) so FileTree's SSE correlation on # ``event.scope.id === result.source_id`` keeps working. assert second.json["source_id"] == first.json["source_id"] assert second.json["source_id"] == str(src["id"]) # Confirm the loser never invoked the file-save path. assert fake_storage.save_file.call_count == 1 def test_remove_same_key_second_post_returns_real_source_id( self, app, pg_conn ): """Regression: the ``remove`` cached branch used to leave the helper's synthetic ``source_id`` (uuid5 of the scoped key) in place. The reingest worker publishes SSE events tagged with the real source row id, so the cached response had to be patched to match what the fresh response returns — otherwise FileTree's SSE correlation silently fails on every idempotent retry and the user never sees the directory refresh. """ from application.api.user.sources.upload import ManageSourceFiles user = "alice-mgr-rmrep" src = _seed_source( pg_conn, user=user, file_path="/data", file_name_map={"a.txt": "a.txt"}, ) fake_storage = MagicMock() fake_storage.file_exists.return_value = True apply_mock = _apply_async_mock() def _do_remove(): return app.test_request_context( "/api/manage_source_files", method="POST", data={ "source_id": str(src["id"]), "operation": "remove", "file_paths": json.dumps(["a.txt"]), }, content_type="multipart/form-data", headers={"Idempotency-Key": "mgr-rmrep"}, ) with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.tasks.reingest_source_task.apply_async", apply_mock, ): with _do_remove(): from flask import request request.decoded_token = {"sub": user} first = ManageSourceFiles().post() with _do_remove(): from flask import request request.decoded_token = {"sub": user} second = ManageSourceFiles().post() assert first.status_code == 200 assert second.status_code == 200 assert apply_mock.call_count == 1 assert second.json["reingest_task_id"] == first.json["reingest_task_id"] # The contract under test: cached source_id matches the fresh # response (the real source row id), not the helper's uuid5. assert second.json["source_id"] == first.json["source_id"] assert second.json["source_id"] == str(src["id"]) def test_remove_directory_same_key_second_post_returns_real_source_id( self, app, pg_conn ): """Same regression as the ``remove`` test, for the ``remove_directory`` branch. """ from application.api.user.sources.upload import ManageSourceFiles user = "alice-mgr-rmdir-rep" src = _seed_source( pg_conn, user=user, file_path="/data", file_name_map={"sub/a.txt": "a.txt"}, ) fake_storage = MagicMock() fake_storage.is_directory.return_value = True fake_storage.remove_directory.return_value = True apply_mock = _apply_async_mock() def _do_remove_dir(): return app.test_request_context( "/api/manage_source_files", method="POST", data={ "source_id": str(src["id"]), "operation": "remove_directory", "directory_path": "sub", }, content_type="multipart/form-data", headers={"Idempotency-Key": "mgr-rmdir-rep"}, ) with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.tasks.reingest_source_task.apply_async", apply_mock, ): with _do_remove_dir(): from flask import request request.decoded_token = {"sub": user} first = ManageSourceFiles().post() with _do_remove_dir(): from flask import request request.decoded_token = {"sub": user} second = ManageSourceFiles().post() assert first.status_code == 200 assert second.status_code == 200 assert apply_mock.call_count == 1 assert second.json["reingest_task_id"] == first.json["reingest_task_id"] assert second.json["source_id"] == first.json["source_id"] assert second.json["source_id"] == str(src["id"]) def test_concurrent_same_key_only_one_apply_async(self, app, pg_engine): """N parallel same-key POSTs → exactly one apply_async.""" from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from application.api.user.sources.upload import ManageSourceFiles from application.storage.db.repositories.sources import ( SourcesRepository, ) user = "alice-mgr-race" with pg_engine.begin() as conn: src = SourcesRepository(conn).create( "race-src", user_id=user, file_path="/data", ) fake_storage = MagicMock() apply_mock = _apply_async_mock() @contextmanager def _engine_session(): with pg_engine.begin() as conn: yield conn @contextmanager def _engine_readonly(): with pg_engine.connect() as conn: yield conn def fire(_idx): # Patches sit outside the thread pool (see below); only the # per-thread Flask request context is set up inside. with self._add_request(app, src["id"], user, key="mgr-race"): from flask import request request.decoded_token = {"sub": user} return ManageSourceFiles().post() # ``unittest.mock.patch`` is not thread-safe; set up the # module-attribute patches once before fanning out so every # thread sees the mock instead of racing on save/restore. with patch( "application.api.user.sources.upload.db_session", _engine_session, ), patch( "application.api.user.sources.upload.db_readonly", _engine_readonly, ), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.tasks.reingest_source_task.apply_async", apply_mock, ), ThreadPoolExecutor(max_workers=8) as ex: responses = list(ex.map(fire, range(8))) assert all(r.status_code == 200 for r in responses) assert apply_mock.call_count == 1 def test_remove_directory_failure_releases_claim(self, app, pg_conn): """When storage.remove_directory returns False the handler must release the dedup row so a client retry can win the claim. Without the release, the next retry would silently 200-cache to a task_id that was never enqueued. """ from sqlalchemy import text from application.api.user.sources.upload import ManageSourceFiles user = "alice-mgr-rmfail" src = _seed_source(pg_conn, user=user, file_path="/data") fake_storage = MagicMock() fake_storage.is_directory.return_value = True fake_storage.remove_directory.return_value = False apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.tasks.reingest_source_task.apply_async", apply_mock, ), app.test_request_context( "/api/manage_source_files", method="POST", data={ "source_id": str(src["id"]), "operation": "remove_directory", "directory_path": "subdir", }, content_type="multipart/form-data", headers={"Idempotency-Key": "mgr-rmfail"}, ): from flask import request request.decoded_token = {"sub": user} response = ManageSourceFiles().post() assert response.status_code == 500 assert apply_mock.call_count == 0 # Claim row was released so a retry can re-claim. n = pg_conn.execute( text( "SELECT count(*) FROM task_dedup " "WHERE idempotency_key = :k AND status = 'pending'" ), {"k": f"{user}:mgr-rmfail"}, ).scalar() assert n == 0 def test_storage_save_failure_releases_claim(self, app, pg_conn): """Regression: an exception from ``storage.save_file`` after the claim must release the dedup row. Pre-fix the outer ``except`` logged + 500'd without releasing, so a retry within 24h returned a cached predetermined ``task_id`` for a task that never enqueued. """ from sqlalchemy import text from application.api.user.sources.upload import ManageSourceFiles user = "alice-mgr-storefail" src = _seed_source(pg_conn, user=user, file_path="/data") fake_storage = MagicMock() fake_storage.save_file.side_effect = RuntimeError("disk full") apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.tasks.reingest_source_task.apply_async", apply_mock, ), self._add_request(app, src["id"], user, key="mgr-storefail"): from flask import request request.decoded_token = {"sub": user} response = ManageSourceFiles().post() assert response.status_code == 500 assert apply_mock.call_count == 0 n = pg_conn.execute( text( "SELECT count(*) FROM task_dedup " "WHERE idempotency_key = :k" ), {"k": f"{user}:mgr-storefail"}, ).scalar() assert n == 0 def test_apply_async_failure_releases_claim(self, app, pg_conn): """If the broker is unreachable, ``apply_async`` raises *after* the claim was made. The claim row must be released so a retry can re-claim with a fresh task_id. """ from sqlalchemy import text from application.api.user.sources.upload import ManageSourceFiles user = "alice-mgr-brokerdown" src = _seed_source(pg_conn, user=user, file_path="/data") fake_storage = MagicMock() def _broker_down(*args, **kwargs): raise ConnectionError("broker unreachable") with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.tasks.reingest_source_task.apply_async", side_effect=_broker_down, ), self._add_request(app, src["id"], user, key="mgr-brokerdown"): from flask import request request.decoded_token = {"sub": user} response = ManageSourceFiles().post() assert response.status_code == 500 n = pg_conn.execute( text( "SELECT count(*) FROM task_dedup " "WHERE idempotency_key = :k" ), {"k": f"{user}:mgr-brokerdown"}, ).scalar() assert n == 0 def test_db_update_failure_releases_claim(self, app, pg_conn): """A DB blip during the ``file_name_map`` update happens after storage mutated and the claim was made. Must still release. """ from sqlalchemy import text from application.api.user.sources.upload import ManageSourceFiles from application.storage.db.repositories import sources as src_module user = "alice-mgr-dbfail" src = _seed_source(pg_conn, user=user, file_path="/data") fake_storage = MagicMock() apply_mock = _apply_async_mock() original_update = src_module.SourcesRepository.update def _explode(self, *args, **kwargs): raise RuntimeError("db transient") src_module.SourcesRepository.update = _explode try: with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.tasks.reingest_source_task.apply_async", apply_mock, ), self._add_request(app, src["id"], user, key="mgr-dbfail"): from flask import request request.decoded_token = {"sub": user} response = ManageSourceFiles().post() finally: src_module.SourcesRepository.update = original_update assert response.status_code == 500 # apply_async never ran because the DB raise pre-empted it. assert apply_mock.call_count == 0 n = pg_conn.execute( text( "SELECT count(*) FROM task_dedup " "WHERE idempotency_key = :k" ), {"k": f"{user}:mgr-dbfail"}, ).scalar() assert n == 0 def test_successful_path_keeps_claim_for_worker(self, app, pg_conn): """The claim row must persist after a successful ``apply_async`` — the worker owns the predetermined task_id and same-key retries should resolve to the in-flight task, not re-enqueue. """ from sqlalchemy import text from application.api.user.sources.upload import ManageSourceFiles user = "alice-mgr-keep" src = _seed_source(pg_conn, user=user, file_path="/data") fake_storage = MagicMock() apply_mock = _apply_async_mock() with _patch_db(pg_conn), patch( "application.api.user.sources.upload.StorageCreator.get_storage", return_value=fake_storage, ), patch( "application.api.user.tasks.reingest_source_task.apply_async", apply_mock, ), self._add_request(app, src["id"], user, key="mgr-keep"): from flask import request request.decoded_token = {"sub": user} response = ManageSourceFiles().post() assert response.status_code == 200 # Row is still there in pending status — worker has not finalised yet. row = pg_conn.execute( text( "SELECT status FROM task_dedup " "WHERE idempotency_key = :k" ), {"k": f"{user}:mgr-keep"}, ).fetchone() assert row is not None assert row[0] == "pending"