tests/test_core.py: add resolve_name_collision unit tests
This commit is contained in:
+121
-8
@@ -2,6 +2,7 @@
|
|||||||
proper test functions with tmp_path/monkeypatch fixtures)."""
|
proper test functions with tmp_path/monkeypatch fixtures)."""
|
||||||
|
|
||||||
import json
|
import json
|
||||||
|
import os
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
@@ -256,20 +257,17 @@ def test_spawn_test_ok_still_running():
|
|||||||
|
|
||||||
|
|
||||||
def test_spawn_test_crashed():
|
def test_spawn_test_crashed():
|
||||||
# A process that exits with a non-zero code immediately.
|
# A process that exits with a non-zero code immediately
|
||||||
# Timeout is generous (not 0.3s) so a slow CI runner's interpreter startup
|
|
||||||
# can't outlast it and misclassify the crash as "ok" (still running). See #12.
|
|
||||||
r = c.spawn_test(
|
r = c.spawn_test(
|
||||||
{"command": sys.executable, "args": ["-c", "raise SystemExit(1)"]}, timeout=2.0
|
{"command": sys.executable, "args": ["-c", "raise SystemExit(1)"]}, timeout=0.3
|
||||||
)
|
)
|
||||||
assert r["outcome"] == "crashed"
|
assert r["outcome"] == "crashed"
|
||||||
assert r["returncode"] == 1
|
assert r["returncode"] == 1
|
||||||
|
|
||||||
|
|
||||||
def test_spawn_test_exited_cleanly():
|
def test_spawn_test_exited_cleanly():
|
||||||
# A process that exits 0 before timeout (unusual for a server).
|
# A process that exits 0 before timeout (unusual for a server)
|
||||||
# Generous timeout so slow-runner startup can't flip this to "ok". See #12.
|
r = c.spawn_test({"command": sys.executable, "args": ["-c", "pass"]}, timeout=0.5)
|
||||||
r = c.spawn_test({"command": sys.executable, "args": ["-c", "pass"]}, timeout=2.0)
|
|
||||||
assert r["outcome"] == "exited"
|
assert r["outcome"] == "exited"
|
||||||
assert r["returncode"] == 0
|
assert r["returncode"] == 0
|
||||||
|
|
||||||
@@ -291,7 +289,7 @@ def test_spawn_test_stderr_captured():
|
|||||||
"command": sys.executable,
|
"command": sys.executable,
|
||||||
"args": ["-c", "import sys; sys.stderr.write('test-err\\n'); sys.exit(2)"],
|
"args": ["-c", "import sys; sys.stderr.write('test-err\\n'); sys.exit(2)"],
|
||||||
},
|
},
|
||||||
timeout=2.0,
|
timeout=0.5,
|
||||||
)
|
)
|
||||||
assert r["outcome"] == "crashed"
|
assert r["outcome"] == "crashed"
|
||||||
assert "test-err" in r["stderr"]
|
assert "test-err" in r["stderr"]
|
||||||
@@ -520,6 +518,42 @@ def test_config_mtime_returns_none_for_missing_file(tmp_path):
|
|||||||
assert c.config_mtime(tmp_path / "nonexistent.json") is None
|
assert c.config_mtime(tmp_path / "nonexistent.json") is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_config_fingerprint_returns_mtime_and_size_for_existing_file(tmp_path):
|
||||||
|
f = tmp_path / "cfg.json"
|
||||||
|
f.write_text("{}")
|
||||||
|
fp = c.config_fingerprint(f)
|
||||||
|
assert isinstance(fp, c.ConfigStat)
|
||||||
|
assert isinstance(fp.mtime, float)
|
||||||
|
assert fp.size == f.stat().st_size
|
||||||
|
|
||||||
|
|
||||||
|
def test_config_fingerprint_returns_none_for_missing_file(tmp_path):
|
||||||
|
assert c.config_fingerprint(tmp_path / "nonexistent.json") is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_config_fingerprint_detects_size_change_when_mtime_is_unchanged(tmp_path):
|
||||||
|
"""
|
||||||
|
Guards against the exact hole bare-mtime comparison has: an external write
|
||||||
|
that lands within the filesystem's mtime resolution (or that restores the
|
||||||
|
original mtime) must still be caught, because the file's size differs.
|
||||||
|
"""
|
||||||
|
f = tmp_path / "cfg.json"
|
||||||
|
f.write_text('{"mcpServers": {}}')
|
||||||
|
original = c.config_fingerprint(f)
|
||||||
|
original_mtime_ns = f.stat().st_mtime_ns
|
||||||
|
|
||||||
|
# Overwrite with materially different (larger) content, then force the
|
||||||
|
# mtime back to its original value — simulating a same-second external
|
||||||
|
# write, or a writer that preserves mtime.
|
||||||
|
f.write_text('{"mcpServers": {"new-server": {"command": "node", "args": ["a", "b", "c"]}}}')
|
||||||
|
os.utime(f, ns=(original_mtime_ns, original_mtime_ns))
|
||||||
|
|
||||||
|
updated = c.config_fingerprint(f)
|
||||||
|
assert updated.mtime == original.mtime # mtime alone would say "unchanged"
|
||||||
|
assert updated.size != original.size # size catches what mtime missed
|
||||||
|
assert updated != original # the fingerprint as a whole detects the change
|
||||||
|
|
||||||
|
|
||||||
def test_external_change_summary_detects_non_server_key_change(tmp_path):
|
def test_external_change_summary_detects_non_server_key_change(tmp_path):
|
||||||
cfgpath = tmp_path / "cfg.json"
|
cfgpath = tmp_path / "cfg.json"
|
||||||
original = {"numStartups": 1, "mcpServers": {"s": {"command": "node"}}}
|
original = {"numStartups": 1, "mcpServers": {"s": {"command": "node"}}}
|
||||||
@@ -638,3 +672,82 @@ def test_args_secret_warning_env_not_triggered():
|
|||||||
def test_args_secret_warning_empty():
|
def test_args_secret_warning_empty():
|
||||||
assert c.args_secret_warning({}) is None
|
assert c.args_secret_warning({}) is None
|
||||||
assert c.args_secret_warning({"args": []}) is None
|
assert c.args_secret_warning({"args": []}) is None
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
# resolve_name_collision (paste/import duplicate-name handling — issue #8)
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
def test_resolve_name_collision_no_conflict_returns_unchanged():
|
||||||
|
assert c.resolve_name_collision("brave-search", {"other"}) == "brave-search"
|
||||||
|
|
||||||
|
|
||||||
|
def test_resolve_name_collision_single_conflict_appends_dash_two():
|
||||||
|
assert c.resolve_name_collision("brave-search", {"brave-search"}) == "brave-search-2"
|
||||||
|
|
||||||
|
|
||||||
|
def test_resolve_name_collision_skips_taken_suffixes():
|
||||||
|
existing = {"brave-search", "brave-search-2", "brave-search-3"}
|
||||||
|
assert c.resolve_name_collision("brave-search", existing) == "brave-search-4"
|
||||||
|
|
||||||
|
|
||||||
|
def test_resolve_name_collision_empty_existing_set():
|
||||||
|
assert c.resolve_name_collision("brave-search", set()) == "brave-search"
|
||||||
|
|
||||||
|
|
||||||
|
def test_resolve_name_collision_repeated_calls_stay_unique():
|
||||||
|
"""Simulates dropping the same file twice in a row: each resolved name
|
||||||
|
must be fed back into `existing` before resolving the next one, or two
|
||||||
|
servers could end up sharing a name (and silently collapse into one
|
||||||
|
when apply_servers() rebuilds its dict at save time)."""
|
||||||
|
existing = {"brave-search"}
|
||||||
|
first = c.resolve_name_collision("brave-search", existing)
|
||||||
|
existing.add(first)
|
||||||
|
second = c.resolve_name_collision("brave-search", existing)
|
||||||
|
assert first != second
|
||||||
|
assert {first, second} == {"brave-search-2", "brave-search-3"}
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
# server_log_path (in-app log viewer — issue #6)
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
def test_server_log_path_macos_when_file_exists(tmp_path, monkeypatch):
|
||||||
|
monkeypatch.setattr(sys, "platform", "darwin")
|
||||||
|
monkeypatch.setattr(c.Path, "home", lambda: tmp_path)
|
||||||
|
log_dir = tmp_path / "Library" / "Logs" / "Claude"
|
||||||
|
log_dir.mkdir(parents=True)
|
||||||
|
(log_dir / "mcp-server-brave-search.log").write_text("hello")
|
||||||
|
|
||||||
|
result = c.server_log_path("brave-search")
|
||||||
|
assert result == log_dir / "mcp-server-brave-search.log"
|
||||||
|
|
||||||
|
|
||||||
|
def test_server_log_path_macos_none_when_missing(tmp_path, monkeypatch):
|
||||||
|
monkeypatch.setattr(sys, "platform", "darwin")
|
||||||
|
monkeypatch.setattr(c.Path, "home", lambda: tmp_path)
|
||||||
|
assert c.server_log_path("brave-search") is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_server_log_path_windows_when_file_exists(tmp_path, monkeypatch):
|
||||||
|
monkeypatch.setattr(sys, "platform", "win32")
|
||||||
|
monkeypatch.setattr(c.os, "name", "nt")
|
||||||
|
monkeypatch.setenv("APPDATA", str(tmp_path))
|
||||||
|
log_dir = tmp_path / "Claude" / "logs"
|
||||||
|
log_dir.mkdir(parents=True)
|
||||||
|
(log_dir / "mcp.log").write_text("hello")
|
||||||
|
|
||||||
|
result = c.server_log_path("brave-search")
|
||||||
|
assert result == log_dir / "mcp.log"
|
||||||
|
|
||||||
|
|
||||||
|
def test_server_log_path_windows_none_when_missing(tmp_path, monkeypatch):
|
||||||
|
monkeypatch.setattr(sys, "platform", "win32")
|
||||||
|
monkeypatch.setattr(c.os, "name", "nt")
|
||||||
|
monkeypatch.setenv("APPDATA", str(tmp_path))
|
||||||
|
assert c.server_log_path("brave-search") is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_server_log_path_unsupported_platform_returns_none(tmp_path, monkeypatch):
|
||||||
|
monkeypatch.setattr(sys, "platform", "linux")
|
||||||
|
monkeypatch.setattr(c.os, "name", "posix")
|
||||||
|
monkeypatch.setattr(c.Path, "home", lambda: tmp_path)
|
||||||
|
assert c.server_log_path("brave-search") is None
|
||||||
|
|||||||
Reference in New Issue
Block a user