fix(restore): drop user-owned collations in the --empty pre-clean

The drop_sql loop covered relations, routines, sequences and types but
not pg_collation, so a dump's CREATE COLLATION (OpenProject's ICU
public.versions_name) aborted the ON_ERROR_STOP replay with 'collation
already exists'. Add the fifth UNION ALL branch; DROP COLLATION IF
EXISTS public.<name> CASCADE rides the existing loop, and the loop
already drops every user table, so CASCADE fallout is absorbed by the
IF EXISTS no-ops.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-12 02:55:29 +02:00
parent 6cb0b8a548
commit 7a7ec57b54
2 changed files with 48 additions and 0 deletions

View File

@@ -88,6 +88,11 @@ BEGIN
OR (t.typtype = 'c' AND EXISTS ( OR (t.typtype = 'c' AND EXISTS (
SELECT 1 FROM pg_class c2 SELECT 1 FROM pg_class c2
WHERE c2.oid = t.typrelid AND c2.relkind = 'c'))) WHERE c2.oid = t.typrelid AND c2.relkind = 'c')))
UNION ALL
SELECT col.collname AS name, 'COLLATION' AS type
FROM pg_collation col JOIN pg_namespace n ON n.oid = col.collnamespace
WHERE n.nspname = 'public'
AND pg_get_userbyid(col.collowner) = current_user
) LOOP ) LOOP
EXECUTE format('DROP %s IF EXISTS public.%I CASCADE', r.type, r.name); EXECUTE format('DROP %s IF EXISTS public.%I CASCADE', r.type, r.name);
END LOOP; END LOOP;

View File

@@ -0,0 +1,43 @@
"""The --empty pre-clean drop must cover every object class a dump can
re-CREATE ahead of tables; a missed class aborts the replay under
ON_ERROR_STOP (OpenProject's ICU collation public.versions_name did)."""
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from baudolo.restore.db import postgres
class TestPostgresEmptyDrop(unittest.TestCase):
def _drop_sql(self) -> str:
with tempfile.NamedTemporaryFile(suffix=".sql") as fh:
Path(fh.name).write_bytes(b"SELECT 1;\n")
with mock.patch.object(postgres, "docker_exec") as run:
postgres.restore_postgres_sql(
container="c",
db_name="db",
user="u",
password="p",
sql_path=fh.name,
empty=True,
)
first_call = run.call_args_list[0]
return first_call.kwargs["stdin"].decode()
def test_drop_covers_collations(self) -> None:
sql = self._drop_sql()
self.assertIn("pg_collation", sql)
self.assertIn("'COLLATION' AS type", sql)
self.assertIn("pg_get_userbyid(col.collowner) = current_user", sql)
def test_drop_still_covers_the_other_classes(self) -> None:
sql = self._drop_sql()
for marker in ("pg_class", "pg_proc", "'SEQUENCE' AS type", "'TYPE' AS type"):
self.assertIn(marker, sql)
self.assertIn("DROP %s IF EXISTS public.%I CASCADE", sql)
if __name__ == "__main__":
unittest.main()