add validation for variables import JSON structure

airflow-ctl/src/airflowctl/ctl/commands/variable_command.py

@@ -29,6 +29,7 @@
     BulkCreateActionVariableBody,
     VariableBody,
 )
+from airflowctl.exceptions import AirflowCtlException
 
 
 @provide_api_client(kind=ClientKind.CLI)
@@ -48,11 +49,23 @@ def import_(args, api_client=NEW_API_CLIENT) -> list[str]:
             sys.exit(1)
 
     action_on_existence = BulkActionOnExistence(args.action_on_existing_key)
+
+    if not isinstance(var_json, dict):
+        raise AirflowCtlException(
+            "Invalid JSON format: expected a JSON object (key-value pairs), "
+            f"but got {type(var_json).__name__}."
+        )
+
     vars_to_update = []
     for k, v in var_json.items():
-        value, description = v, None
-        if isinstance(v, dict) and "value" in v:
-            value, description = v["value"], v.get("description")
+        if isinstance(v, dict):
+            if "value" not in v:
+                raise AirflowCtlException(f"Invalid format for key '{k}': missing 'value'.")
+            value = v["value"]
+            description = v.get("description")
+        else:
+            value = v
+            description = None
 
         vars_to_update.append(
             VariableBody(

airflow-ctl/tests/airflow_ctl/ctl/commands/test_variable_command.py

@@ -29,6 +29,7 @@
 )
 from airflowctl.ctl import cli_parser
 from airflowctl.ctl.commands import variable_command
+from airflowctl.exceptions import AirflowCtlException
 
 
 class TestCliVariableCommands:
@@ -134,3 +135,45 @@ def test_import_error(self, api_client_maker, tmp_path, monkeypatch):
                 self.parser.parse_args(["variables", "import", expected_json_path.as_posix()]),
                 api_client=api_client,
             )
+
+    def test_import_invalid_json_top_level_type(self, api_client_maker, tmp_path, monkeypatch):
+        api_client = api_client_maker(
+            path="/api/v2/variables",
+            response_json=self.bulk_response_success.model_dump(),
+            expected_http_status_code=200,
+            kind=ClientKind.CLI,
+        )
+
+        monkeypatch.chdir(tmp_path)
+        expected_json_path = tmp_path / self.export_file_name
+        invalid_variable_file = ["invalid", "payload"]
+
+        expected_json_path.write_text(json.dumps(invalid_variable_file))
+        with pytest.raises(AirflowCtlException, match="expected a JSON object"):
+            variable_command.import_(
+                self.parser.parse_args(["variables", "import", expected_json_path.as_posix()]),
+                api_client=api_client,
+            )
+
+    def test_import_missing_value_in_nested_object(self, api_client_maker, tmp_path, monkeypatch):
+        api_client = api_client_maker(
+            path="/api/v2/variables",
+            response_json=self.bulk_response_success.model_dump(),
+            expected_http_status_code=200,
+            kind=ClientKind.CLI,
+        )
+
+        monkeypatch.chdir(tmp_path)
+        expected_json_path = tmp_path / self.export_file_name
+        invalid_variable_file = {
+            self.key: {
+                "description": "value field missing",
+            }
+        }
+
+        expected_json_path.write_text(json.dumps(invalid_variable_file))
+        with pytest.raises(AirflowCtlException, match=f"Invalid format for key '{self.key}'"):
+            variable_command.import_(
+                self.parser.parse_args(["variables", "import", expected_json_path.as_posix()]),
+                api_client=api_client,
+            )