fix: a T&E re-upload that fails every row changes nothing (#215)

api/services/te_loader.py

@@ -215,19 +215,18 @@ def _err(row_number: int, column: str, value: str, reason: str) -> None:
                     "info",
                 )
 
-            # Detect reupload before inserting so the all-failed rollback
-            # does not undo the registry cleanup (#170 + reupload compat).
+            # A re-upload overwrites the earlier load, so it needs delete permission
+            # up front. The cleanup itself waits until a row has actually landed
+            # (#170, #215): a re-upload that fails every row must change nothing.
             file_hash = hashlib.sha256(content.encode("utf-8")).hexdigest()
             cur.execute(
                 sql.SQL(
                     "SELECT 1 FROM {s}.csv_files WHERE file_name = %s OR file_hash = %s"
                 ).format(s=sql.Identifier(uploads_schema())),
                 (file_name, file_hash),
             )
-            is_reupload = cur.fetchone() is not None
-            if is_reupload:
+            if cur.fetchone() is not None:
                 _require_delete(session)
-                _cleanup_prior_registry_entries(cur, file_name, file_hash)
 
             # Pre-validate every row client-side; drop failures into row_errors.
             failed_row_count = 0
@@ -301,7 +300,7 @@ def _err(row_number: int, column: str, value: str, reason: str) -> None:
 
             # #170: if every row failed, nothing was migrated. Roll back and
             # preserve any prior dataset rather than cleaning it up.
-            if data_rows and inserted == 0 and not is_reupload:
+            if data_rows and inserted == 0:
                 conn.rollback()
                 msg = f"None of the {len(data_rows)} rows could be imported into {target_table}, so nothing was saved."
                 if row_errors:
@@ -312,11 +311,9 @@ def _err(row_number: int, column: str, value: str, reason: str) -> None:
                         "totalRows": len(data_rows), "failedRows": failed_row_count,
                         "insertedRows": 0, "columns": columns, "logs": logs}
 
-            # Register the load in the shared registry (mode='te').
-            # Reupload cleanup was done before inserts; for first uploads
-            # clean any stale entries now (usually a no-op).
-            if not is_reupload:
-                _cleanup_prior_registry_entries(cur, file_name, file_hash)
+            # Register the load in the shared registry (mode='te'), replacing any
+            # earlier entry for this file. Only reached once a row has landed.
+            _cleanup_prior_registry_entries(cur, file_name, file_hash)
             try:
                 cur.execute(
                     sql.SQL(

docs/TEST_CASES.md

@@ -1,7 +1,7 @@
 # Test Catalogue
 
 > Auto-generated by `scripts/test_catalogue.py` — do not edit by hand.
-> 726 tests across 37 files.
+> 732 tests across 38 files.
 
 To regenerate: `python3 scripts/test_catalogue.py`
 
@@ -147,7 +147,7 @@ To regenerate: `python3 scripts/test_catalogue.py`
 
 | `TeUploadPipeline.test_unknown_target_table_returns_error` | integration | Uploading to a table not in TE_TABLES must return status=error. |
 
-| `TeUploadPipeline.test_te_reupload_same_file_replaces_registry` | integration | Re-uploading the same CSV in T&E mode replaces the registry entry. |
+| `TeUploadPipeline.test_te_reupload_same_file_replaces_registry` | integration | A T&E re-upload that loads rows replaces the registry entry. |
 
 | `TeUploadPipeline.test_te_client_side_type_validation_reports_column_and_value` | integration | BUG-028: a bad value on a client-validatable column (BOOLEAN here) |
 
@@ -225,7 +225,7 @@ To regenerate: `python3 scripts/test_catalogue.py`
 
 | `test_unknown_target_table_returns_error` | — | Uploading to a table not in TE_TABLES must return status=error. |
 
-| `test_te_reupload_same_file_replaces_registry` | — | Re-uploading the same CSV in T&E mode replaces the registry entry. |
+| `test_te_reupload_same_file_replaces_registry` | — | A T&E re-upload that loads rows replaces the registry entry. |
 
 | `test_te_client_side_type_validation_reports_column_and_value` | — | BUG-028: a bad value on a client-validatable column (BOOLEAN here) |
 
@@ -1553,6 +1553,22 @@ To regenerate: `python3 scripts/test_catalogue.py`
 
 | `test_dynamic_overwrite_of_te_origin_entry_succeeds` | — | A dynamic-mode overwrite of a filename previously registered in |
 
+## `tests/test_te_reupload_registry.py`
+
+| Test | Markers | Objective |
+|---|---|---|
+| `TeReuploadRegistry.test_all_failed_reupload_keeps_the_earlier_registry_record` | integration | A T&E re-upload that fails every row leaves the earlier load's registry record alone. |
+
+| `TeReuploadRegistry.test_reupload_that_inserts_rows_still_replaces_the_registry_record` | integration | A T&E re-upload that loads at least one row still replaces the registry record. |
+
+| `TeReuploadRegistry.test_all_failed_te_upload_keeps_a_prior_dynamic_table_of_the_same_name` | integration | An all-failed T&E upload does not drop a dynamic table registered under the same filename (#170). |
+
+| `test_all_failed_reupload_keeps_the_earlier_registry_record` | — | A T&E re-upload that fails every row leaves the earlier load's registry record alone. |
+
+| `test_reupload_that_inserts_rows_still_replaces_the_registry_record` | — | A T&E re-upload that loads at least one row still replaces the registry record. |
+
+| `test_all_failed_te_upload_keeps_a_prior_dynamic_table_of_the_same_name` | — | An all-failed T&E upload does not drop a dynamic table registered under the same filename (#170). |
+
 ## `tests/test_tenant_isolation.py`
 
 | Test | Markers | Objective |

tests/test_api_coverage.py

@@ -453,11 +453,14 @@ def test_unknown_target_table_returns_error(self):
         self.assertIn("non_existent_table", body.get("message", ""))
 
     def test_te_reupload_same_file_replaces_registry(self):
-        """Re-uploading the same CSV in T&E mode replaces the registry entry."""
-        csv = self._org_csv(f"{self.tag}_reup")
-        r1 = self._upload_te(csv)
+        """A T&E re-upload that loads rows replaces the registry entry.
+
+        An identical file would collide on the unique name and change nothing
+        (#215, see tests/test_te_reupload_registry.py).
+        """
+        r1 = self._upload_te(self._org_csv(f"{self.tag}_reup_1"))
         self.assertEqual(r1.json()["status"], "ok", r1.json())
-        r2 = self._upload_te(csv)
+        r2 = self._upload_te(self._org_csv(f"{self.tag}_reup_2"))
         self.assertEqual(r2.json()["status"], "ok", r2.json())
         files = self.client.get("/api/csv/files").json()
         matches = [f for f in files if f["file_name"] == f"{self.tag}.csv"]

tests/test_te_loader_integration.py

@@ -173,8 +173,13 @@ def test_reupload_cleans_prior_registry_entry(self):
         self.__class__._inserted.extend(names)
         fname = f"te_reupload_{os.getpid()}.csv"
 
+        # The second load must insert something: an identical file collides on the
+        # unique name, fails every row and (#215) changes nothing.
+        content_v2 = content.replace(names[0], names[0] + "_v2")
+        self.__class__._inserted.append(names[0] + "_v2")
+
         first  = self._upload_te(content, file_name=fname)
-        second = self._upload_te(content, file_name=fname)
+        second = self._upload_te(content_v2, file_name=fname)
 
         self.assertEqual(first["status"],  "ok", first)
         self.assertEqual(second["status"], "ok", second)

tests/test_te_reupload_registry.py

@@ -0,0 +1,126 @@
+"""#215: a T&E re-upload that fails every row must leave the earlier load alone.
+
+The registry (``csv_files``, what "Migrated files" shows) is the ledger a
+reconciliation reads. Before the fix, re-uploading a file whose rows all failed
+still deleted the earlier load's registry record and replaced it with "0 loaded",
+while the earlier load's rows stayed in the table.
+
+Needs a live PostgreSQL with the dev schema deployed (see tests/test_api_integration.py).
+"""
+from __future__ import annotations
+
+import os
+import unittest
+
+import pytest
+from fastapi.testclient import TestClient
+
+from api.config import settings
+from api.db import Conn
+from api.main import app
+
+HEADER = "name,org_type,country"
+
+
+@pytest.mark.integration
+class TeReuploadRegistry(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls._orig_allow_destructive = settings.allow_destructive
+        settings.allow_destructive = True  # tearDown deletes its tagged files
+        cls.ctx = TestClient(app)
+        try:
+            cls.client = cls.ctx.__enter__()
+        except Exception as exc:  # noqa: BLE001
+            raise AssertionError(f"Could not start the API (is PostgreSQL up, is PGPASSWORD set?): {exc}") from None
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.ctx.__exit__(None, None, None)
+        settings.allow_destructive = cls._orig_allow_destructive
+
+    def setUp(self):
+        self.tag = f"reup215_{os.getpid()}_{self._testMethodName}"
+        self.file_name = f"{self.tag}.csv"
+
+    def tearDown(self):
+        for f in self.client.get("/api/csv/files").json():
+            if self.tag in f["file_name"]:
+                self.client.delete(f"/api/csv/files/{f['id']}")
+        with Conn() as conn:
+            with conn.cursor() as cur:
+                cur.execute(f"DELETE FROM {settings.TE_SCHEMA}.organisations WHERE name LIKE %s", (self.tag + "%",))
+            conn.commit()
+
+    # -- helpers -------------------------------------------------------------
+
+    def _upload_te(self, content: str):
+        return self.client.post("/api/csv/upload", json={
+            "fileName": self.file_name, "content": content, "mode": "te", "targetTable": "organisations",
+        }).json()
+
+    def _registry(self) -> list[dict]:
+        return [f for f in self.client.get("/api/csv/files").json() if f["file_name"] == self.file_name]
+
+    def _loaded_orgs(self) -> int:
+        with Conn() as conn:
+            with conn.cursor() as cur:
+                cur.execute(f"SELECT count(*) FROM {settings.TE_SCHEMA}.organisations WHERE name LIKE %s",
+                            (self.tag + "%",))
+                return cur.fetchone()[0]
+
+    # -- the bug -------------------------------------------------------------
+
+    def test_all_failed_reupload_keeps_the_earlier_registry_record(self):
+        """A T&E re-upload that fails every row leaves the earlier load's registry record alone."""
+        # Two valid rows and one that the check constraint rejects.
+        csv = f"{HEADER}\n{self.tag}_a,government,AU\n{self.tag}_b,prime,AU\n{self.tag}_bad,not_a_type,AU\n"
+        first = self._upload_te(csv)
+        self.assertEqual((first["status"], first["insertedRows"], first["failedRows"]), ("ok", 2, 1), first)
+        before = self._registry()
+        self.assertEqual([(r["row_count"], r["failed_rows"]) for r in before], [(2, 1)])
+
+        # The same file again: both valid rows now collide on the unique name, so every row fails.
+        second = self._upload_te(csv)
+        self.assertEqual(second["status"], "error", second)
+        self.assertEqual(second["insertedRows"], 0, second)
+
+        after = self._registry()
+        self.assertEqual([(r["id"], r["row_count"], r["failed_rows"]) for r in after],
+                         [(before[0]["id"], 2, 1)],
+                         "the earlier load's registry record must survive an all-failed re-upload")
+        self.assertEqual(self._loaded_orgs(), 2, "registry and table must still agree")
+
+    # -- guard against over-correcting ---------------------------------------
+
+    def test_reupload_that_inserts_rows_still_replaces_the_registry_record(self):
+        """A T&E re-upload that loads at least one row still replaces the registry record."""
+        first = self._upload_te(f"{HEADER}\n{self.tag}_a,government,AU\n")
+        self.assertEqual(first["status"], "ok", first)
+        old_id = self._registry()[0]["id"]
+
+        second = self._upload_te(f"{HEADER}\n{self.tag}_b,prime,AU\n{self.tag}_bad,not_a_type,AU\n")
+        self.assertEqual((second["status"], second["insertedRows"], second["failedRows"]), ("ok", 1, 1), second)
+
+        reg = self._registry()
+        self.assertEqual(len(reg), 1, "a re-upload replaces the record, it does not add a second")
+        self.assertNotEqual(reg[0]["id"], old_id)
+        self.assertEqual((reg[0]["row_count"], reg[0]["failed_rows"]), (1, 1))
+
+    # -- #170's original scenario, in the same place -------------------------
+
+    def test_all_failed_te_upload_keeps_a_prior_dynamic_table_of_the_same_name(self):
+        """An all-failed T&E upload does not drop a dynamic table registered under the same filename (#170)."""
+        dyn = self.client.post("/api/csv/upload", json={
+            "fileName": self.file_name, "content": "col_a,col_b\n1,2\n", "mode": "dynamic",
+        }).json()
+        self.assertEqual(dyn["status"], "ok", dyn)
+        table = dyn["tableName"]
+
+        te = self._upload_te(f"{HEADER}\n{self.tag}_bad,not_a_type,AU\n")
+        self.assertEqual(te["status"], "error", te)
+
+        reg = self._registry()
+        self.assertEqual([(r["mode"], r["table_name"]) for r in reg], [("dynamic", table)])
+        self.assertEqual(self.client.get(f"/api/csv/tables/{table}/rows").status_code, 200,
+                         "the prior dynamic table must not have been dropped")