Fix airflowctl pools export/import dropping pool team ownership (#73132)

Multi-team deployments assign pools to a team via team_name. Backing pools
up and restoring them silently lost that ownership: the exported file never
carried the field, and an import with --action-on-existing-key overwrite
sent it as null, which the server then wrote back over the existing pool.

The deprecated airflow pools export/import that this command supersedes does
round-trip team_name, so the gap only bites users who follow the migration
notice onto airflowctl.

Co-authored-by: Eason09053360 <185830721+Eason09053360@users.noreply.github.com>

Author: Eason09053360

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

@@ -68,6 +68,7 @@ def export(args, api_client: Client = NEW_API_CLIENT) -> None:
                 "scheduled_slots": pool.scheduled_slots,
                 "open_slots": pool.open_slots,
                 "deferred_slots": pool.deferred_slots,
+                "team_name": pool.team_name,
             }
             for pool in pools_response.pools
         ]
@@ -105,6 +106,7 @@ def _import_helper(api_client: Client, filepath: Path, action_on_existence: Bulk
                 slots=pool_config["slots"],
                 description=pool_config.get("description", ""),
                 include_deferred=pool_config.get("include_deferred", False),
+                team_name=pool_config.get("team_name"),
             )
         )
 

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

@@ -136,6 +136,23 @@ def test_import_action_on_existing_key(
         action = call_args["pools"].actions[0]
         assert action.action_on_existence == expected_enum
 
+    def test_import_forwards_team_name(self, mock_client, tmp_path):
+        """Test that team_name in the file reaches the server instead of being sent as null."""
+        pools_file = tmp_path / "pools.json"
+        pools_file.write_text(json.dumps([{"name": "test_pool", "slots": 1, "team_name": "data-team"}]))
+
+        mock_response = mock.MagicMock()
+        mock_response.success = ["test_pool"]
+        mock_response.errors = []
+        mock_bulk_builder = mock.MagicMock()
+        mock_bulk_builder.create = mock_response
+        mock_client.pools.bulk.return_value = mock_bulk_builder
+
+        pool_command.import_(mock.MagicMock(file=pools_file, action_on_existing_key="overwrite"))
+
+        sent = mock_client.pools.bulk.call_args[1]["pools"].model_dump(mode="json")
+        assert sent["actions"][0]["entities"][0]["team_name"] == "data-team"
+
 
 class TestPoolExportCommand:
     """Test cases for pool export command."""
@@ -155,6 +172,7 @@ def test_export_json_to_file(self, mock_client, tmp_path, capsys):
             "scheduled_slots": 0,
             "open_slots": 1,
             "deferred_slots": 0,
+            "team_name": None,
         }
         # Create a mock response with proper dictionary attributes
         mock_pools = mock.MagicMock()
@@ -191,6 +209,7 @@ def test_export_non_json_uses_airflow_console(self, mock_client, tmp_path, outpu
             "scheduled_slots": 0,
             "open_slots": 5,
             "deferred_slots": 0,
+            "team_name": None,
         }
         mock_pools = mock.MagicMock()
         mock_pools.pools = [type("Pool", (), pool_attrs)()]
@@ -204,6 +223,32 @@ def test_export_non_json_uses_airflow_console(self, mock_client, tmp_path, outpu
                 output=output_format,
             )
 
+    def test_export_includes_team_name(self, mock_client, tmp_path):
+        """Test that export writes team_name so the file can be imported back without losing it."""
+        export_file = tmp_path / "export.json"
+        pool_attrs = {
+            "name": "test_pool",
+            "slots": 1,
+            "description": "Test pool",
+            "include_deferred": True,
+            "occupied_slots": 0,
+            "running_slots": 0,
+            "queued_slots": 0,
+            "scheduled_slots": 0,
+            "open_slots": 1,
+            "deferred_slots": 0,
+            "team_name": "data-team",
+        }
+        mock_pools = mock.MagicMock()
+        mock_pools.pools = [type("Pool", (), pool_attrs)()]
+        mock_pools.total_entries = 1
+        mock_client.pools.list.return_value = mock_pools
+
+        pool_command.export(mock.MagicMock(file=export_file, output="json"))
+
+        exported_data = json.loads(export_file.read_text())
+        assert exported_data[0]["team_name"] == "data-team"
+
     def test_export_failure(self, mock_client, tmp_path):
         """Test pool export with API failure."""
         export_file = tmp_path / "export.json"