Function workflow_dev

test/mz-deploy/mzcompose.py:1039–1233

Test the dev command — developer-role overlay for inner-loop iteration.

(c: Composition, parser: WorkflowArgumentParser)

def workflow_dev(c: Composition, parser: WorkflowArgumentParser) -> None:
    """Test the dev command — developer-role overlay for inner-loop iteration."""
    setup_base(c)

    # Apply v1 for infrastructure, then stage + wait + promote so the ops
    # schema's MV is tracked as a production deployment on the `compute`
    # cluster. That promotion is what the dev-time cluster guard reads from.
    result = run_mz_deploy(c, "dev-basic/v1", "apply")
    assert result.returncode == 0, f"apply v1 failed: {result.stderr}"

    result = run_mz_deploy(
        c, "dev-basic/v1", "stage", "--deploy-id", "v1", "--allow-dirty"
    )
    assert result.returncode == 0, f"stage v1 failed: {result.stderr}"

    result = run_mz_deploy(
        c,
        "dev-basic/v1",
        "wait",
        "v1",
        "--timeout",
        "300",
        "--allowed-lag",
        "86400",
    )
    assert result.returncode == 0, f"wait v1 failed: {result.stderr}"

    result = run_mz_deploy(c, "dev-basic/v1", "promote", "v1", "--no-ready-check")
    assert result.returncode == 0, f"promote v1 failed: {result.stderr}"

    # Grant dev_user privileges on the app database so it can read production
    # tables when building the overlay MV (CREATEDB is already granted by
    # setup_base; only schema/table access needs to be added here).
    # Use mz_system (port 6877) because app is owned by deploy_user, not
    # materialize, so the default superuser cannot grant on it directly.
    c.sql(
        "GRANT USAGE ON DATABASE app TO dev_user",
        user="mz_system",
        port=6877,
    )
    rows = c.sql_query(
        "SELECT name FROM mz_schemas WHERE database_id = "
        "(SELECT id FROM mz_databases WHERE name = 'app')",
    )
    for (schema,) in rows:
        c.sql(
            f"GRANT USAGE ON SCHEMA app.{schema} TO dev_user",
            user="mz_system",
            port=6877,
        )
        c.sql(
            f"GRANT SELECT ON ALL TABLES IN SCHEMA app.{schema} TO dev_user",
            user="mz_system",
            port=6877,
        )

    # Provision a dedicated dev cluster that the `dev` profile routes to
    # via [dev.variables].compute_cluster = "compute_dev". The