diff --git a/src/apm_cli/config.py b/src/apm_cli/config.py index 6e758526d..7ce73396c 100644 --- a/src/apm_cli/config.py +++ b/src/apm_cli/config.py @@ -16,7 +16,7 @@ def ensure_config_exists(): os.makedirs(CONFIG_DIR) if not os.path.exists(CONFIG_FILE): - with open(CONFIG_FILE, "w") as f: + with open(CONFIG_FILE, "w", encoding="utf-8") as f: json.dump({"default_client": "vscode"}, f) @@ -32,7 +32,7 @@ def get_config(): if _config_cache is not None: return _config_cache ensure_config_exists() - with open(CONFIG_FILE) as f: + with open(CONFIG_FILE, encoding="utf-8") as f: _config_cache = json.load(f) return _config_cache @@ -53,7 +53,7 @@ def update_config(updates): config = get_config() config.update(updates) - with open(CONFIG_FILE, "w") as f: + with open(CONFIG_FILE, "w", encoding="utf-8") as f: json.dump(config, f, indent=2) _invalidate_config_cache() @@ -135,7 +135,7 @@ def unset_temp_dir() -> None: config = get_config() if "temp_dir" in config: del config["temp_dir"] - with open(CONFIG_FILE, "w") as f: + with open(CONFIG_FILE, "w", encoding="utf-8") as f: json.dump(config, f, indent=2) _invalidate_config_cache() @@ -185,7 +185,7 @@ def unset_copilot_cowork_skills_dir() -> None: config = get_config() if "copilot_cowork_skills_dir" in config: del config["copilot_cowork_skills_dir"] - with open(CONFIG_FILE, "w") as f: + with open(CONFIG_FILE, "w", encoding="utf-8") as f: json.dump(config, f, indent=2) _invalidate_config_cache() diff --git a/src/apm_cli/marketplace/client.py b/src/apm_cli/marketplace/client.py index 1d32a2c14..f33b850e4 100644 --- a/src/apm_cli/marketplace/client.py +++ b/src/apm_cli/marketplace/client.py @@ -88,13 +88,13 @@ def _read_cache(name: str) -> dict | None: if not os.path.exists(data_path) or not os.path.exists(meta_path): return None try: - with open(meta_path) as f: + with open(meta_path, encoding="utf-8") as f: meta = json.load(f) fetched_at = meta.get("fetched_at", 0) ttl = meta.get("ttl_seconds", _CACHE_TTL_SECONDS) if time.time() - fetched_at > ttl: return None # Expired - with open(data_path) as f: + with open(data_path, encoding="utf-8") as f: return json.load(f) except (json.JSONDecodeError, OSError, KeyError) as exc: logger.debug("Cache read failed for '%s': %s", name, exc) @@ -107,7 +107,7 @@ def _read_stale_cache(name: str) -> dict | None: if not os.path.exists(data_path): return None try: - with open(data_path) as f: + with open(data_path, encoding="utf-8") as f: return json.load(f) except (json.JSONDecodeError, OSError): return None @@ -118,9 +118,9 @@ def _write_cache(name: str, data: dict) -> None: data_path = _cache_data_path(name) meta_path = _cache_meta_path(name) try: - with open(data_path, "w") as f: + with open(data_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) - with open(meta_path, "w") as f: + with open(meta_path, "w", encoding="utf-8") as f: json.dump( {"fetched_at": time.time(), "ttl_seconds": _CACHE_TTL_SECONDS}, f, diff --git a/src/apm_cli/marketplace/registry.py b/src/apm_cli/marketplace/registry.py index 9faa4c0ec..4248349ef 100644 --- a/src/apm_cli/marketplace/registry.py +++ b/src/apm_cli/marketplace/registry.py @@ -32,7 +32,7 @@ def _ensure_file() -> str: ensure_config_exists() path = _marketplaces_path() if not os.path.exists(path): - with open(path, "w") as f: + with open(path, "w", encoding="utf-8") as f: json.dump({"marketplaces": []}, f, indent=2) return path @@ -51,7 +51,7 @@ def _load() -> list[MarketplaceSource]: return list(_registry_cache) path = _ensure_file() try: - with open(path) as f: + with open(path, encoding="utf-8") as f: data = json.load(f) except (json.JSONDecodeError, OSError) as exc: logger.warning("Failed to read %s: %s", path, exc) @@ -72,7 +72,7 @@ def _save(sources: list[MarketplaceSource]) -> None: path = _ensure_file() data = {"marketplaces": [s.to_dict() for s in sources]} tmp = path + ".tmp" - with open(tmp, "w") as f: + with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) os.replace(tmp, path) with _registry_lock: diff --git a/src/apm_cli/models/plugin.py b/src/apm_cli/models/plugin.py index 85c6c65c9..5db8fdcab 100644 --- a/src/apm_cli/models/plugin.py +++ b/src/apm_cli/models/plugin.py @@ -113,7 +113,7 @@ def from_path(cls, plugin_path: Path) -> "Plugin": f"Plugin metadata not found in any expected location: {plugin_path}" ) - with open(metadata_file) as f: + with open(metadata_file, encoding="utf-8") as f: metadata_dict = json.load(f) metadata = PluginMetadata.from_dict(metadata_dict) diff --git a/src/apm_cli/runtime/copilot_runtime.py b/src/apm_cli/runtime/copilot_runtime.py index 184ff1c84..b344c2831 100644 --- a/src/apm_cli/runtime/copilot_runtime.py +++ b/src/apm_cli/runtime/copilot_runtime.py @@ -207,7 +207,7 @@ def get_mcp_servers(self) -> dict[str, Any]: return {} try: - with open(mcp_config_path) as f: + with open(mcp_config_path, encoding="utf-8") as f: config = json.load(f) return config.get("servers", {}) except Exception as e: diff --git a/tests/unit/marketplace/test_marketplace_client.py b/tests/unit/marketplace/test_marketplace_client.py index 92fd9583f..93718cdfa 100644 --- a/tests/unit/marketplace/test_marketplace_client.py +++ b/tests/unit/marketplace/test_marketplace_client.py @@ -429,3 +429,36 @@ def test_different_hosts_different_keys(self): s1 = MarketplaceSource(name="mkt", owner="o", repo="r", host="a.com") s2 = MarketplaceSource(name="mkt", owner="o", repo="r", host="b.com") assert client_mod._cache_key(s1) != client_mod._cache_key(s2) + + +class TestCacheUtf8RoundTrip: + """Cache I/O preserves non-ASCII content (Windows cp1252/cp950 guard).""" + + def test_write_and_read_non_ascii(self, tmp_path): + data = { + "name": "Marketplace -- cafe", + "description": "\u4e2d\u6587 description", + "plugins": [{"name": "skill-\u958b\u59cb", "author": "cafe"}], + } + client_mod._write_cache("utf8-mkt", data) + + cached = client_mod._read_cache("utf8-mkt") + assert cached is not None + assert cached["name"] == "Marketplace -- cafe" + assert cached["description"] == "\u4e2d\u6587 description" + assert cached["plugins"][0]["name"] == "skill-\u958b\u59cb" + + def test_stale_cache_read_non_ascii(self, tmp_path): + import os as _os + + data = {"plugins": [{"name": "\u4e2d\u6587-skill"}]} + client_mod._write_cache("stale-mkt", data) + + # Drop the meta file so _read_cache treats the entry as missing and + # _read_stale_cache is the only path that returns content. + _os.remove(client_mod._cache_meta_path("stale-mkt")) + assert client_mod._read_cache("stale-mkt") is None + + stale = client_mod._read_stale_cache("stale-mkt") + assert stale is not None + assert stale["plugins"][0]["name"] == "\u4e2d\u6587-skill" diff --git a/tests/unit/marketplace/test_marketplace_registry.py b/tests/unit/marketplace/test_marketplace_registry.py index 1310933ce..b4204fecc 100644 --- a/tests/unit/marketplace/test_marketplace_registry.py +++ b/tests/unit/marketplace/test_marketplace_registry.py @@ -92,3 +92,45 @@ def test_corrupted_file_returns_empty(self, tmp_path): registry_mod._invalidate_cache() assert registry_mod.get_registered_marketplaces() == [] + + +class TestRegistryUtf8RoundTrip: + """Registry persistence preserves non-ASCII content (Windows cp1252/cp950 guard).""" + + def test_add_and_read_non_ascii_marketplace(self): + # Note: name/owner/repo are typically ASCII per the marketplace spec, + # but the registry file itself must still be UTF-8 to handle any + # non-ASCII content that may flow through future fields. We use a + # description-bearing source by writing a custom entry directly. + src = MarketplaceSource(name="cafe-mkt", owner="cafe-org", repo="plugins-\u958b\u59cb") + registry_mod.add_marketplace(src) + + # Force re-load from disk by clearing the cache. + registry_mod._invalidate_cache() + fetched = registry_mod.get_marketplace_by_name("cafe-mkt") + assert fetched is not None + assert fetched.repo == "plugins-\u958b\u59cb" + + def test_registry_file_is_readable_with_utf8_external_writes(self, tmp_path): + """A registry file written externally with raw UTF-8 (ensure_ascii=False) + must still load cleanly. This is the regression case for cp1252/cp950 + Windows locales where the default open() would fail to decode.""" + import json as _json + + path = registry_mod._marketplaces_path() + # Ensure parent dir exists. + registry_mod._ensure_file() + payload = { + "marketplaces": [ + {"name": "cafe-mkt", "owner": "o", "repo": "repo-\u4e2d\u6587"}, + ] + } + # Write raw UTF-8 (no \uXXXX escaping) to mimic what a non-Python + # tool or a future writer with ensure_ascii=False would produce. + with open(path, "w", encoding="utf-8") as f: + _json.dump(payload, f, ensure_ascii=False) + + registry_mod._invalidate_cache() + fetched = registry_mod.get_marketplace_by_name("cafe-mkt") + assert fetched is not None + assert fetched.repo == "repo-\u4e2d\u6587" diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py new file mode 100644 index 000000000..82db51f65 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,53 @@ +"""Tests for apm_cli.config module-level config file I/O. + +These tests exercise the round-trip of non-ASCII content through the global +config file to guard against the cp1252/cp950 UnicodeDecodeError class of +bugs on Windows when ``open()`` is called without an explicit encoding. +""" + +import json + +import pytest + +from apm_cli import config as config_mod + + +@pytest.fixture +def isolated_config(tmp_path, monkeypatch): + """Point CONFIG_DIR / CONFIG_FILE to a temp directory and clear cache.""" + config_dir = tmp_path / ".apm" + config_file = config_dir / "config.json" + monkeypatch.setattr(config_mod, "CONFIG_DIR", str(config_dir)) + monkeypatch.setattr(config_mod, "CONFIG_FILE", str(config_file)) + monkeypatch.setattr(config_mod, "_config_cache", None) + return config_file + + +class TestConfigUtf8RoundTrip: + """Round-trip non-ASCII content through the config file.""" + + def test_update_config_preserves_non_ascii(self, isolated_config): + non_ascii_value = "/Users/cafe/projets/\u958b\u59cb" + config_mod.update_config({"copilot_cowork_skills_dir": non_ascii_value}) + + # Force re-read from disk by invalidating the cache. + config_mod._invalidate_config_cache() + loaded = config_mod.get_config() + + assert loaded["copilot_cowork_skills_dir"] == non_ascii_value + + def test_config_file_is_utf8_on_disk(self, isolated_config): + non_ascii_value = "# \u958b\u59cb -- cafe" + config_mod.update_config({"note": non_ascii_value}) + + # Read raw bytes and decode as UTF-8 to assert the on-disk encoding. + raw = isolated_config.read_bytes() + decoded = json.loads(raw.decode("utf-8")) + assert decoded["note"] == non_ascii_value + + def test_ensure_config_exists_uses_utf8(self, isolated_config, monkeypatch): + # Force ensure_config_exists() to create the file. + config_mod.ensure_config_exists() + assert isolated_config.exists() + # File must be readable as UTF-8 JSON. + json.loads(isolated_config.read_bytes().decode("utf-8")) diff --git a/tests/unit/test_copilot_runtime.py b/tests/unit/test_copilot_runtime.py index cd9455cad..172786d68 100644 --- a/tests/unit/test_copilot_runtime.py +++ b/tests/unit/test_copilot_runtime.py @@ -154,3 +154,30 @@ def test_str_representation(self): str_repr = str(runtime) assert "CopilotRuntime" in str_repr assert "test-model" in str_repr + + +class TestMcpConfigUtf8RoundTrip: + """Reading MCP config preserves non-ASCII content (Windows cp1252/cp950 guard).""" + + def test_get_mcp_servers_reads_non_ascii(self, tmp_path): + import json as _json + + mcp_path = tmp_path / "mcp-config.json" + servers = { + "servers": { + "demo-cafe": { + "command": "node", + "args": ["server.js"], + "description": "\u4e2d\u6587 description -- cafe", + } + } + } + mcp_path.write_bytes(_json.dumps(servers).encode("utf-8")) + + with patch.object(CopilotRuntime, "is_available", return_value=True): + runtime = CopilotRuntime() + with patch.object(CopilotRuntime, "get_mcp_config_path", return_value=mcp_path): + got = runtime.get_mcp_servers() + + assert "demo-cafe" in got + assert got["demo-cafe"]["description"] == "\u4e2d\u6587 description -- cafe" diff --git a/tests/unit/test_plugin.py b/tests/unit/test_plugin.py new file mode 100644 index 000000000..a597e160a --- /dev/null +++ b/tests/unit/test_plugin.py @@ -0,0 +1,31 @@ +"""Tests for apm_cli.models.plugin module I/O. + +Round-trips non-ASCII content through Plugin.from_path to guard against +cp1252/cp950 UnicodeDecodeError on Windows when reading plugin.json. +""" + +import json + +from apm_cli.models.plugin import Plugin + + +class TestPluginUtf8RoundTrip: + """Round-trip non-ASCII content through plugin.json reads.""" + + def test_from_path_reads_non_ascii_metadata(self, tmp_path): + metadata = { + "id": "demo-plugin", + "name": "Demo plugin -- cafe", + "version": "1.0.0", + "description": "Plugin de demo with \u4e2d\u6587 description", + "author": "Cafe Author", + } + plugin_json = tmp_path / "plugin.json" + plugin_json.write_bytes(json.dumps(metadata).encode("utf-8")) + + plugin = Plugin.from_path(tmp_path) + + assert plugin.metadata.id == "demo-plugin" + assert plugin.metadata.name == "Demo plugin -- cafe" + assert plugin.metadata.description == "Plugin de demo with \u4e2d\u6587 description" + assert plugin.metadata.author == "Cafe Author"