test_update.py

import json
import subprocess
import sys

import pytest


def test_update_reconciles_unique_unqualified_mcp_name(tmp_home, tmp_path):
    source = tmp_path / "mempalace"
    source.mkdir()
    (source / "capability.yaml").write_text(
        "kind: mcp-server\n"
        "name: mempalace\n"
        "version: 1.0.0\n"
        "frameworks: [opencode]\n"
        "mcp:\n"
        "  transport: stdio\n"
        "  command: uvx\n"
        "  args: [mempalace-mcp]\n"
    )

    from capacium.commands.install import install_capability
    assert install_capability(
        "MemPalace/mempalace",
        source_dir=source,
        no_lock=True,
        skip_runtime_check=True,
        force=True,
        yes=True,
    ) is True

    config_path = tmp_home / ".config" / "opencode" / "opencode.json"
    config_path.write_text(json.dumps({}))

    from capacium.commands.update import update_capability
    assert update_capability("mempalace", skip_runtime_check=True) is True

    data = json.loads(config_path.read_text())
    assert data["mcp"]["mempalace"] == {
        "type": "local",
        "command": ["uvx", "mempalace-mcp"],
        "enabled": True,
    }


def test_update_reports_ambiguous_unqualified_name(tmp_home, tmp_path, capsys):
    from capacium.registry import Registry
    from capacium.models import Capability, Kind
    from datetime import datetime

    registry = Registry()
    for owner in ("alice", "bob"):
        registry.add_capability(Capability(
            owner=owner,
            name="shared",
            version="1.0.0",
            kind=Kind.SKILL,
            fingerprint="fp",
            install_path=tmp_path,
            installed_at=datetime.now(),
            framework="opencode",
        ))

    from capacium.commands.update import update_capability
    assert update_capability("shared", skip_runtime_check=True) is False

    out = capsys.readouterr().out
    assert "ambiguous" in out
    assert "alice/shared" in out
    assert "bob/shared" in out


def test_parse_version_orders_correctly():
    from capacium.commands.update import _parse_version
    assert _parse_version("1.0.0") < _parse_version("2.0.0")
    assert _parse_version("0.5.1") < _parse_version("0.5.6")
    assert _parse_version("0.9.9") < _parse_version("1.0.0")
    assert _parse_version("0.5.6") > _parse_version("0.5.1")
    assert _parse_version("1.0.0") == _parse_version("1.0.0")


def test_is_git_url_accepts_http_and_ssh():
    from capacium.commands.update import _is_git_url
    assert _is_git_url("https://github.com/foo/bar.git")
    assert _is_git_url("git@github.com:foo/bar.git")
    assert not _is_git_url("")


def test_fetch_remote_git_tags_finds_newer_tags(tmp_path):
    remote = tmp_path / "remote-repo"
    remote.mkdir()
    subprocess.run(["git", "init", "--bare"], cwd=remote, capture_output=True)

    clone = tmp_path / "clone"
    clone.mkdir()
    subprocess.run(["git", "init"], cwd=clone, capture_output=True)
    subprocess.run(["git", "config", "user.email", "t@t"], cwd=clone, capture_output=True)
    subprocess.run(["git", "config", "user.name", "t"], cwd=clone, capture_output=True)
    (clone / "readme.md").write_text("hello")
    subprocess.run(["git", "add", "."], cwd=clone, capture_output=True)
    subprocess.run(["git", "commit", "-m", "init"], cwd=clone, capture_output=True)
    subprocess.run(["git", "remote", "add", "origin", str(remote)], cwd=clone, capture_output=True)
    subprocess.run(["git", "push", "origin", "main"], cwd=clone, capture_output=True)
    subprocess.run(["git", "tag", "v1.0.0"], cwd=clone, capture_output=True)
    subprocess.run(["git", "tag", "v2.0.0"], cwd=clone, capture_output=True)
    subprocess.run(["git", "push", "origin", "--tags"], cwd=clone, capture_output=True)

    from capacium.commands.update import _fetch_remote_git_tags
    tags = _fetch_remote_git_tags(str(remote))
    assert "1.0.0" in tags
    assert "2.0.0" in tags


def test_fetch_remote_git_tags_no_remote():
    from capacium.commands.update import _fetch_remote_git_tags
    tags = _fetch_remote_git_tags("https://invalid.local/repo.git")
    assert tags == []


@pytest.mark.skipif(sys.platform == "win32", reason="npm/uv subprocess resolution on Windows — tracked as V16")
def test_check_for_newer_version_via_remote(tmp_path):
    remote = tmp_path / "remote-repo"
    remote.mkdir()
    subprocess.run(["git", "init", "--bare"], cwd=remote, capture_output=True)

    clone = tmp_path / "clone"
    clone.mkdir()
    subprocess.run(["git", "init"], cwd=clone, capture_output=True)
    subprocess.run(["git", "config", "user.email", "t@t"], cwd=clone, capture_output=True)
    subprocess.run(["git", "config", "user.name", "t"], cwd=clone, capture_output=True)
    (clone / "capability.yaml").write_text(
        "kind: skill\nname: test-cap\nversion: 2.0.0\n"
    )
    subprocess.run(["git", "add", "."], cwd=clone, capture_output=True)
    subprocess.run(["git", "commit", "-m", "init"], cwd=clone, capture_output=True)
    subprocess.run(["git", "remote", "add", "origin", str(remote)], cwd=clone, capture_output=True)
    subprocess.run(["git", "push", "origin", "main"], cwd=clone, capture_output=True)
    subprocess.run(
        ["git", "tag", "-a", "v2.0.0", "-m", "annotated v2"],
        cwd=clone,
        capture_output=True,
    )
    subprocess.run(["git", "tag", "v3.0.0-rc.1"], cwd=clone, capture_output=True)
    subprocess.run(["git", "push", "origin", "--tags"], cwd=clone, capture_output=True)

    from capacium.commands.update import _check_for_newer_version
    from unittest.mock import patch
    with patch("capacium.commands.update.install_capability", return_value=True) as mock:
        result = _check_for_newer_version(
            "global/test-cap", "1.0.0", remote.as_uri()
        )
        assert result is True
        mock.assert_called_once_with(
            "global/test-cap@2.0.0",
            source_dir=remote.as_uri(),
            force=True,
            yes=True,
        )


@pytest.mark.skipif(sys.platform == "win32", reason="npm/uv subprocess resolution on Windows — tracked as V16")
def test_reconcile_rewrites_entry_for_new_entrypoint(tmp_home, tmp_path, monkeypatch):
    """When a capability update adds an entrypoint: field, config entries
    are rewritten to point to the entrypoint subdirectory."""
    source = tmp_path / "entry-evolve"
    source.mkdir()
    (source / "capability.yaml").write_text(
        "kind: mcp-server\n"
        "name: entry-evolve\n"
        "version: 1.0.0\n"
        "frameworks: [opencode]\n"
        "mcp:\n  transport: stdio\n  command: npx\n"
    )

    from capacium.commands.install import install_capability
    assert install_capability(
        "EntryEvolve/entry-evolve",
        source_dir=source,
        no_lock=True,
        skip_runtime_check=True,
        force=True,
        yes=True,
    ) is True

    config_path = tmp_home / ".config" / "opencode" / "opencode.json"

    data = json.loads(config_path.read_text())
    assert data["mcp"]["entry-evolve"]["type"] == "local"

    from capacium.registry import Registry
    registry = Registry()
    cap = registry.get_capability("EntryEvolve/entry-evolve")
    assert cap is not None
    install_path = cap.install_path
    assert install_path is not None
    assert install_path.exists()

    sub = install_path / "server"
    sub.mkdir()
    (sub / "package.json").write_text('{"bin": {"serve": "dist/cli.js"}}')
    (install_path / "capability.yaml").write_text(
        "kind: mcp-server\n"
        "name: entry-evolve\n"
        "version: 1.0.0\n"
        "frameworks: [opencode]\n"
        "entrypoint: server\n"
        "mcp:\n  transport: stdio\n  command: npx\n"
    )

    from capacium.commands.update import update_capability
    assert update_capability("entry-evolve", skip_runtime_check=True) is True

    data = json.loads(config_path.read_text())
    full_entry = data["mcp"]["entry-evolve"]
    assert full_entry["type"] == "local"
    assert full_entry["enabled"] is True
    assert full_entry["command"][0] == "npx"