Function workflow_default

test/gcp/mzcompose.py:279–353

(c: Composition)

def workflow_default(c: Composition) -> None:
    bucket = _require_env("ICEBERG_GCS_BUCKET")
    project = _require_env("ICEBERG_GCP_PROJECT")
    sa_json_b64 = _require_env("ICEBERG_GCP_SA_JSON_B64")

    service_account = json.loads(base64.b64decode(sa_json_b64))

    # Per-run namespace so concurrent / repeated runs against the shared
    # bucket don't collide on table state. The embedded date lets the pre-test
    # sweep age out namespaces left behind by killed runs.
    seed = random.getrandbits(32)
    today = datetime.now(timezone.utc).strftime(NAMESPACE_DATE_FORMAT)
    namespace = f"{NAMESPACE_PREFIX}_{today}_{seed:08x}"
    table = "demo_table"
    # The .td inserts these three rows; verification asserts the table's live
    # row count matches.
    expected_rows = 3

    materialized = Materialized()
    with c.override(materialized):
        c.down()
        c.up("materialized")
        c.sql(
            port=6877,
            user="mz_system",
            sql="""
            ALTER SYSTEM SET enable_connection_validation_syntax = true;
            """,
        )

        # Mint once and reuse for verification + cleanup. Tokens last an hour;
        # minting up front also fails fast if the service-account key is broken.
        token = mint_gcp_access_token(service_account)
        # Bootstrap the catalog if absent; /v1/config 403s otherwise.
        ensure_catalog(token, project, bucket)
        # Discover the per-warehouse catalog prefix once; it's identical for
        # the verify and cleanup paths.
        prefix = resolve_warehouse_prefix(token, project, bucket)
        # Garbage-collect namespaces from previous runs that were killed before
        # their `finally` could run. Best-effort; logged failures don't block.
        _sweep_stale_biglake_namespaces(token, project, prefix)
        # BigLake doesn't auto-create namespaces on first commit, so we have to
        # pre-create (matching how the AWS test pre-creates the S3 Tables namespace).
        create_namespace(token, project, prefix, namespace)

        try:
            c.run_testdrive_files(
                "--no-reset",
                f"--var=gcp-sa-json-b64={sa_json_b64}",
                f"--var=gcs-bucket={bucket}",
                f"--var=namespace={namespace}",
                f"--var=table={table}",
                "gcp-iceberg-e2e.td",
            )

            try:
                _verify_sink_committed(
                    token, project, prefix, namespace, table, expected_rows