Function workflow_default

test/feature-flag-consistency/mzcompose.py:47–155

Test the consistency with another mz version.

(c: Composition, parser: WorkflowArgumentParser)

def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
    """
    Test the consistency with another mz version.
    """

    c.down(destroy_volumes=True)

    test = FeatureFlagConsistencyTest()

    parser.add_argument("--configuration", action="append", default=[], type=str)

    parser.add_argument(
        "--evaluation-strategy",
        default=EVALUATION_STRATEGY_NAME_DFR,
        type=str,
        choices=INTERNAL_EVALUATION_STRATEGY_NAMES,
    )

    parser.add_argument(
        "--other-tag",
        type=str,
        default="common-ancestor",
    )

    parser.add_argument(
        "--query-output-mode",
        type=lambda mode: QueryOutputMode[mode.upper()],
        choices=QUERY_OUTPUT_MODE_CHOICES,
        default=QueryOutputMode.SELECT,
    )

    args = test.parse_output_consistency_input_args(parser)

    port_mz_default_internal, port_mz_system_internal = 6875, 6877
    port_mz_default_this, port_mz_system_this = 6875, 6877
    port_mz_default_other, port_mz_system_other = 16875, 16877

    configurations = args.configuration

    if len(configurations) == 0:
        configurations = FEATURE_FLAG_CONFIGURATION_PAIRS.keys()

    runtime_per_config_in_sec = args.max_runtime_in_sec / len(configurations)

    test_summaries = []

    for configuration in configurations:
        test.set_configuration_pair_by_name(configuration)
        assert test.flag_configuration_pair is not None
        print(f"Running flag configuration: {test.flag_configuration_pair.name}")

        with c.override(
            Materialized(
                name="mz_this",
                ports=[
                    f"{port_mz_default_this}:{port_mz_default_internal}",
                    f"{port_mz_system_this}:{port_mz_system_internal}",
                ],