ifcurl/tests/test_discover.py

"""Tests for ifcurl.discover — local git repository discovery."""

from __future__ import annotations

import json
import time
from pathlib import Path

import pytest

from ifcurl.discover import (
    _cache_file,
    _dump_config,
    _genesis_for_local,
    _get_project_origin,
    _get_search_paths,
    _load_cache,
    _save_cache,
    _save_project_origin,
    _scan_for_git_repos,
    find_local_repos,
    select_local_repo,
)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_repo(path: Path, filename: str = "model.ifc") -> str:
    """Init a git repo at *path* with one commit; return genesis hexsha."""
    import git as gitpkg

    path.mkdir(parents=True, exist_ok=True)
    (path / filename).write_bytes(b"IFC4\n")
    repo = gitpkg.Repo.init(str(path))
    with repo.config_writer() as cw:
        cw.set_value("user", "name", "Test")
        cw.set_value("user", "email", "t@t.com")
    repo.index.add([filename])
    commit = repo.index.commit("init")
    return commit.hexsha


def _write_config(config_dir: Path, paths: list[str]) -> None:
    config_dir.mkdir(parents=True, exist_ok=True)
    (config_dir / "config.toml").write_text(
        "[repo_search]\npaths = [" + ", ".join(f'"{p}"' for p in paths) + "]\n"
    )


# ---------------------------------------------------------------------------
# _scan_for_git_repos
# ---------------------------------------------------------------------------


class TestScanForGitRepos:
    def test_finds_repo_at_top_level(self, tmp_path):
        _make_repo(tmp_path / "proj")
        found = list(_scan_for_git_repos(tmp_path))
        assert tmp_path / "proj" in found

    def test_finds_nested_repo(self, tmp_path):
        _make_repo(tmp_path / "client" / "proj")
        found = list(_scan_for_git_repos(tmp_path))
        assert tmp_path / "client" / "proj" in found

    def test_does_not_descend_into_sub_repo(self, tmp_path):
        _make_repo(tmp_path / "outer")
        _make_repo(tmp_path / "outer" / "inner")
        found = list(_scan_for_git_repos(tmp_path))
        assert tmp_path / "outer" in found
        assert tmp_path / "outer" / "inner" not in found

    def test_skips_hidden_directories(self, tmp_path):
        _make_repo(tmp_path / ".hidden" / "proj")
        found = list(_scan_for_git_repos(tmp_path))
        assert not any(".hidden" in str(p) for p in found)

    def test_empty_search_path_yields_nothing(self, tmp_path):
        assert list(_scan_for_git_repos(tmp_path)) == []

    def test_nonexistent_path_yields_nothing(self, tmp_path):
        assert list(_scan_for_git_repos(tmp_path / "no_such_dir")) == []

    def test_plain_directory_not_yielded(self, tmp_path):
        (tmp_path / "not_a_repo").mkdir()
        assert list(_scan_for_git_repos(tmp_path)) == []

    def test_multiple_repos_all_found(self, tmp_path):
        _make_repo(tmp_path / "a")
        _make_repo(tmp_path / "b")
        found = list(_scan_for_git_repos(tmp_path))
        assert tmp_path / "a" in found
        assert tmp_path / "b" in found


# ---------------------------------------------------------------------------
# _genesis_for_local
# ---------------------------------------------------------------------------


class TestGenesisForLocal:
    def test_returns_hexsha(self, tmp_path):
        hexsha = _make_repo(tmp_path / "repo")
        result = _genesis_for_local(tmp_path / "repo")
        assert result == hexsha

    def test_returns_none_for_non_repo(self, tmp_path):
        (tmp_path / "not_a_repo").mkdir()
        assert _genesis_for_local(tmp_path / "not_a_repo") is None

    def test_returns_none_for_nonexistent_path(self, tmp_path):
        assert _genesis_for_local(tmp_path / "missing") is None


# ---------------------------------------------------------------------------
# _get_search_paths
# ---------------------------------------------------------------------------


class TestGetSearchPaths:
    def test_returns_configured_paths(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_config_dir", lambda *a, **kw: str(tmp_path))
        search_dir = tmp_path / "projects"
        search_dir.mkdir()
        _write_config(tmp_path, [str(search_dir)])
        paths = _get_search_paths()
        assert search_dir in paths

    def test_skips_nonexistent_paths(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_config_dir", lambda *a, **kw: str(tmp_path))
        _write_config(tmp_path, [str(tmp_path / "does_not_exist")])
        assert _get_search_paths() == []

    def test_no_config_returns_empty(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_config_dir", lambda *a, **kw: str(tmp_path))
        assert _get_search_paths() == []


# ---------------------------------------------------------------------------
# Cache helpers
# ---------------------------------------------------------------------------


class TestCache:
    def test_save_and_load(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_cache_dir", lambda *a, **kw: str(tmp_path))
        paths = [Path("/home/user/proj"), Path("/home/user/other")]
        _save_cache("abc123", paths)
        result = _load_cache("abc123")
        assert result == paths

    def test_missing_cache_returns_none(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_cache_dir", lambda *a, **kw: str(tmp_path))
        assert _load_cache("deadbeef" * 5) is None

    def test_expired_cache_returns_none(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_cache_dir", lambda *a, **kw: str(tmp_path))
        cf = _cache_file("abc123")
        cf.parent.mkdir(parents=True, exist_ok=True)
        cf.write_text(json.dumps({"timestamp": time.time() - 7200, "paths": []}))
        assert _load_cache("abc123") is None

    def test_fresh_cache_returned(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_cache_dir", lambda *a, **kw: str(tmp_path))
        _save_cache("abc123", [Path("/some/repo")])
        assert _load_cache("abc123") == [Path("/some/repo")]

    def test_corrupt_cache_returns_none(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_cache_dir", lambda *a, **kw: str(tmp_path))
        cf = _cache_file("abc123")
        cf.parent.mkdir(parents=True, exist_ok=True)
        cf.write_text("not json {{{")
        assert _load_cache("abc123") is None


# ---------------------------------------------------------------------------
# find_local_repos
# ---------------------------------------------------------------------------


class TestFindLocalRepos:
    def _setup(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_config_dir", lambda *a, **kw: str(tmp_path / "config"))
        monkeypatch.setattr("ifcurl.discover.user_cache_dir", lambda *a, **kw: str(tmp_path / "cache"))
        search_dir = tmp_path / "projects"
        search_dir.mkdir()
        _write_config(tmp_path / "config", [str(search_dir)])
        return search_dir

    def test_finds_matching_repo(self, tmp_path, monkeypatch):
        search_dir = self._setup(tmp_path, monkeypatch)
        genesis = _make_repo(search_dir / "building")
        result = find_local_repos(genesis)
        assert search_dir / "building" in result

    def test_ignores_unrelated_repo(self, tmp_path, monkeypatch):
        import git as gitpkg

        search_dir = self._setup(tmp_path, monkeypatch)
        # Give each repo distinct content so their genesis hashes differ
        _make_repo(search_dir / "building")
        other = search_dir / "other"
        other.mkdir()
        (other / "model.ifc").write_bytes(b"IFC4\nDISTINCT CONTENT\n")
        repo = gitpkg.Repo.init(str(other))
        with repo.config_writer() as cw:
            cw.set_value("user", "name", "Test")
            cw.set_value("user", "email", "t@t.com")
        repo.index.add(["model.ifc"])
        unrelated_genesis = repo.index.commit("init other").hexsha

        result = find_local_repos(unrelated_genesis)
        assert search_dir / "building" not in result
        assert search_dir / "other" in result

    def test_finds_fork_by_genesis(self, tmp_path, monkeypatch):
        import git as gitpkg

        search_dir = self._setup(tmp_path, monkeypatch)
        genesis = _make_repo(tmp_path / "upstream")
        gitpkg.Repo.clone_from(str(tmp_path / "upstream"), str(search_dir / "myfork"))
        result = find_local_repos(genesis)
        assert search_dir / "myfork" in result

    def test_no_search_paths_returns_empty(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_config_dir", lambda *a, **kw: str(tmp_path / "config"))
        monkeypatch.setattr("ifcurl.discover.user_cache_dir", lambda *a, **kw: str(tmp_path / "cache"))
        assert find_local_repos("a" * 40) == []

    def test_result_is_cached(self, tmp_path, monkeypatch):
        search_dir = self._setup(tmp_path, monkeypatch)
        genesis = _make_repo(search_dir / "proj")

        find_local_repos(genesis)  # populate cache

        # Remove the actual repo — second call must still return a result from cache
        import shutil
        shutil.rmtree(str(search_dir / "proj"))

        result = find_local_repos(genesis)
        assert search_dir / "proj" in result

    def test_stale_cache_rescans(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_config_dir", lambda *a, **kw: str(tmp_path / "config"))
        monkeypatch.setattr("ifcurl.discover.user_cache_dir", lambda *a, **kw: str(tmp_path / "cache"))
        monkeypatch.setattr("ifcurl.discover._CACHE_TTL", 0)

        search_dir = tmp_path / "projects"
        search_dir.mkdir()
        _write_config(tmp_path / "config", [str(search_dir)])
        genesis = _make_repo(search_dir / "proj")

        # Write a stale cache that claims no repos
        cf = _cache_file(genesis)
        cf.parent.mkdir(parents=True, exist_ok=True)
        cf.write_text(json.dumps({"timestamp": 0, "paths": []}))

        result = find_local_repos(genesis)
        assert search_dir / "proj" in result


# ---------------------------------------------------------------------------
# _dump_config
# ---------------------------------------------------------------------------


class TestDumpConfig:
    def test_string_value(self):
        out = _dump_config({"sec": {"key": "val"}})
        assert '[sec]\nkey = "val"' in out

    def test_bool_false(self):
        out = _dump_config({"sec": {"key": False}})
        assert "key = false" in out

    def test_bool_true(self):
        out = _dump_config({"sec": {"key": True}})
        assert "key = true" in out

    def test_list_value(self):
        out = _dump_config({"sec": {"paths": ["/a", "/b"]}})
        assert 'paths = ["/a", "/b"]' in out

    def test_multiple_sections(self):
        out = _dump_config({"a": {"x": "1"}, "b": {"y": "2"}})
        assert "[a]" in out
        assert "[b]" in out

    def test_roundtrip_via_tomllib(self):
        try:
            import tomllib
        except ImportError:
            import tomli as tomllib
        cfg = {"repo_search": {"paths": ["/home/user/proj"]}, "project_origins": {"abc": "/repo"}}
        out = _dump_config(cfg)
        parsed = tomllib.loads(out)
        assert parsed == cfg


# ---------------------------------------------------------------------------
# _get_project_origin / _save_project_origin
# ---------------------------------------------------------------------------


class TestProjectOriginConfig:
    def _setup(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_config_dir", lambda *a, **kw: str(tmp_path))
        monkeypatch.setattr("ifcurl.discover.user_cache_dir", lambda *a, **kw: str(tmp_path / "cache"))

    def test_missing_returns_none(self, tmp_path, monkeypatch):
        self._setup(tmp_path, monkeypatch)
        assert _get_project_origin("a" * 40) is None

    def test_save_and_read_back(self, tmp_path, monkeypatch):
        self._setup(tmp_path, monkeypatch)
        path = tmp_path / "myrepo"
        _save_project_origin("a" * 40, path)
        assert _get_project_origin("a" * 40) == path

    def test_false_returns_false(self, tmp_path, monkeypatch):
        self._setup(tmp_path, monkeypatch)
        config_dir = tmp_path
        config_dir.mkdir(parents=True, exist_ok=True)
        genesis = "b" * 40
        (config_dir / "config.toml").write_text(f'[project_origins]\n{genesis} = false\n')
        result = _get_project_origin(genesis)
        assert result is False

    def test_save_preserves_existing_sections(self, tmp_path, monkeypatch):
        self._setup(tmp_path, monkeypatch)
        search_dir = tmp_path / "projects"
        search_dir.mkdir()
        _write_config(tmp_path, [str(search_dir)])
        _save_project_origin("c" * 40, tmp_path / "repo")
        try:
            import tomllib
        except ImportError:
            import tomli as tomllib
        cfg = tomllib.loads((tmp_path / "config.toml").read_text())
        assert "repo_search" in cfg
        assert "project_origins" in cfg


# ---------------------------------------------------------------------------
# select_local_repo
# ---------------------------------------------------------------------------


class TestSelectLocalRepo:
    def _setup(self, tmp_path, monkeypatch):
        monkeypatch.setattr("ifcurl.discover.user_config_dir", lambda *a, **kw: str(tmp_path / "config"))
        monkeypatch.setattr("ifcurl.discover.user_cache_dir", lambda *a, **kw: str(tmp_path / "cache"))
        search_dir = tmp_path / "projects"
        search_dir.mkdir()
        _write_config(tmp_path / "config", [str(search_dir)])
        return search_dir

    def test_no_candidates_returns_none(self, tmp_path, monkeypatch):
        self._setup(tmp_path, monkeypatch)
        result = select_local_repo("a" * 40)
        assert result is None

    def test_single_candidate_returned_and_cached(self, tmp_path, monkeypatch):
        search_dir = self._setup(tmp_path, monkeypatch)
        genesis = _make_repo(search_dir / "building")
        result = select_local_repo(genesis)
        assert result == search_dir / "building"
        assert _get_project_origin(genesis) == search_dir / "building"

    def test_cached_path_gone_rescans(self, tmp_path, monkeypatch):
        search_dir = self._setup(tmp_path, monkeypatch)
        genesis = _make_repo(search_dir / "building")
        select_local_repo(genesis)  # caches it

        # Remove the repo — cached path is stale, rescan finds nothing
        import shutil
        shutil.rmtree(str(search_dir / "building"))
        result = select_local_repo(genesis)
        assert result is None

    def test_false_in_config_returns_none(self, tmp_path, monkeypatch):
        search_dir = self._setup(tmp_path, monkeypatch)
        genesis = _make_repo(search_dir / "building")
        config_dir = tmp_path / "config"
        config_path = config_dir / "config.toml"
        existing = config_path.read_text()
        config_path.write_text(existing + f'\n[project_origins]\n{genesis} = false\n')
        result = select_local_repo(genesis)
        assert result is None

    def test_multiple_candidates_calls_prompt_fn(self, tmp_path, monkeypatch):
        import git as gitpkg

        search_dir = self._setup(tmp_path, monkeypatch)
        genesis = _make_repo(tmp_path / "upstream")
        gitpkg.Repo.clone_from(str(tmp_path / "upstream"), str(search_dir / "clone1"))
        gitpkg.Repo.clone_from(str(tmp_path / "upstream"), str(search_dir / "clone2"))

        chosen = search_dir / "clone2"
        called_with: list[list[Path]] = []

        def fake_prompt(candidates):
            called_with.append(candidates)
            return chosen

        result = select_local_repo(genesis, prompt_fn=fake_prompt)
        assert result == chosen
        assert len(called_with) == 1
        assert len(called_with[0]) == 2

    def test_prompt_none_returns_none(self, tmp_path, monkeypatch):
        import git as gitpkg

        search_dir = self._setup(tmp_path, monkeypatch)
        genesis = _make_repo(tmp_path / "upstream")
        gitpkg.Repo.clone_from(str(tmp_path / "upstream"), str(search_dir / "clone1"))
        gitpkg.Repo.clone_from(str(tmp_path / "upstream"), str(search_dir / "clone2"))

        result = select_local_repo(genesis, prompt_fn=lambda _: None)
        assert result is None

    def test_prompt_choice_is_cached(self, tmp_path, monkeypatch):
        import git as gitpkg

        search_dir = self._setup(tmp_path, monkeypatch)
        genesis = _make_repo(tmp_path / "upstream")
        gitpkg.Repo.clone_from(str(tmp_path / "upstream"), str(search_dir / "clone1"))
        gitpkg.Repo.clone_from(str(tmp_path / "upstream"), str(search_dir / "clone2"))

        chosen = search_dir / "clone1"
        select_local_repo(genesis, prompt_fn=lambda _: chosen)
        assert _get_project_origin(genesis) == chosen