feat: add test_secrets for k8s secrets verification

Author: Nissan Pow

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

@@ -0,0 +1,29 @@
+from metaflow import FlowSpec, step, secrets, project
+
+
+@project(name="hello_secrets")
+class HelloSecretsFlow(FlowSpec):
+    @step
+    def start(self):
+        from metaflow import metaflow_version
+
+        print(f"In start step and using metaflow: {metaflow_version.get_version()}")
+        print("HelloSecretsFlow is starting.")
+        self.next(self.hello)
+
+    @secrets(sources=["mf_secrets_test"])
+    @step
+    def hello(self):
+        import os
+
+        self.secrets = os.environ["mf_secrets_test"]
+        print("Hello secrets: %s" % self.secrets)
+        self.next(self.end)
+
+    @step
+    def end(self):
+        pass
+
+
+if __name__ == "__main__":
+    HelloSecretsFlow()

test/ux/core/test_basic.py

@@ -150,6 +150,22 @@ def test_timeout(exec_mode, decospecs, compute_env, tag, scheduler_config):
     assert run["work"].task.data.done is True, "Timeout step did not complete"
 
 
+@pytest.mark.scheduler_only
+def test_secrets(exec_mode, decospecs, compute_env, tag, scheduler_config):
+    """Test that k8s secrets can be read inside a step."""
+    run = execute_test_flow(
+        flow_name="basic/hellosecrets.py",
+        exec_mode=exec_mode,
+        decospecs=decospecs,
+        tag=tag,
+        scheduler_config=scheduler_config,
+        test_name="secrets",
+        tl_args_extra={"env": compute_env},
+    )
+    assert run.successful, "Run was not successful"
+    assert run["hello"].task.data.secrets == "hellosecrets", "Secret value mismatch"
+
+
 @pytest.mark.conda
 def test_hello_conda(exec_mode, decospecs, compute_env, tag, scheduler_config):
     run = execute_test_flow(