Fix list-envs auth status for env names containing .json (#64677)

Author: henry3260

airflow-ctl/src/airflowctl/api/client.py

@@ -168,6 +168,11 @@ def input_cli_config_file(self) -> str:
         """Generate path for the CLI config file."""
         return f"{self.api_environment}.json"
 
+    @staticmethod
+    def token_key_for_environment(api_environment: str) -> str:
+        """Build the keyring/debug token key for a given environment name."""
+        return f"api_token_{api_environment}"
+
     def save(self, skip_keyring: bool = False):
         """
         Save the credentials to keyring and URL to disk as a file.
@@ -185,7 +190,7 @@ def save(self, skip_keyring: bool = False):
                 with open(
                     os.path.join(default_config_dir, f"debug_creds_{self.input_cli_config_file}"), "w"
                 ) as f:
-                    json.dump({f"api_token_{self.api_environment}": self.api_token}, f)
+                    json.dump({self.token_key_for_environment(self.api_environment): self.api_token}, f)
             else:
                 if skip_keyring:
                     return
@@ -198,7 +203,11 @@ def save(self, skip_keyring: bool = False):
                 for candidate in candidates:
                     if hasattr(candidate, "_get_new_password"):
                         candidate._get_new_password = _bounded_get_new_password
-                keyring.set_password("airflowctl", f"api_token_{self.api_environment}", self.api_token)  # type: ignore[arg-type]
+                keyring.set_password(
+                    "airflowctl",
+                    self.token_key_for_environment(self.api_environment),
+                    self.api_token,  # type: ignore[arg-type]
+                )
         except (NoKeyringError, NotImplementedError) as e:
             log.error(e)
             raise AirflowCtlKeyringException(
@@ -229,11 +238,13 @@ def load(self) -> Credentials:
                     )
                     with open(debug_creds_path) as df:
                         debug_credentials = json.load(df)
-                        self.api_token = debug_credentials.get(f"api_token_{self.api_environment}")
+                        self.api_token = debug_credentials.get(
+                            self.token_key_for_environment(self.api_environment)
+                        )
                 else:
                     try:
                         self.api_token = keyring.get_password(
-                            "airflowctl", f"api_token_{self.api_environment}"
+                            "airflowctl", self.token_key_for_environment(self.api_environment)
                         )
                     except ValueError as e:
                         # Incorrect keyring password

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

@@ -144,7 +144,7 @@ def list_envs(args) -> None:
         if filename.startswith("debug_creds_") or filename.endswith("_generated.json"):
             continue
 
-        env_name = filename.replace(".json", "")
+        env_name, _ = os.path.splitext(filename)
 
         # Try to read config file
         api_url = None
@@ -168,11 +168,11 @@ def list_envs(args) -> None:
                 if os.path.exists(debug_path):
                     with open(debug_path) as f:
                         debug_creds = json.load(f)
-                        if f"api_token_{env_name}" in debug_creds:
+                        if Credentials.token_key_for_environment(env_name) in debug_creds:
                             token_status = "authenticated"
             else:
                 # Check keyring
-                token = keyring.get_password("airflowctl", f"api_token_{env_name}")
+                token = keyring.get_password("airflowctl", Credentials.token_key_for_environment(env_name))
                 if token:
                     token_status = "authenticated"
         except NoKeyringError:

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

@@ -477,3 +477,21 @@ def test_list_envs_filters_special_files(self, monkeypatch):
 
             # Only production environment should be checked, not the special files
             mock_get_password.assert_called_once_with("airflowctl", "api_token_production")
+
+    def test_list_envs_environment_name_with_json_substring(self, monkeypatch):
+        """Test list-envs keeps '.json' substrings in environment name for key lookup."""
+        with (
+            tempfile.TemporaryDirectory() as temp_airflow_home,
+            patch("keyring.get_password") as mock_get_password,
+        ):
+            monkeypatch.setenv("AIRFLOW_HOME", temp_airflow_home)
+
+            with open(os.path.join(temp_airflow_home, "prod.json.region.json"), "w") as f:
+                json.dump({"api_url": "http://localhost:8080"}, f)
+
+            mock_get_password.return_value = "test_token"
+
+            args = self.parser.parse_args(["auth", "list-envs"])
+            auth_command.list_envs(args)
+
+            mock_get_password.assert_called_once_with("airflowctl", "api_token_prod.json.region")