diff --git a/CHANGELOG.md b/CHANGELOG.md index 7b96b22..444aa42 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ ### Added +- **Test coverage Phase 2 of #14** (#17) — total coverage 85% → 93%. `cli/version.py` 27% → 100% and `cli/setup.py` 63% → 100%, the two largest gaps remaining after Phase 1. Test count 392 → 457 (+65 cases) across two new test files: `tests/core/test_cli_version.py` (40 cases covering `_get_current_version`/`_get_latest_pypi_version`/`_version_tuple`/`_detect_installer`/`_load_global_state`/`_save_global_state`/`_check_for_update`/`_do_auto_upgrade`/`_do_revert`, with `urlopen` and `subprocess.run` mocked at the boundary), and `tests/core/test_cli_setup.py` (25 cases covering the async helpers `_attempt_login`/`_connect_and_login`/`_setup_login` including the 2FA bootstrap path with device-token storage, plus `_setup_credential_flow` error paths, the `setup` command's discovered-config valid-and-invalid branches, the `_setup_interactive` validation-failure exit, and the `_emit_claude_desktop_snippet` Linux DBUS fallback). No production code touched. - **Test coverage Phase 1 of #14** (#16) — total coverage 81% → 85%. Five files brought to 100%: `cli/check.py` (51%), `cli/main.py` (56%), `cli/logging_.py` (78%), `modules/system/__init__.py` (23%), `modules/filestation/__init__.py` (70%). Test count 336 → 392 (+56 cases). New test classes in `tests/core/test_cli.py` cover the `_check_login` async path, every top-level option in the `main` group (`--check-update`, `--auto-upgrade`, `--revert`, version-change tracking, auto-upgrade trigger), and the early/configured logging setup. Two new test files (`tests/modules/{system,filestation}/test_register.py`) exercise module registration closure bodies via `server._tool_manager._tools[name].fn` extraction with sentinel `AsyncMock` return values, walking the tool body lines that the prior `assert server is not None` style left uncovered. No production code touched. - **`CLAUDE.md` documents the per-PR CHANGELOG convention** (#16) — adds an "Adding a CHANGELOG entry on every PR" section under "Common Tasks" specifying that every PR updates `## Unreleased` in `CHANGELOG.md` using strict Keep a Changelog categories (`### Added`, `### Changed`, `### Fixed`). Updates the "Bumping the version for a release" steps to rename `## Unreleased` to `## ()` and add a fresh empty `## Unreleased` section, plus notes that the `publish.yml` `github-release` awk extractor (`## ( |\()`) walks past `## Unreleased` harmlessly during tag-push releases. diff --git a/tests/core/test_cli_setup.py b/tests/core/test_cli_setup.py new file mode 100644 index 0000000..daf3a31 --- /dev/null +++ b/tests/core/test_cli_setup.py @@ -0,0 +1,492 @@ +"""Tests for cli/setup.py — interactive setup, credential flow, 2FA bootstrap. + +cli/setup.py was at 63% (86/233 missing) after PR #16. The remaining gaps +are concentrated in three async helpers — `_attempt_login`, `_connect_and_login`, +`_setup_login` — plus a handful of small error paths in the synchronous +flow (`_setup_credential_flow`, `_setup_with_config`, the load_config +validation-error branch in the `setup` command itself). + +Strategy: +- Async helpers tested directly with `AsyncMock` clients (no respx; the + client's `request()` method is mocked at the boundary) +- Sync flows tested via `CliRunner` with `input=` and patched module + globals (`_CONFIG_DIR`, `_store_keyring`, etc.) — same pattern as the + pre-existing `TestSetupInteractive` class in test_cli.py +- The `_emit_claude_desktop_snippet` Linux fallback (when + `DBUS_SESSION_BUS_ADDRESS` is unset) is tested by clearing the env var + and confirming the constructed `unix:path=/run/user/{uid}/bus` value + appears in the snippet + +Coverage target per #14: cli/setup.py 63% → 90%+. +""" + +from __future__ import annotations + +import os +from typing import TYPE_CHECKING, Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from click.testing import CliRunner + +from mcp_synology.cli import setup as setup_mod +from mcp_synology.cli.main import main + +if TYPE_CHECKING: + from pathlib import Path + + +# Helper used across multiple test classes +def _make_test_app_config(host: str = "1.2.3.4", https: bool = False) -> Any: + """Build an AppConfig pointing at the given host/protocol.""" + from mcp_synology.core.config import AppConfig + + raw: dict[str, Any] = { + "schema_version": 1, + "instance_id": host.replace(".", "-"), + "connection": {"host": host, "https": https}, + "modules": {"filestation": {"enabled": True}}, + } + return AppConfig(**raw) + + +# ---------- _attempt_login ---------- + + +class TestAttemptLogin: + """Direct tests of the async _attempt_login helper. + + The helper takes a `client` object and calls `client.request("SYNO.API.Auth", "login", ...)`. + All paths can be exercised by passing a MagicMock with an AsyncMock `request`. + """ + + @staticmethod + def _make_client(request_mock: AsyncMock) -> MagicMock: + client = MagicMock() + client.request = request_mock + client.sid = None + return client + + async def test_login_success_no_2fa(self) -> None: + request = AsyncMock(return_value={"sid": "test-sid-123"}) + client = self._make_client(request) + result = await setup_mod._attempt_login(client, "admin", "pw", "service") + assert result["success"] is True + assert result["sid"] == "test-sid-123" + assert client.sid == "test-sid-123" + + async def test_login_failure_non_2fa_error(self) -> None: + from mcp_synology.core.errors import SynologyError + + request = AsyncMock(side_effect=SynologyError("bad password", code=400)) + client = self._make_client(request) + result = await setup_mod._attempt_login(client, "admin", "pw", "service") + assert result["success"] is False + + async def test_login_2fa_required_then_success_with_device_token(self) -> None: + from mcp_synology.core.errors import SynologyError + + # First call: 403 (2FA required). Second call: success with did + request = AsyncMock( + side_effect=[ + SynologyError("2fa", code=403), + {"sid": "sid-after-2fa", "did": "device-token-xyz"}, + ] + ) + client = self._make_client(request) + + with patch("keyring.set_password") as set_pw: + runner = CliRunner() + # Drive the OTP prompt — _attempt_login uses click.prompt internally + with runner.isolation(input="123456\n"): + result = await setup_mod._attempt_login(client, "admin", "pw", "service") + + assert result["success"] is True + assert result["sid"] == "sid-after-2fa" + # Device token stored under the service + set_pw.assert_called_once_with("service", "device_id", "device-token-xyz") + # Both requests issued + assert request.await_count == 2 + + async def test_login_2fa_required_then_success_without_device_token(self) -> None: + """2FA succeeds but DSM doesn't echo back a `did` — still succeeds.""" + from mcp_synology.core.errors import SynologyError + + request = AsyncMock( + side_effect=[ + SynologyError("2fa", code=403), + {"sid": "sid-after-2fa"}, # no `did` field + ] + ) + client = self._make_client(request) + + runner = CliRunner() + with runner.isolation(input="123456\n"): + result = await setup_mod._attempt_login(client, "admin", "pw", "service") + + assert result["success"] is True + assert result["sid"] == "sid-after-2fa" + + async def test_login_2fa_then_failure(self) -> None: + from mcp_synology.core.errors import SynologyError + + request = AsyncMock( + side_effect=[ + SynologyError("2fa", code=403), + SynologyError("bad otp", code=404), + ] + ) + client = self._make_client(request) + + runner = CliRunner() + with runner.isolation(input="000000\n"): + result = await setup_mod._attempt_login(client, "admin", "pw", "service") + assert result["success"] is False + + async def test_login_oserror(self) -> None: + request = AsyncMock(side_effect=OSError("connection refused")) + client = self._make_client(request) + result = await setup_mod._attempt_login(client, "admin", "pw", "service") + assert result["success"] is False + + +# ---------- _connect_and_login ---------- + + +class TestConnectAndLogin: + @staticmethod + def _make_client_cm(request_mock: AsyncMock, dsm_info: dict[str, str]) -> MagicMock: + """Construct a fake DsmClient suitable for `async with DsmClient(...) as client`.""" + client = MagicMock() + client.request = request_mock + client.sid = None + client.query_api_info = AsyncMock(return_value=None) + client.fetch_dsm_info = AsyncMock(return_value=dsm_info) + client.__aenter__ = AsyncMock(return_value=client) + client.__aexit__ = AsyncMock(return_value=None) + return client + + async def test_connect_and_login_success_with_hostname(self) -> None: + config = _make_test_app_config() + request = AsyncMock( + side_effect=[ + {"sid": "test-sid"}, # login + {}, # logout + ] + ) + client = self._make_client_cm(request, {"hostname": "MyNAS", "version_string": "DSM 7.1"}) + with patch("mcp_synology.core.client.DsmClient", return_value=client): + result = await setup_mod._connect_and_login(config, "admin", "pw", "svc", verbose=False) + assert result["success"] is True + assert result["hostname"] == "MyNAS" + assert result["dsm_version"] == "DSM 7.1" + + async def test_connect_and_login_success_no_hostname(self) -> None: + config = _make_test_app_config() + request = AsyncMock(side_effect=[{"sid": "test-sid"}, {}]) + client = self._make_client_cm(request, {}) # empty dsm_info + with patch("mcp_synology.core.client.DsmClient", return_value=client): + result = await setup_mod._connect_and_login(config, "admin", "pw", "svc", verbose=False) + assert result["success"] is True + assert "hostname" not in result + + async def test_connect_and_login_https(self) -> None: + config = _make_test_app_config(host="nas.example.com", https=True) + request = AsyncMock(side_effect=[{"sid": "test-sid"}, {}]) + client = self._make_client_cm(request, {}) + with patch("mcp_synology.core.client.DsmClient", return_value=client) as dsm_client: + await setup_mod._connect_and_login(config, "admin", "pw", "svc", verbose=False) + # Confirm https URL was constructed + kwargs = dsm_client.call_args.kwargs + assert kwargs["base_url"].startswith("https://") + + async def test_connect_and_login_login_failure(self) -> None: + from mcp_synology.core.errors import SynologyError + + config = _make_test_app_config() + request = AsyncMock(side_effect=SynologyError("bad creds", code=400)) + client = self._make_client_cm(request, {}) + with patch("mcp_synology.core.client.DsmClient", return_value=client): + result = await setup_mod._connect_and_login(config, "admin", "pw", "svc", verbose=False) + assert result["success"] is False + + async def test_connect_and_login_rejects_non_appconfig(self) -> None: + with pytest.raises(RuntimeError, match="AppConfig"): + await setup_mod._connect_and_login("not a config", "u", "p", "s", verbose=False) + + async def test_connect_and_login_rejects_missing_connection(self) -> None: + config = _make_test_app_config() + config.connection = None # type: ignore[assignment] + with pytest.raises(RuntimeError, match="connection"): + await setup_mod._connect_and_login(config, "u", "p", "s", verbose=False) + + +# ---------- _setup_login ---------- + + +class TestSetupLogin: + @staticmethod + def _make_client_cm(request_mock: AsyncMock) -> MagicMock: + client = MagicMock() + client.request = request_mock + client.sid = None + client.query_api_info = AsyncMock(return_value=None) + client.__aenter__ = AsyncMock(return_value=client) + client.__aexit__ = AsyncMock(return_value=None) + return client + + async def test_setup_login_success_logs_out(self) -> None: + config = _make_test_app_config() + request = AsyncMock(side_effect=[{"sid": "test-sid"}, {}]) + client = self._make_client_cm(request) + with patch("mcp_synology.core.client.DsmClient", return_value=client): + await setup_mod._setup_login(config, "admin", "pw", "svc") + # Login + logout: 2 requests + assert request.await_count == 2 + + async def test_setup_login_failure_skips_logout(self) -> None: + from mcp_synology.core.errors import SynologyError + + config = _make_test_app_config() + request = AsyncMock(side_effect=SynologyError("nope", code=400)) + client = self._make_client_cm(request) + with patch("mcp_synology.core.client.DsmClient", return_value=client): + await setup_mod._setup_login(config, "admin", "pw", "svc") + # Only the login attempt was made; logout skipped because login failed + assert request.await_count == 1 + + async def test_setup_login_rejects_non_appconfig(self) -> None: + with pytest.raises(RuntimeError, match="AppConfig"): + await setup_mod._setup_login("not a config", "u", "p", "s") + + async def test_setup_login_rejects_missing_connection(self) -> None: + config = _make_test_app_config() + config.connection = None # type: ignore[assignment] + with pytest.raises(RuntimeError, match="connection"): + await setup_mod._setup_login(config, "u", "p", "s") + + +# ---------- _setup_credential_flow ---------- + + +class TestSetupCredentialFlow: + def test_rejects_non_appconfig(self) -> None: + with pytest.raises(RuntimeError, match="AppConfig"): + setup_mod._setup_credential_flow("not a config", verbose=False) + + def test_rejects_missing_connection(self) -> None: + config = _make_test_app_config() + config.connection = None # type: ignore[assignment] + with pytest.raises(RuntimeError, match="connection"): + setup_mod._setup_credential_flow(config, verbose=False) + + def test_keyring_failure_returns_early(self, tmp_path: Path) -> None: + """Keyring failure short-circuits before the asyncio.run(_setup_login) call.""" + config_file = tmp_path / "config.yaml" + config_file.write_text( + "schema_version: 1\n" + "instance_id: test-nas\n" + "connection:\n" + " host: 192.168.1.100\n" + "modules:\n" + " filestation:\n" + " enabled: true\n" + ) + runner = CliRunner() + with ( + patch("mcp_synology.cli.setup._store_keyring", return_value=False), + patch("mcp_synology.cli.setup.asyncio.run") as run_mock, + ): + result = runner.invoke( + main, + ["setup", "-c", str(config_file)], + input="admin\nsecret\n", + ) + assert result.exit_code == 0 + # Login should NOT have been attempted because keyring failed + run_mock.assert_not_called() + + +# ---------- setup command top-level error paths ---------- + + +class TestSetupCommandErrorPaths: + def test_setup_with_invalid_config_path_exits_nonzero(self, tmp_path: Path) -> None: + """`setup -c ` triggers _setup_with_config's load_config exception path.""" + config_file = tmp_path / "broken.yaml" + config_file.write_text( + "schema_version: 999\n" + "connection:\n" + " host: 1.2.3.4\n" + "modules:\n" + " filestation:\n" + " enabled: true\n" + ) + runner = CliRunner() + result = runner.invoke(main, ["setup", "-c", str(config_file)]) + assert result.exit_code == 1 + assert "Error" in result.output + + def test_setup_with_discovered_valid_config_runs_credential_flow(self, tmp_path: Path) -> None: + """`setup` (no -c) → discover_config_path returns a valid config → credential flow.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + good = config_dir / "default.yaml" + good.write_text( + "schema_version: 1\n" + "instance_id: test-nas\n" + "connection:\n" + " host: 1.2.3.4\n" + "modules:\n" + " filestation:\n" + " enabled: true\n" + ) + + clean_env: dict[str, str] = { + k: val for k, val in os.environ.items() if not k.startswith("SYNOLOGY_") + } + + runner = CliRunner() + with ( + patch("mcp_synology.core.config.discover_config_path", return_value=good), + patch("mcp_synology.cli.setup._store_keyring", return_value=True), + patch("mcp_synology.cli.setup.asyncio.run", return_value=None), + patch.dict(os.environ, clean_env, clear=True), + ): + result = runner.invoke(main, ["setup"], input="admin\nsecret\n") + assert result.exit_code == 0 + assert "Setting up credentials" in result.output + + def test_setup_with_validation_error_in_discovered_config(self, tmp_path: Path) -> None: + """`setup` (no -c) → load_config(None) raises ValueError → exits 1.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + # Place a config that discover_config_path will find but fails validation + bad = config_dir / "default.yaml" + bad.write_text( + "schema_version: 999\n" + "connection:\n" + " host: 1.2.3.4\n" + "modules:\n" + " filestation:\n" + " enabled: true\n" + ) + + clean_env: dict[str, str] = { + k: val for k, val in os.environ.items() if not k.startswith("SYNOLOGY_") + } + + runner = CliRunner() + with ( + patch( + "mcp_synology.core.config.discover_config_path", + return_value=bad, + ), + patch.dict(os.environ, clean_env, clear=True), + ): + result = runner.invoke(main, ["setup"]) + assert result.exit_code == 1 + assert "Error" in result.output + + +# ---------- _setup_interactive validation failure ---------- + + +class TestSetupInteractiveValidationFailure: + def test_validation_failure_during_interactive_exits_nonzero(self, tmp_path: Path) -> None: + """If AppConfig(**config_dict) raises after the user fills out prompts, exit 1. + + Triggered by patching `_derive_instance_id` to return an invalid id (with + characters AppConfig rejects), which fails validation at the first + AppConfig(**config_dict) construction call inside _setup_interactive. + """ + config_dir = tmp_path / "config" + + clean_env: dict[str, str] = { + k: val for k, val in os.environ.items() if not k.startswith("SYNOLOGY_") + } + + runner = CliRunner() + with ( + patch("mcp_synology.cli.setup._CONFIG_DIR", config_dir), + patch("mcp_synology.core.config.discover_config_path", side_effect=FileNotFoundError), + patch( + "mcp_synology.core.config._derive_instance_id", + return_value="INVALID ID WITH SPACES!", + ), + patch.dict(os.environ, clean_env, clear=True), + ): + # host, https, permission, alias — fewer prompts since validation fails + result = runner.invoke( + main, + ["setup"], + input="192.168.1.50\nn\nread\n\n", + ) + assert result.exit_code == 1 + assert "Config validation failed" in result.output + + +# ---------- _emit_claude_desktop_snippet Linux fallback ---------- + + +class TestEmitClaudeDesktopSnippetLinuxFallback: + def test_linux_constructs_dbus_path_when_env_unset(self, tmp_path: Path) -> None: + """When DBUS_SESSION_BUS_ADDRESS isn't set, the Linux branch constructs + /run/user/{uid}/bus and includes it in the snippet.""" + config_dir = tmp_path / "config" + + clean_env: dict[str, str] = { + k: val + for k, val in os.environ.items() + if not k.startswith("SYNOLOGY_") and k != "DBUS_SESSION_BUS_ADDRESS" + } + + connect_result: dict[str, Any] = {"success": True} + + runner = CliRunner() + with ( + patch("mcp_synology.cli.setup._CONFIG_DIR", config_dir), + patch("mcp_synology.core.config.discover_config_path", side_effect=FileNotFoundError), + patch("mcp_synology.cli.setup._store_keyring", return_value=True), + patch("mcp_synology.cli.setup.asyncio.run", return_value=connect_result), + patch.dict(os.environ, clean_env, clear=True), + patch("sys.platform", "linux"), + patch("os.getuid", return_value=1234, create=True), + ): + result = runner.invoke( + main, + ["setup"], + input="192.168.1.50\nn\nread\n\nadmin\npassword\n", + ) + + assert result.exit_code == 0 + assert "/run/user/1234/bus" in result.output + + def test_linux_uses_dbus_env_when_set(self, tmp_path: Path) -> None: + """When DBUS_SESSION_BUS_ADDRESS is set, the snippet uses that value.""" + config_dir = tmp_path / "config" + + clean_env: dict[str, str] = { + k: val for k, val in os.environ.items() if not k.startswith("SYNOLOGY_") + } + clean_env["DBUS_SESSION_BUS_ADDRESS"] = "unix:path=/custom/bus" + + connect_result: dict[str, Any] = {"success": True} + + runner = CliRunner() + with ( + patch("mcp_synology.cli.setup._CONFIG_DIR", config_dir), + patch("mcp_synology.core.config.discover_config_path", side_effect=FileNotFoundError), + patch("mcp_synology.cli.setup._store_keyring", return_value=True), + patch("mcp_synology.cli.setup.asyncio.run", return_value=connect_result), + patch.dict(os.environ, clean_env, clear=True), + patch("sys.platform", "linux"), + ): + result = runner.invoke( + main, + ["setup"], + input="192.168.1.50\nn\nread\n\nadmin\npassword\n", + ) + + assert result.exit_code == 0 + assert "/custom/bus" in result.output diff --git a/tests/core/test_cli_version.py b/tests/core/test_cli_version.py new file mode 100644 index 0000000..d385972 --- /dev/null +++ b/tests/core/test_cli_version.py @@ -0,0 +1,426 @@ +"""Tests for cli/version.py — version checking, PyPI queries, auto-upgrade, revert. + +This file targets the largest single coverage gap remaining after PR #16: +cli/version.py was at 27% (93/127 statements missing). The module has no +direct unit tests because it talks to PyPI via stdlib `urllib.request` +(not httpx), reads/writes a YAML file under `~/.local/state`, and shells +out to `uv` or `pipx` for upgrade and revert. Each of those external +boundaries is mockable with `monkeypatch` + `unittest.mock.patch`. + +Coverage target per #14: cli/version.py 27% → 90%+. +""" + +from __future__ import annotations + +import json +import subprocess +from typing import TYPE_CHECKING, Any +from unittest.mock import MagicMock, patch + +import yaml + +from mcp_synology.cli import version as v + +if TYPE_CHECKING: + from pathlib import Path + + +# ---------- _get_current_version ---------- + + +class TestGetCurrentVersion: + def test_returns_importlib_version_when_package_installed(self) -> None: + with patch("importlib.metadata.version", return_value="9.9.9"): + assert v._get_current_version() == "9.9.9" + + def test_falls_back_to_module_version_on_package_not_found(self) -> None: + from importlib.metadata import PackageNotFoundError + + with patch("importlib.metadata.version", side_effect=PackageNotFoundError("nope")): + result = v._get_current_version() + # Falls back to mcp_synology.__version__ + from mcp_synology import __version__ + + assert result == __version__ + + +# ---------- _get_latest_pypi_version ---------- + + +class TestGetLatestPypiVersion: + @staticmethod + def _fake_response(payload: dict[str, Any]) -> MagicMock: + body = json.dumps(payload).encode("utf-8") + resp = MagicMock() + resp.read.return_value = body + resp.__enter__.return_value = resp + resp.__exit__.return_value = None + return resp + + def test_returns_version_on_success(self) -> None: + resp = self._fake_response({"info": {"version": "1.2.3"}}) + with patch("mcp_synology.cli.version.urlopen", return_value=resp): + assert v._get_latest_pypi_version() == "1.2.3" + + def test_returns_none_on_oserror(self) -> None: + with patch("mcp_synology.cli.version.urlopen", side_effect=OSError("network down")): + assert v._get_latest_pypi_version() is None + + def test_returns_none_on_keyerror_in_response(self) -> None: + resp = self._fake_response({"unexpected": "shape"}) + with patch("mcp_synology.cli.version.urlopen", return_value=resp): + assert v._get_latest_pypi_version() is None + + def test_returns_none_on_invalid_json(self) -> None: + resp = MagicMock() + resp.read.return_value = b"not json at all" + resp.__enter__.return_value = resp + resp.__exit__.return_value = None + with patch("mcp_synology.cli.version.urlopen", return_value=resp): + assert v._get_latest_pypi_version() is None + + +# ---------- _version_tuple ---------- + + +class TestVersionTuple: + def test_simple(self) -> None: + assert v._version_tuple("1.2.3") == (1, 2, 3) + + def test_two_part(self) -> None: + assert v._version_tuple("0.5") == (0, 5) + + def test_invalid_returns_zero_tuple(self) -> None: + assert v._version_tuple("not.a.version") == (0,) + + def test_empty_returns_zero_tuple(self) -> None: + # int("") raises ValueError + assert v._version_tuple("") == (0,) + + +# ---------- _detect_installer ---------- + + +class TestDetectInstaller: + def test_uv_path(self) -> None: + with patch("shutil.which", return_value="/home/me/.local/share/uv/tools/mcp-synology"): + assert v._detect_installer() == "uv" + + def test_pipx_path(self) -> None: + with patch("shutil.which", return_value="/home/me/.local/pipx/venvs/mcp-synology/bin/x"): + assert v._detect_installer() == "pipx" + + def test_unknown_path(self) -> None: + with patch("shutil.which", return_value="/usr/local/bin/mcp-synology"): + assert v._detect_installer() is None + + def test_not_on_path(self) -> None: + with patch("shutil.which", return_value=None): + assert v._detect_installer() is None + + +# ---------- _load_global_state / _save_global_state ---------- + + +class TestGlobalState: + def test_load_returns_empty_when_file_missing(self, tmp_path: Path) -> None: + with patch("pathlib.Path.home", return_value=tmp_path): + assert v._load_global_state() == {} + + def test_load_returns_yaml_contents(self, tmp_path: Path) -> None: + state_dir = tmp_path / ".local" / "state" / "mcp-synology" + state_dir.mkdir(parents=True) + (state_dir / "global.yaml").write_text("auto_upgrade: true\nrunning_version: 0.5.0\n") + with patch("pathlib.Path.home", return_value=tmp_path): + assert v._load_global_state() == { + "auto_upgrade": True, + "running_version": "0.5.0", + } + + def test_load_returns_empty_on_yaml_error(self, tmp_path: Path) -> None: + state_dir = tmp_path / ".local" / "state" / "mcp-synology" + state_dir.mkdir(parents=True) + (state_dir / "global.yaml").write_text("not: valid: yaml: at: all") + with patch("pathlib.Path.home", return_value=tmp_path): + assert v._load_global_state() == {} + + def test_load_returns_empty_when_yaml_is_null(self, tmp_path: Path) -> None: + """Empty YAML file (yaml.safe_load returns None) → {}.""" + state_dir = tmp_path / ".local" / "state" / "mcp-synology" + state_dir.mkdir(parents=True) + (state_dir / "global.yaml").write_text("") + with patch("pathlib.Path.home", return_value=tmp_path): + assert v._load_global_state() == {} + + def test_save_creates_directory_and_file(self, tmp_path: Path) -> None: + with patch("pathlib.Path.home", return_value=tmp_path): + v._save_global_state({"running_version": "1.0.0", "auto_upgrade": True}) + + state_file = tmp_path / ".local" / "state" / "mcp-synology" / "global.yaml" + assert state_file.exists() + loaded = yaml.safe_load(state_file.read_text()) + assert loaded == {"running_version": "1.0.0", "auto_upgrade": True} + assert state_file.read_text().startswith("# Auto-generated by mcp-synology.\n") + + def test_save_then_load_round_trips(self, tmp_path: Path) -> None: + original = {"latest_known_version": "0.6.0", "previous_version": "0.5.0"} + with patch("pathlib.Path.home", return_value=tmp_path): + v._save_global_state(original) + assert v._load_global_state() == original + + +# ---------- _check_for_update ---------- + + +class TestCheckForUpdate: + def test_force_bypasses_cache_and_fetches(self) -> None: + state: dict[str, Any] = { + "last_version_check": "2200-01-01T00:00:00+00:00", # far in the future + "latest_known_version": "0.0.1", + } + with ( + patch("mcp_synology.cli.version._get_latest_pypi_version", return_value="9.9.9"), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + ): + assert v._check_for_update(state, force=True) == "9.9.9" + assert state["latest_known_version"] == "9.9.9" + + def test_cache_valid_and_newer_returns_cached(self) -> None: + from datetime import UTC, datetime + + state: dict[str, Any] = { + "last_version_check": datetime.now(tz=UTC).isoformat(), + "latest_known_version": "9.9.9", + } + with ( + patch("mcp_synology.cli.version._get_latest_pypi_version") as fetch, + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + ): + assert v._check_for_update(state) == "9.9.9" + fetch.assert_not_called() # cache hit, no network call + + def test_cache_valid_and_not_newer_returns_none(self) -> None: + from datetime import UTC, datetime + + state: dict[str, Any] = { + "last_version_check": datetime.now(tz=UTC).isoformat(), + "latest_known_version": "0.5.0", + } + with ( + patch("mcp_synology.cli.version._get_latest_pypi_version") as fetch, + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + ): + assert v._check_for_update(state) is None + fetch.assert_not_called() + + def test_cache_stale_triggers_fetch(self) -> None: + state: dict[str, Any] = { + "last_version_check": "2000-01-01T00:00:00+00:00", # very old + "latest_known_version": "0.0.1", + } + with ( + patch("mcp_synology.cli.version._get_latest_pypi_version", return_value="0.6.0"), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + ): + assert v._check_for_update(state) == "0.6.0" + + def test_cache_corrupt_timestamp_falls_through_to_fetch(self) -> None: + state: dict[str, Any] = { + "last_version_check": "not a timestamp", + "latest_known_version": "0.0.1", + } + with ( + patch("mcp_synology.cli.version._get_latest_pypi_version", return_value="0.6.0"), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + ): + assert v._check_for_update(state) == "0.6.0" + + def test_pypi_returns_none_returns_none(self) -> None: + state: dict[str, Any] = {} + with ( + patch("mcp_synology.cli.version._get_latest_pypi_version", return_value=None), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + ): + assert v._check_for_update(state) is None + # Even on failure the timestamp gets updated (so we don't hammer PyPI) + assert "last_version_check" in state + + def test_pypi_returns_same_version(self) -> None: + state: dict[str, Any] = {} + with ( + patch("mcp_synology.cli.version._get_latest_pypi_version", return_value="0.5.0"), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + ): + assert v._check_for_update(state) is None + assert state["latest_known_version"] == "0.5.0" + + def test_pypi_returns_newer_version(self) -> None: + state: dict[str, Any] = {} + with ( + patch("mcp_synology.cli.version._get_latest_pypi_version", return_value="9.9.9"), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + ): + assert v._check_for_update(state) == "9.9.9" + + +# ---------- _do_auto_upgrade ---------- + + +class TestDoAutoUpgrade: + @staticmethod + def _fake_completed(returncode: int, stderr: str = "") -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess( + args=["fake"], returncode=returncode, stdout="", stderr=stderr + ) + + def test_uv_installer_success(self, tmp_path: Path) -> None: + state: dict[str, Any] = {} + with ( + patch("pathlib.Path.home", return_value=tmp_path), + patch("mcp_synology.cli.version._detect_installer", return_value="uv"), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + patch("subprocess.run", return_value=self._fake_completed(0)) as run, + ): + assert v._do_auto_upgrade(state) is True + cmd = run.call_args.args[0] + assert cmd[:3] == ["uv", "tool", "install"] + assert "mcp-synology@latest" in cmd + assert state["previous_version"] == "0.5.0" + + def test_pipx_installer_success(self, tmp_path: Path) -> None: + state: dict[str, Any] = {} + with ( + patch("pathlib.Path.home", return_value=tmp_path), + patch("mcp_synology.cli.version._detect_installer", return_value="pipx"), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + patch("subprocess.run", return_value=self._fake_completed(0)), + ): + assert v._do_auto_upgrade(state) is True + assert state["previous_version"] == "0.5.0" + + def test_unknown_installer_returns_false(self) -> None: + state: dict[str, Any] = {} + with ( + patch("mcp_synology.cli.version._detect_installer", return_value=None), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + ): + assert v._do_auto_upgrade(state) is False + assert "previous_version" not in state + + def test_subprocess_failure_returns_false(self, tmp_path: Path) -> None: + state: dict[str, Any] = {} + with ( + patch("pathlib.Path.home", return_value=tmp_path), + patch("mcp_synology.cli.version._detect_installer", return_value="uv"), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + patch("subprocess.run", return_value=self._fake_completed(1, stderr="boom")), + ): + assert v._do_auto_upgrade(state) is False + assert "previous_version" not in state + + +# ---------- _do_revert ---------- + + +class TestDoRevert: + @staticmethod + def _fake_completed(returncode: int, stderr: str = "") -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess( + args=["fake"], returncode=returncode, stdout="", stderr=stderr + ) + + def test_revert_with_no_state_and_no_explicit(self, tmp_path: Path) -> None: + """No previous_version recorded and no explicit version → message + return.""" + with ( + patch("pathlib.Path.home", return_value=tmp_path), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + patch("subprocess.run") as run, + ): + v._do_revert(None) + run.assert_not_called() + + def test_revert_with_target_version_uv_installer(self, tmp_path: Path) -> None: + with ( + patch("pathlib.Path.home", return_value=tmp_path), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + patch("mcp_synology.cli.version._detect_installer", return_value="uv"), + patch("subprocess.run", return_value=self._fake_completed(0)) as run, + ): + v._do_revert("0.4.1") + cmd = run.call_args.args[0] + assert cmd[:4] == ["uv", "tool", "install", "--force"] + assert any("==0.4.1" in arg for arg in cmd) + + def test_revert_with_target_version_pipx_installer(self, tmp_path: Path) -> None: + with ( + patch("pathlib.Path.home", return_value=tmp_path), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + patch("mcp_synology.cli.version._detect_installer", return_value="pipx"), + patch("subprocess.run", return_value=self._fake_completed(0)) as run, + ): + v._do_revert("0.4.1") + cmd = run.call_args.args[0] + assert cmd[:3] == ["pipx", "install", "--force"] + assert any("==0.4.1" in arg for arg in cmd) + + def test_revert_with_unknown_installer(self, tmp_path: Path) -> None: + with ( + patch("pathlib.Path.home", return_value=tmp_path), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + patch("mcp_synology.cli.version._detect_installer", return_value=None), + patch("subprocess.run") as run, + ): + v._do_revert("0.4.1") + run.assert_not_called() + + def test_revert_uses_previous_version_from_state(self, tmp_path: Path) -> None: + # Pre-populate state with a previous_version + state_dir = tmp_path / ".local" / "state" / "mcp-synology" + state_dir.mkdir(parents=True) + (state_dir / "global.yaml").write_text("previous_version: 0.4.1\n") + + with ( + patch("pathlib.Path.home", return_value=tmp_path), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + patch("mcp_synology.cli.version._detect_installer", return_value="uv"), + patch("subprocess.run", return_value=self._fake_completed(0)) as run, + ): + v._do_revert(None) + cmd = run.call_args.args[0] + assert any("==0.4.1" in arg for arg in cmd) + + def test_revert_to_current_version_is_noop(self, tmp_path: Path) -> None: + with ( + patch("pathlib.Path.home", return_value=tmp_path), + patch("mcp_synology.cli.version._get_current_version", return_value="0.4.1"), + patch("subprocess.run") as run, + ): + v._do_revert("0.4.1") + run.assert_not_called() + + def test_revert_subprocess_failure(self, tmp_path: Path) -> None: + with ( + patch("pathlib.Path.home", return_value=tmp_path), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + patch("mcp_synology.cli.version._detect_installer", return_value="uv"), + patch("subprocess.run", return_value=self._fake_completed(1, stderr="bad")), + ): + # Just verify it doesn't crash; the function returns None either way + v._do_revert("0.4.1") + + def test_revert_success_clears_previous_and_disables_auto_upgrade(self, tmp_path: Path) -> None: + state_dir = tmp_path / ".local" / "state" / "mcp-synology" + state_dir.mkdir(parents=True) + (state_dir / "global.yaml").write_text("previous_version: 0.4.1\nauto_upgrade: true\n") + + with ( + patch("pathlib.Path.home", return_value=tmp_path), + patch("mcp_synology.cli.version._get_current_version", return_value="0.5.0"), + patch("mcp_synology.cli.version._detect_installer", return_value="uv"), + patch("subprocess.run", return_value=self._fake_completed(0)), + ): + v._do_revert(None) + + loaded = yaml.safe_load((state_dir / "global.yaml").read_text()) + assert loaded["previous_version"] is None + assert loaded["auto_upgrade"] is False