Function run_test

test/sql-feature-flags/mzcompose.py:224–281

(c: Composition, args: argparse.Namespace)

def run_test(c: Composition, args: argparse.Namespace) -> None:
    c.up("redpanda", "materialized", Service("testdrive", idle=True))

    scenarios = (
        [globals()[args.scenario]]
        if args.scenario
        else all_subclasses(FeatureTestScenario)
    )

    # To add a new scenario create a new FeatureTestScenario subclass
    for scenario in scenarios:
        print(f"--- Running scenario {scenario.__name__} phase 1")
        c.testdrive(scenario.phase1())

        c.stop("materialized")
        c.up("materialized")

        print(f"--- Running scenario {scenario.__name__} phase 2")
        c.testdrive(scenario.phase2())

        materialized = Materialized(
            unsafe_mode=False,
            additional_system_parameter_defaults={
                scenario.feature_name(): "on",
            },
        )

        with c.override(materialized):
            c.stop("materialized")
            c.up("materialized")

            print(f"--- Running scenario {scenario.__name__} phase 3")
            c.testdrive(scenario.phase3())

    # Dedicated test for ALTER SYSTEM RESET ALL
    print("--- Running ALTER SYSTEM RESET ALL")
    tmp = [header("(phase reset-all)", drop_schema=False)]
    for scenario in scenarios:
        # Turn all features off.
        tmp.append(alter_system_set(scenario.feature_name(), "off"))

    # Run ALTER SYSTEM RESET ALL
    tmp.append(alter_system_reset_all())
    for scenario in scenarios:
        # Write each scenarios reset all data
        tmp.append(scenario.reset_all())

    # Create MZ config with all features set on by default
    materialized = Materialized(
        unsafe_mode=False,
        additional_system_parameter_defaults={
            scenario.feature_name(): "on" for scenario in scenarios
        },
    )
    with c.override(materialized):
        c.stop("materialized")
        c.up("materialized")
        c.testdrive("\n\n".join(tmp))