Class PgDisruption

test/source-sink-errors/mzcompose.py:328–422

@dataclass
class PgDisruption:
    name: str
    breakage: Callable
    expected_error: str
    fixage: Callable | None

    def run_test(self, c: Composition) -> None:
        print(f"+++ Running disruption scenario {self.name}")
        seed = random.randint(0, 256**4)

        c.up(
            "postgres",
            "materialized",
            "clusterd",
            Service("testdrive", idle=True),
        )

        with c.override(
            Testdrive(
                no_reset=True,
                seed=seed,
                entrypoint_extra=["--initial-backoff=1s", "--backoff-factor=0"],
            )
        ):
            self.populate(c)
            self.breakage(c, seed)
            self.assert_error(c, self.expected_error)

            if self.fixage:
                self.fixage(c, seed)
                self.assert_recovery(c)

    def populate(self, c: Composition) -> None:
        # Create a source and a sink
        c.testdrive(dedent("""
                > CREATE SECRET pgpass AS 'postgres'
                > CREATE CONNECTION pg TO POSTGRES (
                    HOST postgres,
                    DATABASE postgres,
                    USER postgres,
                    PASSWORD SECRET pgpass
                  )

                $ postgres-execute connection=postgres://postgres:postgres@postgres
                ALTER USER postgres WITH replication;
                DROP SCHEMA IF EXISTS public CASCADE;
                CREATE SCHEMA public;

                DROP PUBLICATION IF EXISTS mz_source;
                CREATE PUBLICATION mz_source FOR ALL TABLES;

                CREATE TABLE source1 (f1 INTEGER PRIMARY KEY, f2 integer[]);
                INSERT INTO source1 VALUES (1, NULL);
                ALTER TABLE source1 REPLICA IDENTITY FULL;
                INSERT INTO source1 VALUES (2, NULL);

                > CREATE SOURCE "pg_source"
                  FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source');