diff --git a/src/apm_cli/cli.py b/src/apm_cli/cli.py index 5263aafbd..d276439bb 100644 --- a/src/apm_cli/cli.py +++ b/src/apm_cli/cli.py @@ -12,6 +12,7 @@ from apm_cli.commands._helpers import ( ERROR, RESET, + WARNING, _check_and_notify_updates, print_version, ) @@ -74,34 +75,117 @@ def cli(ctx): cli.add_command(marketplace_search, name="search") +def _get_current_code_page() -> "Optional[int]": + """Get current Windows console code page using WinAPI. + + Returns the code page number (e.g., 65001 for UTF-8, 950 for CP950). + Returns None if detection fails or on non-Windows platforms. + """ + if sys.platform != "win32": + return None + + try: + kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] + return kernel32.GetConsoleOutputCP() + except Exception: + return None + + +def _code_page_to_encoding_name(cp: int) -> str: + """Map code page number to readable encoding name. + + Args: + cp: Code page number (e.g., 950, 65001). + + Returns: + Human-readable encoding name or fallback name. + """ + cp_map = { + 65001: "UTF-8", + 950: "cp950 (Traditional Chinese)", + 936: "cp936 (Simplified Chinese)", + 932: "cp932 (Japanese)", + 949: "cp949 (Korean)", + 1252: "cp1252 (Western European)", + 1251: "cp1251 (Cyrillic)", + } + return cp_map.get(cp, f"cp{cp}") + + +def _try_switch_to_utf8() -> bool: + """Try to switch console to UTF-8 (code page 65001). + + This function: + 1. Checks if console is already UTF-8. + 2. If not, attempts to switch using SetConsoleCP/SetConsoleOutputCP. + 3. Verifies success by re-checking the code page. + + Returns: + True if already UTF-8 or successfully switched, False otherwise. + """ + if sys.platform != "win32": + return True + + try: + kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] + + # Check current console code page + current_cp = kernel32.GetConsoleOutputCP() + if current_cp == 65001: + return True # Already UTF-8 + + # Attempt to switch to UTF-8 + kernel32.SetConsoleOutputCP(65001) + kernel32.SetConsoleCP(65001) + + # Verify success + new_cp = kernel32.GetConsoleOutputCP() + return new_cp == 65001 + except Exception: + return False + + +def _warn_encoding_issue(failed_cp: int) -> None: + """Warn user if console UTF-8 switch failed. + + Args: + failed_cp: The code page that failed to switch from. + """ + encoding_name = _code_page_to_encoding_name(failed_cp) + click.echo( + f"\n{WARNING}Warning: Console is {encoding_name}, UTF-8 switch failed.{RESET}\n", + err=True, + ) + click.echo( + f"{WARNING}Display issues may occur. Suggestions:{RESET}", + err=True, + ) + click.echo(" - Run: chcp 65001 (if available)", err=True) + click.echo(" - Or use: Windows Terminal or VS Code terminal\n", err=True) + + def _configure_encoding() -> None: """Configure stdout/stderr for full Unicode on Windows. - The default Windows console encoding (cp1252) cannot represent many Unicode - characters used in APM output (box-drawing, check marks, arrows, etc.). + The default Windows console encoding (cp1252 or cp950) cannot represent many + Unicode characters used in APM output (box-drawing, check marks, arrows, etc.). This function: - 1. Sets ``PYTHONIOENCODING`` so child processes and redirected pipes - default to UTF-8. - 2. Switches the console codepage to 65001 (UTF-8) via the Win32 API so - the terminal itself renders UTF-8 byte sequences correctly. - 3. Reconfigures the Python text-mode streams to UTF-8. + 1. Attempts to switch console to UTF-8 (code page 65001) via WinAPI. + 2. Sets ``PYTHONIOENCODING`` for child processes. + 3. Reconfigures Python text-mode streams to UTF-8. + 4. Only warns if UTF-8 switch fails. On non-Windows platforms this is a no-op. """ if sys.platform != "win32": return - # 1. Help child processes / pipes default to UTF-8 - os.environ.setdefault("PYTHONIOENCODING", "utf-8") + # 1. Try to switch console to UTF-8 + utf8_success = _try_switch_to_utf8() - # 2. Switch the console codepage to UTF-8 - try: - kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] - kernel32.SetConsoleOutputCP(65001) - kernel32.SetConsoleCP(65001) - except (OSError, AttributeError): - pass # not a real console or ctypes unavailable + # 2. Help child processes / pipes default to UTF-8 + os.environ.setdefault("PYTHONIOENCODING", "utf-8") # 3. Reconfigure Python streams to UTF-8 for name in ("stdout", "stderr"): @@ -115,6 +199,12 @@ def _configure_encoding() -> None: except Exception: pass + # 4. Warn only if UTF-8 switch failed + if not utf8_success: + current_cp = _get_current_code_page() + if current_cp and current_cp != 65001: + _warn_encoding_issue(current_cp) + def main(): """Main entry point for the CLI.""" diff --git a/src/apm_cli/commands/_helpers.py b/src/apm_cli/commands/_helpers.py index 3bf0ba9e4..6f220455f 100644 --- a/src/apm_cli/commands/_helpers.py +++ b/src/apm_cli/commands/_helpers.py @@ -363,7 +363,7 @@ def _auto_detect_author(): try: result = subprocess.run( - ["git", "config", "user.name"], capture_output=True, text=True, timeout=5 + ["git", "config", "user.name"], capture_output=True, text=True, encoding="utf-8", timeout=5 ) if result.returncode == 0 and result.stdout.strip(): return result.stdout.strip() @@ -382,6 +382,7 @@ def _auto_detect_description(project_name): ["git", "config", "--get", "remote.origin.url"], capture_output=True, text=True, + encoding="utf-8", timeout=5, ) if result.returncode == 0 and result.stdout.strip(): diff --git a/src/apm_cli/commands/install.py b/src/apm_cli/commands/install.py index 10813f347..9e8b0119b 100644 --- a/src/apm_cli/commands/install.py +++ b/src/apm_cli/commands/install.py @@ -436,6 +436,7 @@ def _validate_package_exists(package, verbose=False, auth_resolver=None): cmd, capture_output=True, text=True, + encoding="utf-8", timeout=30, env=validate_env, ) diff --git a/src/apm_cli/core/token_manager.py b/src/apm_cli/core/token_manager.py index 1bc152acb..93afdd603 100644 --- a/src/apm_cli/core/token_manager.py +++ b/src/apm_cli/core/token_manager.py @@ -111,6 +111,7 @@ def resolve_credential_from_git(host: str) -> Optional[str]: input=f"protocol=https\nhost={host}\n\n", capture_output=True, text=True, + encoding="utf-8", timeout=GitHubTokenManager._get_credential_timeout(), env={**os.environ, 'GIT_TERMINAL_PROMPT': '0', 'GIT_ASKPASS': '' if sys.platform != 'win32' else 'echo'}, diff --git a/src/apm_cli/deps/github_downloader.py b/src/apm_cli/deps/github_downloader.py index 9776f7efb..0e5a15355 100644 --- a/src/apm_cli/deps/github_downloader.py +++ b/src/apm_cli/deps/github_downloader.py @@ -1557,7 +1557,7 @@ def _try_sparse_checkout(self, dep_ref: DependencyReference, temp_clone_path: Pa for cmd in cmds: result = subprocess.run( cmd, cwd=str(temp_clone_path), env=env, - capture_output=True, text=True, timeout=120, + capture_output=True, text=True, encoding="utf-8", timeout=120, ) if result.returncode != 0: _debug(f"Sparse-checkout step failed ({' '.join(cmd)}): {result.stderr.strip()}") diff --git a/src/apm_cli/policy/discovery.py b/src/apm_cli/policy/discovery.py index 437bc1160..d4e13ee9f 100644 --- a/src/apm_cli/policy/discovery.py +++ b/src/apm_cli/policy/discovery.py @@ -134,6 +134,7 @@ def _extract_org_from_git_remote( ["git", "remote", "get-url", "origin"], capture_output=True, text=True, + encoding="utf-8", cwd=project_root, timeout=5, ) diff --git a/src/apm_cli/runtime/codex_runtime.py b/src/apm_cli/runtime/codex_runtime.py index 002531b93..9e88e8804 100644 --- a/src/apm_cli/runtime/codex_runtime.py +++ b/src/apm_cli/runtime/codex_runtime.py @@ -41,8 +41,8 @@ def execute_prompt(self, prompt_content: str, **kwargs) -> str: stdout=subprocess.PIPE, stderr=subprocess.STDOUT, # Merge stderr into stdout for streaming text=True, + encoding="utf-8", bufsize=1, # Line buffered - universal_newlines=True ) output_lines = [] @@ -107,6 +107,7 @@ def get_runtime_info(self) -> Dict[str, Any]: ["codex", "--version"], capture_output=True, text=True, + encoding="utf-8", timeout=10 ) diff --git a/src/apm_cli/runtime/copilot_runtime.py b/src/apm_cli/runtime/copilot_runtime.py index 46dd49b90..485057f11 100644 --- a/src/apm_cli/runtime/copilot_runtime.py +++ b/src/apm_cli/runtime/copilot_runtime.py @@ -59,8 +59,8 @@ def execute_prompt(self, prompt_content: str, **kwargs) -> str: stdout=subprocess.PIPE, stderr=subprocess.STDOUT, # Merge stderr into stdout for streaming text=True, + encoding="utf-8", bufsize=1, # Line buffered - universal_newlines=True ) output_lines = [] @@ -125,6 +125,7 @@ def get_runtime_info(self) -> Dict[str, Any]: ["copilot", "--version"], capture_output=True, text=True, + encoding="utf-8", timeout=10 ) diff --git a/src/apm_cli/runtime/llm_runtime.py b/src/apm_cli/runtime/llm_runtime.py index df24869ff..a43e0839d 100644 --- a/src/apm_cli/runtime/llm_runtime.py +++ b/src/apm_cli/runtime/llm_runtime.py @@ -20,8 +20,8 @@ def __init__(self, model_name: Optional[str] = None): # Verify llm CLI is available try: - result = subprocess.run(['llm', '--version'], - capture_output=True, text=True, check=True) + result = subprocess.run(['llm', '--version'], + capture_output=True, text=True, encoding="utf-8", check=True) except (subprocess.CalledProcessError, FileNotFoundError): raise RuntimeError("llm CLI not found. Please install: pip install llm") @@ -52,8 +52,8 @@ def execute_prompt(self, prompt_content: str, **kwargs) -> str: stdout=subprocess.PIPE, stderr=subprocess.STDOUT, # Merge stderr into stdout for streaming text=True, + encoding="utf-8", bufsize=1, # Line buffered - universal_newlines=True ) output_lines = [] @@ -86,8 +86,8 @@ def list_available_models(self) -> Dict[str, Any]: Dict[str, Any]: Dictionary of available models and their info """ try: - result = subprocess.run(['llm', 'models', 'list'], - capture_output=True, text=True, check=True) + result = subprocess.run(['llm', 'models', 'list'], + capture_output=True, text=True, encoding="utf-8", check=True) models = {} for line in result.stdout.strip().split('\n'): if line.strip(): @@ -136,8 +136,8 @@ def is_available() -> bool: bool: True if runtime is available, False otherwise """ try: - subprocess.run(['llm', '--version'], - capture_output=True, text=True, check=True) + subprocess.run(['llm', '--version'], + capture_output=True, text=True, encoding="utf-8", check=True) return True except (subprocess.CalledProcessError, FileNotFoundError): return False diff --git a/src/apm_cli/runtime/manager.py b/src/apm_cli/runtime/manager.py index c3ebc3946..bc78b79dd 100644 --- a/src/apm_cli/runtime/manager.py +++ b/src/apm_cli/runtime/manager.py @@ -53,7 +53,7 @@ def get_embedded_script(self, script_name: str) -> str: bundle_dir = Path(sys._MEIPASS) script_path = bundle_dir / "scripts" / "runtime" / script_name if script_path.exists(): - return script_path.read_text() + return script_path.read_text(encoding="utf-8") # Fall back to direct file access for development # Look for scripts relative to the repo structure @@ -61,7 +61,7 @@ def get_embedded_script(self, script_name: str) -> str: repo_root = current_file.parent.parent.parent.parent # Go up to repo root script_path = repo_root / "scripts" / "runtime" / script_name if script_path.exists(): - return script_path.read_text() + return script_path.read_text(encoding="utf-8") raise FileNotFoundError(f"Script not found: {script_name}") except Exception as e: @@ -85,15 +85,15 @@ def get_token_helper_script(self) -> str: bundle_dir = Path(sys._MEIPASS) script_path = bundle_dir / "scripts" / "github-token-helper.sh" if script_path.exists(): - return script_path.read_text() - + return script_path.read_text(encoding="utf-8") + # Fall back to direct file access for development # Look for scripts relative to the repo structure current_file = Path(__file__) repo_root = current_file.parent.parent.parent.parent # Go up to repo root script_path = repo_root / "scripts" / "github-token-helper.sh" if script_path.exists(): - return script_path.read_text() + return script_path.read_text(encoding="utf-8") raise FileNotFoundError("github-token-helper.sh not found") except Exception as e: @@ -111,35 +111,35 @@ def run_embedded_script(self, script_content: str, common_content: str, if self._is_windows: # Write common utilities as PowerShell common_script = temp_path / "setup-common.ps1" - common_script.write_text(common_content) + common_script.write_text(common_content, encoding="utf-8") # Write GitHub token helper (empty on Windows) token_helper_content = self.get_token_helper_script() if token_helper_content: token_helper_script = temp_path / "github-token-helper.ps1" - token_helper_script.write_text(token_helper_content) + token_helper_script.write_text(token_helper_content, encoding="utf-8") # Write main script as PowerShell main_script = temp_path / "setup-script.ps1" - main_script.write_text(script_content) + main_script.write_text(script_content, encoding="utf-8") else: # Write common utilities as bash common_script = temp_path / "setup-common.sh" - common_script.write_text(common_content) + common_script.write_text(common_content, encoding="utf-8") common_script.chmod(0o755) # Write GitHub token helper try: token_helper_content = self.get_token_helper_script() token_helper_script = temp_path / "github-token-helper.sh" - token_helper_script.write_text(token_helper_content) + token_helper_script.write_text(token_helper_content, encoding="utf-8") token_helper_script.chmod(0o755) except Exception as e: click.echo(f"{Fore.YELLOW}[!] Token helper not available, scripts may use fallback authentication: {e}{Style.RESET_ALL}") # Write main script as bash main_script = temp_path / "setup-script.sh" - main_script.write_text(script_content) + main_script.write_text(script_content, encoding="utf-8") main_script.chmod(0o755) # Execute script with environment that includes npm authentication @@ -166,6 +166,7 @@ def run_embedded_script(self, script_content: str, common_content: str, cwd=temp_dir, capture_output=False, # Show output to user text=True, + encoding="utf-8", env=env ) return result.returncode == 0 @@ -255,6 +256,7 @@ def list_runtimes(self) -> Dict[str, Dict[str, str]]: version_cmd, capture_output=True, text=True, + encoding="utf-8", timeout=5 ) if result.returncode == 0: @@ -293,7 +295,8 @@ def remove_runtime(self, runtime_name: str) -> bool: result = subprocess.run( ["npm", "uninstall", "-g", "@github/copilot"], capture_output=True, - text=True + text=True, + encoding="utf-8", ) if result.returncode == 0: click.echo(f"{Fore.GREEN}[+] Successfully removed {runtime_name} runtime{Style.RESET_ALL}") diff --git a/src/apm_cli/version.py b/src/apm_cli/version.py index cd68a8144..20adddfe1 100644 --- a/src/apm_cli/version.py +++ b/src/apm_cli/version.py @@ -85,6 +85,7 @@ def get_build_sha() -> str: cwd=repo_root, capture_output=True, text=True, + encoding="utf-8", timeout=5, ) if result.returncode == 0: diff --git a/tests/unit/policy/test_discovery.py b/tests/unit/policy/test_discovery.py index 250599001..98ff2c7f6 100644 --- a/tests/unit/policy/test_discovery.py +++ b/tests/unit/policy/test_discovery.py @@ -106,6 +106,7 @@ def test_successful_remote(self, mock_run): ["git", "remote", "get-url", "origin"], capture_output=True, text=True, + encoding="utf-8", cwd=Path("/fake"), timeout=5, ) diff --git a/tests/unit/test_init_command.py b/tests/unit/test_init_command.py index af21b9e79..6da5accb6 100644 --- a/tests/unit/test_init_command.py +++ b/tests/unit/test_init_command.py @@ -123,7 +123,7 @@ def test_init_existing_project_with_force(self): assert result.exit_code == 0 assert "APM project initialized successfully!" in result.output # Should overwrite the file with minimal structure - with open("apm.yml") as f: + with open("apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) # Minimal structure assert "dependencies" in config @@ -146,7 +146,7 @@ def test_init_preserves_existing_config(self): "description": "Custom description", "author": "Custom Author", } - with open("apm.yml", "w") as f: + with open("apm.yml", "w", encoding="utf-8") as f: yaml.dump(existing_config, f) result = self.runner.invoke(cli, ["init", "--yes"]) @@ -176,7 +176,7 @@ def test_init_interactive_mode(self): assert "Author" in result.output # Verify the interactive values were applied to apm.yml - with open("apm.yml") as f: + with open("apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) assert config["name"] == "my-test-project" assert config["version"] == "1.5.0" @@ -234,7 +234,7 @@ def test_init_validates_project_structure(self): project_path = Path(tmp_dir) / "test-project" # Verify apm.yml minimal structure - with open(project_path / "apm.yml") as f: + with open(project_path / "apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) assert config["name"] == "test-project" assert "version" in config @@ -273,7 +273,7 @@ def test_init_auto_detection(self): assert result.exit_code == 0 - with open("apm.yml") as f: + with open("apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) # Should auto-detect author from git assert config["author"] == "Test User" diff --git a/tests/unit/test_init_plugin.py b/tests/unit/test_init_plugin.py index 1b27237e0..60faf4704 100644 --- a/tests/unit/test_init_plugin.py +++ b/tests/unit/test_init_plugin.py @@ -68,7 +68,7 @@ def test_plugin_json_structure(self): result = self.runner.invoke(cli, ["init", "--plugin", "--yes"]) assert result.exit_code == 0, result.output - with open("plugin.json") as f: + with open("plugin.json", encoding="utf-8") as f: data = json.load(f) assert data["name"] == "my-plugin" @@ -90,7 +90,7 @@ def test_apm_yml_has_dev_dependencies(self): result = self.runner.invoke(cli, ["init", "--plugin", "--yes"]) assert result.exit_code == 0, result.output - with open("apm.yml") as f: + with open("apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) assert "devDependencies" in config @@ -179,7 +179,7 @@ def test_yes_mode_works_with_plugin(self): assert Path("apm.yml").exists() assert Path("plugin.json").exists() - with open("apm.yml") as f: + with open("apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) # --yes + --plugin uses 0.1.0 version assert config["version"] == "0.1.0" @@ -197,7 +197,7 @@ def test_plugin_flag_without_plugin(self): assert Path("apm.yml").exists() assert not Path("plugin.json").exists() - with open("apm.yml") as f: + with open("apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) assert "devDependencies" not in config finally: @@ -212,11 +212,11 @@ def test_plugin_version_defaults_to_0_1_0(self): try: self.runner.invoke(cli, ["init", "--plugin", "--yes"]) - with open("plugin.json") as f: + with open("plugin.json", encoding="utf-8") as f: data = json.load(f) assert data["version"] == "0.1.0" - with open("apm.yml") as f: + with open("apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) assert config["version"] == "0.1.0" finally: @@ -231,7 +231,7 @@ def test_plugin_author_is_object(self): try: self.runner.invoke(cli, ["init", "--plugin", "--yes"]) - with open("plugin.json") as f: + with open("plugin.json", encoding="utf-8") as f: data = json.load(f) assert isinstance(data["author"], dict) assert "name" in data["author"] @@ -265,7 +265,7 @@ def test_plugin_with_project_name_argument(self): assert (project_path / "apm.yml").exists() assert (project_path / "plugin.json").exists() - with open(project_path / "plugin.json") as f: + with open(project_path / "plugin.json", encoding="utf-8") as f: data = json.load(f) assert data["name"] == "cool-plugin" finally: @@ -293,7 +293,7 @@ def test_plugin_apm_yml_has_dependencies(self): try: self.runner.invoke(cli, ["init", "--plugin", "--yes"]) - with open("apm.yml") as f: + with open("apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) assert "dependencies" in config assert config["dependencies"] == {"apm": [], "mcp": []} diff --git a/tests/unit/test_install_command.py b/tests/unit/test_install_command.py index 282b6e0b9..e9bbe47a6 100644 --- a/tests/unit/test_install_command.py +++ b/tests/unit/test_install_command.py @@ -88,7 +88,7 @@ def test_install_no_apm_yml_with_packages_creates_minimal_apm_yml( assert Path("apm.yml").exists() # Verify apm.yml structure - with open("apm.yml") as f: + with open("apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) assert "dependencies" in config assert "apm" in config["dependencies"] @@ -125,7 +125,7 @@ def test_install_no_apm_yml_with_multiple_packages( assert Path("apm.yml").exists() # Verify both packages are in apm.yml - with open("apm.yml") as f: + with open("apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) assert "org1/pkg1" in config["dependencies"]["apm"] assert "org2/pkg2" in config["dependencies"]["apm"] @@ -147,7 +147,7 @@ def test_install_existing_apm_yml_preserves_behavior( "dependencies": {"apm": [], "mcp": []}, "scripts": {}, } - with open("apm.yml", "w") as f: + with open("apm.yml", "w", encoding="utf-8") as f: yaml.dump(existing_config, f) # Mock APMPackage @@ -165,7 +165,7 @@ def test_install_existing_apm_yml_preserves_behavior( assert "Created apm.yml" not in result.output # Verify original config is preserved - with open("apm.yml") as f: + with open("apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) assert config["name"] == "test-project" assert config["author"] == "Test Author" @@ -202,7 +202,7 @@ def test_install_auto_created_apm_yml_has_correct_metadata( assert Path("apm.yml").exists() # Verify auto-detected project name - with open("apm.yml") as f: + with open("apm.yml", encoding="utf-8") as f: config = yaml.safe_load(f) assert config["name"] == "my-awesome-project" assert "version" in config diff --git a/tests/unit/test_llm_runtime.py b/tests/unit/test_llm_runtime.py index 46c0209ce..5c0b517f8 100644 --- a/tests/unit/test_llm_runtime.py +++ b/tests/unit/test_llm_runtime.py @@ -17,8 +17,8 @@ def test_init_success(self, mock_run): runtime = LLMRuntime("gpt-4o-mini") assert runtime.model_name == "gpt-4o-mini" - mock_run.assert_called_once_with(['llm', '--version'], - capture_output=True, text=True, check=True) + mock_run.assert_called_once_with(['llm', '--version'], + capture_output=True, text=True, encoding="utf-8", check=True) @patch('apm_cli.runtime.llm_runtime.subprocess.run') def test_init_fallback(self, mock_run): diff --git a/tests/unit/test_runtime_manager.py b/tests/unit/test_runtime_manager.py index 25578c589..8fb21e46b 100644 --- a/tests/unit/test_runtime_manager.py +++ b/tests/unit/test_runtime_manager.py @@ -236,6 +236,7 @@ def test_copilot_npm_success(self): ["npm", "uninstall", "-g", "@github/copilot"], capture_output=True, text=True, + encoding="utf-8", ) def test_copilot_npm_failure(self):