migration tests: pg-cdc-old-syntax: WIP

Author: nrainer-materialize

test/pg-cdc-old-syntax/mzcompose.py

@@ -21,7 +21,10 @@
 from materialize.mzcompose.composition import Composition, WorkflowArgumentParser
 from materialize.mzcompose.service import Service, ServiceConfig
 from materialize.mzcompose.services.materialized import Materialized
-from materialize.mzcompose.services.postgres import Postgres
+from materialize.mzcompose.services.postgres import (
+    CockroachOrPostgresMetadata,
+    Postgres, PostgresMetadata,
+)
 from materialize.mzcompose.services.test_certs import TestCerts
 from materialize.mzcompose.services.testdrive import Testdrive
 from materialize.mzcompose.services.toxiproxy import Toxiproxy
@@ -90,6 +93,7 @@ def create_postgres(
         },
     ),
     Testdrive(),
+    PostgresMetadata(),
     TestCerts(),
     Toxiproxy(),
     create_postgres(pg_version=None),
@@ -354,3 +358,86 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
 
             with c.test_case(name):
                 c.workflow(name)
+
+
+def workflow_migration(c: Composition, parser: WorkflowArgumentParser) -> None:
+    matching_files = glob.glob("*.td", root_dir="test/pg-cdc-old-syntax")
+    sharded_files: list[str] = sorted(
+        buildkite.shard_list(matching_files, lambda file: file)
+    )
+    print(f"Files: {sharded_files}")
+
+    ssl_ca = c.run("test-certs", "cat", "/secrets/ca.crt", capture=True).stdout
+    ssl_cert = c.run("test-certs", "cat", "/secrets/certuser.crt", capture=True).stdout
+    ssl_key = c.run("test-certs", "cat", "/secrets/certuser.key", capture=True).stdout
+    ssl_wrong_cert = c.run(
+        "test-certs", "cat", "/secrets/postgres.crt", capture=True
+    ).stdout
+    ssl_wrong_key = c.run(
+        "test-certs", "cat", "/secrets/postgres.key", capture=True
+    ).stdout
+
+    pg_version = get_targeted_pg_version(parser)
+
+    for file in sharded_files:
+
+        mz_old = Materialized(
+            name="materialized",
+            image="materialize/materialized:v0.122.0",
+            volumes_extra=["secrets:/share/secrets"],
+            metadata_store="postgres-metadata",
+            external_metadata_store=True,
+            additional_system_parameter_defaults={
+                "log_filter": "mz_storage::source::postgres=trace,debug,info,warn,error"
+            },
+        )
+
+        mz_new = Materialized(
+            name="materialized",
+            image=None,
+            volumes_extra=["secrets:/share/secrets"],
+            metadata_store="postgres-metadata",
+            external_metadata_store=True,
+            additional_system_parameter_defaults={
+                "log_filter": "mz_storage::source::postgres=trace,debug,info,warn,error",
+                "force_source_table_syntax": "true",
+            },
+        )
+        with c.override(mz_old, create_postgres(pg_version=pg_version)):
+            c.up("materialized", "test-certs", "postgres")
+
+            print(f"Running {file} with mz_old")
+
+            c.run_testdrive_files(
+                f"--var=ssl-ca={ssl_ca}",
+                f"--var=ssl-cert={ssl_cert}",
+                f"--var=ssl-key={ssl_key}",
+                f"--var=ssl-wrong-cert={ssl_wrong_cert}",
+                f"--var=ssl-wrong-key={ssl_wrong_key}",
+                f"--var=default-replica-size={Materialized.Size.DEFAULT_SIZE}-{Materialized.Size.DEFAULT_SIZE}",
+                f"--var=default-storage-size={Materialized.Size.DEFAULT_SIZE}-1",
+                "--no-reset",
+                file,
+            )
+            c.kill("materialized", wait=True)
+
+            with c.override(mz_new):
+                c.up("materialized")
+
+                print("Running mz_new")
+
+                source_names = c.sql_query(
+                    "SELECT name FROM mz_sources WHERE id LIKE 'u%';"
+                )
+
+                for row in source_names:
+                    source_name = row[0]
+                    print(f"Checking source: {source_name}")
+                    c.sql_query(f"SELECT count(*) FROM {source_name};")
+                else:
+                    print("No sources found")
+
+                c.kill("materialized", wait=True)
+                c.kill("postgres", wait=True)
+                c.rm("materialized")
+                c.rm("postgres")