port 5 NFLX tests to OSS: build_env_conda, resources_cpu/gpu, config_from_deployment, config_parser

Adapted from mf-6 (branch npow/ux-test-compute-env-refactor):
- test_build_env_conda: conda env built at runtime within a step
- test_resources_cpu: @resources(cpu=N, memory=N) on scheduler backends
- test_resources_gpu: @resources(gpu=1) deployment (skipped, no GPU nodes)
- test_config_from_deployment: Config + DeployedFlow.from_deployment()
- test_config_parser_flow_default: Config with requirements_txt_parser

Adaptations: @titus -> @resources, MaestroDeployedFlow -> DeployedFlow,
send_signals -> standard test harness patterns.

Author: Nissan Pow

test/ux/core/flows/basic/resources_cpu_flow.py

@@ -0,0 +1,45 @@
+from metaflow import FlowSpec, step, resources, project
+
+
+@project(name="hello_resources_cpu")
+class ResourcesCpuFlow(FlowSpec):
+
+    @step
+    def start(self):
+        from metaflow import metaflow_version
+
+        print(f"In start step and using metaflow: {metaflow_version.get_version()}")
+        print("ResourcesCpuFlow is starting.")
+        self.next(
+            self.default,
+            self.cpu2,
+            self.cpu4,
+        )
+
+    @resources()
+    @step
+    def default(self):
+        self.next(self.join)
+
+    @resources(cpu=2)
+    @step
+    def cpu2(self):
+        self.next(self.join)
+
+    @resources(cpu=4, memory=8000)
+    @step
+    def cpu4(self):
+        self.next(self.join)
+
+    @step
+    def join(self, inputs):
+        self.next(self.end)
+
+    @step
+    def end(self):
+        self.message = "Metaflow says: Hi Resources CPU!"
+        print("ResourcesCpuFlow is all done.")
+
+
+if __name__ == "__main__":
+    ResourcesCpuFlow()

test/ux/core/flows/basic/resources_gpu_flow.py

@@ -0,0 +1,31 @@
+from metaflow import FlowSpec, step, resources, project
+
+
+@project(name="hello_resources_gpu")
+class ResourcesGpuFlow(FlowSpec):
+
+    @step
+    def start(self):
+        from metaflow import metaflow_version
+
+        print(f"In start step and using metaflow: {metaflow_version.get_version()}")
+        print("ResourcesGpuFlow is starting.")
+        self.next(self.gpu_step)
+
+    @resources(gpu=1)
+    @step
+    def gpu_step(self):
+        # No actual GPU validation — devstack has no GPU nodes.
+        # This test verifies the @resources(gpu=N) decorator compiles
+        # and deploys successfully on each scheduler backend.
+        self.gpu_requested = True
+        self.next(self.end)
+
+    @step
+    def end(self):
+        self.message = "Metaflow says: Hi Resources GPU!"
+        print("ResourcesGpuFlow is all done.")
+
+
+if __name__ == "__main__":
+    ResourcesGpuFlow()

test/ux/core/flows/basic/self_building_env.py

@@ -0,0 +1,90 @@
+import subprocess
+import sys
+import tempfile
+
+from metaflow import (
+    FlowSpec,
+    Parameter,
+    conda,
+    current,
+    named_env,
+    project,
+    step,
+)
+
+
+def trigger_name_func(context):
+    return [current.project_flow_name + "Trigger"]
+
+
+@project(name="selfbuild")
+class BuildCondaEnvInStep(FlowSpec):
+    my_var = Parameter(
+        "my_var",
+        default=123,
+        external_artifact=trigger_name_func,
+        external_trigger=True,
+    )
+
+    @conda(disabled=True)
+    @step
+    def start(self):
+        from metaflow import metaflow_version
+
+        print(f"In start step and using metaflow: {metaflow_version.get_version()}")
+        full_run_id = current.run_id
+        if "-" in full_run_id:
+            base_run_id = full_run_id.split("-")[1]
+        else:
+            base_run_id = full_run_id
+        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8") as req_file:
+            req_file.write("itsdangerous==2.1.2")
+            req_file.flush()
+            subprocess.check_call(
+                [
+                    sys.executable,
+                    "-m",
+                    "metaflow.cmd.main_cli",
+                    "environment",
+                    "resolve",
+                    "--alias",
+                    "mlp/metaflow/test/build_in_step/id_%s_%s_%s"
+                    % (current.run_id, base_run_id, str(self.my_var)),
+                    "-r",
+                    req_file.name,
+                    "--python",
+                    "3.8.*",
+                ]
+            )
+        print(
+            "Build environment and aliased using mlp/metaflow/test/build_in_step/id_%s_%s_%s"
+            % (current.run_id, base_run_id, str(self.my_var))
+        )
+        self.next(self.fetch_old)
+
+    @conda(
+        name="mlp/metaflow/test/build_in_step/id_@{METAFLOW_RUN_ID}_@{METAFLOW_RUN_ID_BASE}_@{METAFLOW_INIT_MY_VAR}",
+        fetch_at_exec=True,
+    )
+    @step
+    def fetch_old(self):
+        import itsdangerous
+
+        print("Imported itsdangerous and found version %s" % itsdangerous.__version__)
+        self.found_version_old = itsdangerous.__version__
+        self.next(self.end)
+
+    @named_env(
+        name="mlp/metaflow/test/build_in_step/id_@{METAFLOW_RUN_ID}_@{METAFLOW_RUN_ID_BASE}_@{METAFLOW_INIT_MY_VAR}",
+        fetch_at_exec=True,
+    )
+    @step
+    def end(self):
+        import itsdangerous
+
+        print("Imported itsdangerous and found version %s" % itsdangerous.__version__)
+        self.found_version = itsdangerous.__version__
+
+
+if __name__ == "__main__":
+    BuildCondaEnvInStep()

test/ux/core/flows/config/config_parser.py

@@ -0,0 +1,54 @@
+from metaflow import (
+    Config,
+    FlowSpec,
+    Parameter,
+    config_expr,
+    current,
+    project,
+    pypi_base,
+    requirements_txt_parser,
+    step,
+)
+
+default_config = {"project_name": "config_parser"}
+
+
+def trigger_name_func(ctx):
+    return [current.project_flow_name + "Trigger"]
+
+
+@project(name=config_expr("cfg.project_name"))
+@pypi_base(**config_expr("req_config"))
+class ConfigParser(FlowSpec):
+
+    trigger_param = Parameter(
+        "trigger_param",
+        default="",
+        external_trigger=True,
+        external_artifact=trigger_name_func,
+    )
+    cfg = Config("cfg", default_value=default_config)
+
+    req_config = Config(
+        "req_config",
+        default="flows/config/config_parser_requirements.txt",
+        parser=requirements_txt_parser,
+    )
+
+    @step
+    def start(self):
+        from metaflow import metaflow_version
+
+        print(f"In start step and using metaflow: {metaflow_version.get_version()}")
+        import regex
+
+        self.lib_version = regex.__version__
+        self.next(self.end)
+
+    @step
+    def end(self):
+        pass
+
+
+if __name__ == "__main__":
+    ConfigParser()

test/ux/core/flows/config/config_parser_requirements.txt

@@ -0,0 +1,2 @@
+python==3.10.*
+regex==2024.11.6

test/ux/core/flows/config/hello_from_deployment_with_config.py

@@ -0,0 +1,26 @@
+from metaflow import FlowSpec, Config, step, config_expr, project
+
+default_config = {"batch_size": 32, "packages": {"pandas": "2.2.3"}}
+
+
+@project(name="hello_from_deployment_config")
+class HelloFromDeploymentFlowConfig(FlowSpec):
+    """Simple flow combining config, config_expr, and from_deployment testing."""
+
+    simple_config = Config("simple_config", default_value=default_config)
+
+    @step
+    def start(self):
+        from metaflow import metaflow_version
+
+        print(f"In start step and using metaflow: {metaflow_version.get_version()}")
+        self.batch_size = self.simple_config.batch_size
+        self.next(self.end)
+
+    @step
+    def end(self):
+        pass
+
+
+if __name__ == "__main__":
+    HelloFromDeploymentFlowConfig()

test/ux/core/test_basic.py

@@ -6,6 +6,7 @@
     execute_test_flow,
     deploy_flow_to_scheduler,
     wait_for_deployed_run,
+    wait_for_deployed_run_allow_failure,
     verify_run_provenance,
 )
 
@@ -172,3 +173,63 @@ def test_hello_conda(exec_mode, decospecs, compute_env, tag, scheduler_config):
     assert (
         run["combo"].task.data.itsdangerous_version == "2.2.0"
     ), "itsdangerous version incorrect"
+
+
+@pytest.mark.conda
+def test_build_env_conda(exec_mode, decospecs, compute_env, tag, scheduler_config):
+    """Verify that a conda environment can be built at runtime within a step."""
+    run = execute_test_flow(
+        flow_name="basic/self_building_env.py",
+        exec_mode=exec_mode,
+        decospecs=decospecs,
+        tag=tag,
+        scheduler_config=scheduler_config,
+        test_name="build_env_conda",
+        tl_args_extra={
+            "environment": "conda",
+            "env": compute_env,
+        },
+    )
+
+    assert run.successful, "Run was not successful"
+    assert (
+        run["fetch_old"].task.data.found_version_old == "2.1.2"
+    ), "fetch_old version incorrect"
+    assert run["end"].task.data.found_version == "2.1.2", "end version incorrect"
+
+
+@pytest.mark.scheduler_only
+def test_resources_cpu(exec_mode, decospecs, compute_env, tag, scheduler_config):
+    """Verify @resources(cpu=N, memory=N) deploys and runs on each scheduler backend."""
+    run = execute_test_flow(
+        flow_name="basic/resources_cpu_flow.py",
+        exec_mode=exec_mode,
+        decospecs=decospecs,
+        tag=tag,
+        scheduler_config=scheduler_config,
+        test_name="resources_cpu",
+        tl_args_extra={"env": compute_env},
+    )
+
+    assert run.successful, "Run was not successful"
+    assert (
+        run["end"].task.data.message == "Metaflow says: Hi Resources CPU!"
+    ), "Message didn't match"
+
+
+@pytest.mark.scheduler_only
+@pytest.mark.skip(reason="devstack has no GPU nodes")
+def test_resources_gpu(exec_mode, decospecs, compute_env, tag, scheduler_config):
+    """Verify @resources(gpu=1) deploys successfully (no actual GPU validation)."""
+    run = execute_test_flow(
+        flow_name="basic/resources_gpu_flow.py",
+        exec_mode=exec_mode,
+        decospecs=decospecs,
+        tag=tag,
+        scheduler_config=scheduler_config,
+        test_name="resources_gpu",
+        tl_args_extra={"env": compute_env},
+    )
+
+    assert run.successful, "Run was not successful"
+    assert run["gpu_step"].task.data.gpu_requested is True