Be able to check for None flag and empty string flag (#395)

Co-authored-by: kmontemayor <kyle.e.montemayor@gmail.com>

Author: kmontemayor2-sc

python/gigl/orchestration/kubeflow/runner.py

@@ -146,7 +146,10 @@ def _assert_required_flags(args: argparse.Namespace) -> None:
     for flag in required_flags:
         if not hasattr(args, flag):
             missing_flags.append(flag)
-        elif len(getattr(args, flag)) == 0:
+        flag = getattr(args, flag)
+        if flag is None:
+            missing_flags.append(flag)
+        elif not flag:
             missing_values.append(flag)
 
     if missing_flags:

python/tests/unit/orchestration/kubeflow/kfp_runner_test.py

@@ -2,6 +2,7 @@
 
 from gigl.common.logger import Logger
 from gigl.orchestration.kubeflow.runner import (
+    _assert_required_flags,
     _get_parser,
     _parse_additional_job_args,
     _parse_labels,
@@ -66,6 +67,48 @@ def test_parse_args_from_cli(self):
         self.assertEqual(parsed_args, expected_parsed_args)
         self.assertEqual(parsed_labels, expected_parsed_labels)
 
+    def test_assert_required_flags_missing_value(self):
+        """Test that _assert_required_flags raises ValueError when a required flag has no value."""
+        parser = _get_parser()
+        # Parse with RUN action but task_config_uri not provided (will be None)
+        args = parser.parse_args(
+            [
+                "--action=run",
+                "--resource_config_uri=gs://bucket/resource_config.yaml",
+                "--task_config_uri=",
+            ]
+        )
+        with self.assertRaises(ValueError):
+            _assert_required_flags(args)
+
+    def test_assert_required_flags_none_value(self):
+        """Test that _assert_required_flags raises ValueError when a required flag has no value."""
+        parser = _get_parser()
+        # Parse with RUN action but task_config_uri not provided (will be None)
+        args = parser.parse_args(
+            [
+                "--action=run",
+                "--resource_config_uri=gs://bucket/resource_config.yaml",
+            ]
+        )
+        with self.assertRaises(ValueError):
+            _assert_required_flags(args)
+
+    def test_assert_required_flags_success(self):
+        """Test that _assert_required_flags succeeds when all required flags are present."""
+        parser = _get_parser()
+        # Parse with RUN action and all required flags
+        args = parser.parse_args(
+            [
+                "--action=run",
+                "--task_config_uri=gs://bucket/task_config.yaml",
+                "--resource_config_uri=gs://bucket/resource_config.yaml",
+            ]
+        )
+
+        # Should not raise any exception
+        _assert_required_flags(args)
+
 
 if __name__ == "__main__":
     unittest.main()