mirror of
https://github.com/brunopostle/ifcurl.git
synced 2026-07-12 02:08:13 +00:00
Bruno Postle
a6206a459e
Import sorting, quoted type annotation removal, and black reformatting. No logic changes. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
175 lines
6.6 KiB
Python
175 lines
6.6 KiB
Python
"""Tests for ifcurl.git — git repository access."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from ifcurl.git import fetch_ifc, fetch_ifc_bytes
|
|
from ifcurl.url import IfcUrl
|
|
|
|
|
|
def _local_url(repo_path: str, ref: str = "HEAD", path: str = "model.ifc") -> IfcUrl:
|
|
"""Build an ifc:// URL pointing at a local repo fixture."""
|
|
return IfcUrl.parse(f"ifc://{repo_path}@{ref}?path={path}")
|
|
|
|
|
|
class TestFetchIfc:
|
|
def test_returns_hexsha_and_bytes(self, local_ifc_repo):
|
|
url = _local_url(local_ifc_repo["path"])
|
|
hexsha, data = fetch_ifc(url)
|
|
assert len(hexsha) == 40
|
|
assert all(c in "0123456789abcdef" for c in hexsha)
|
|
assert isinstance(data, bytes)
|
|
assert len(data) > 0
|
|
|
|
def test_hexsha_matches_commit(self, local_ifc_repo):
|
|
url = _local_url(local_ifc_repo["path"])
|
|
hexsha, _ = fetch_ifc(url)
|
|
assert hexsha == local_ifc_repo["hexsha"]
|
|
|
|
def test_bytes_match_committed_file(self, local_ifc_repo):
|
|
url = _local_url(local_ifc_repo["path"])
|
|
_, data = fetch_ifc(url)
|
|
assert data == local_ifc_repo["bytes"]
|
|
|
|
def test_explicit_commit_hash_ref(self, local_ifc_repo):
|
|
url = _local_url(local_ifc_repo["path"], ref=local_ifc_repo["hexsha"])
|
|
hexsha, data = fetch_ifc(url)
|
|
assert hexsha == local_ifc_repo["hexsha"]
|
|
assert data == local_ifc_repo["bytes"]
|
|
|
|
def test_tag_ref(self, local_ifc_repo):
|
|
url = _local_url(local_ifc_repo["path"], ref="tags/v1.0")
|
|
hexsha, data = fetch_ifc(url)
|
|
assert hexsha == local_ifc_repo["hexsha"]
|
|
|
|
def test_branch_ref(self, local_ifc_repo):
|
|
import git as gitpkg
|
|
|
|
branch = gitpkg.Repo(local_ifc_repo["path"]).active_branch.name
|
|
url = _local_url(local_ifc_repo["path"], ref=f"heads/{branch}")
|
|
hexsha, data = fetch_ifc(url)
|
|
assert hexsha == local_ifc_repo["hexsha"]
|
|
assert data == local_ifc_repo["bytes"]
|
|
|
|
def test_bad_ref_raises(self, local_ifc_repo):
|
|
url = _local_url(local_ifc_repo["path"], ref="heads/nonexistent")
|
|
with pytest.raises(ValueError, match="not found"):
|
|
fetch_ifc(url)
|
|
|
|
def test_bad_path_raises(self, local_ifc_repo):
|
|
url = _local_url(local_ifc_repo["path"], path="missing.ifc")
|
|
with pytest.raises(ValueError, match="not found"):
|
|
fetch_ifc(url)
|
|
|
|
def test_directory_path_raises(self, local_ifc_repo):
|
|
# Committing a subdir would be complex; test that asking for a
|
|
# non-blob raises — here we ask for the root tree entry.
|
|
url = _local_url(local_ifc_repo["path"], path=".")
|
|
with pytest.raises((ValueError, KeyError)):
|
|
fetch_ifc(url)
|
|
|
|
def test_no_path_raises(self, local_ifc_repo):
|
|
url = IfcUrl.parse(f"ifc://{local_ifc_repo['path']}@HEAD")
|
|
with pytest.raises(ValueError, match="'path' parameter"):
|
|
fetch_ifc(url)
|
|
|
|
|
|
class TestFetchIfcBytes:
|
|
def test_returns_bytes(self, local_ifc_repo):
|
|
url = _local_url(local_ifc_repo["path"])
|
|
data = fetch_ifc_bytes(url)
|
|
assert data == local_ifc_repo["bytes"]
|
|
|
|
def test_backwards_compat_with_fetch_ifc(self, local_ifc_repo):
|
|
url = _local_url(local_ifc_repo["path"])
|
|
_, expected = fetch_ifc(url)
|
|
assert fetch_ifc_bytes(url) == expected
|
|
|
|
|
|
class TestInvalidRepo:
|
|
def test_nonexistent_path_raises(self, tmp_path):
|
|
url = IfcUrl.parse(f"ifc://{tmp_path / 'no_such_repo'}@HEAD?path=m.ifc")
|
|
with pytest.raises(ValueError):
|
|
fetch_ifc(url)
|
|
|
|
def test_non_git_directory_raises(self, tmp_path):
|
|
(tmp_path / "model.ifc").write_text("not a repo")
|
|
url = IfcUrl.parse(f"ifc://{tmp_path}@HEAD?path=model.ifc")
|
|
with pytest.raises(ValueError):
|
|
fetch_ifc(url)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cache management helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
from ifcurl.git import _dir_size, _evict_if_needed, _repo_cache_entries, _touch_cache
|
|
|
|
|
|
def _make_fake_repo_cache(base: Path, url: str, size_bytes: int = 1024) -> Path:
|
|
"""Create a fake cached-repo directory structure under *base*."""
|
|
import hashlib
|
|
|
|
key = hashlib.sha256(url.encode()).hexdigest()[:24]
|
|
entry = base / key
|
|
git_dir = entry / "repo.git"
|
|
git_dir.mkdir(parents=True)
|
|
(git_dir / "dummy").write_bytes(b"\x00" * size_bytes)
|
|
(entry / "remote_url").write_text(url)
|
|
return entry
|
|
|
|
|
|
class TestCacheHelpers:
|
|
def test_dir_size_counts_nested_files(self, tmp_path):
|
|
(tmp_path / "a").write_bytes(b"x" * 100)
|
|
(tmp_path / "sub").mkdir()
|
|
(tmp_path / "sub" / "b").write_bytes(b"y" * 200)
|
|
assert _dir_size(tmp_path) == 300
|
|
|
|
def test_touch_cache_updates_mtime(self, tmp_path):
|
|
import time
|
|
|
|
old_mtime = tmp_path.stat().st_mtime
|
|
time.sleep(0.05)
|
|
_touch_cache(tmp_path)
|
|
assert tmp_path.stat().st_mtime > old_mtime
|
|
|
|
def test_repo_cache_entries_lists_repos(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("ifcurl.git.user_cache_dir", lambda *a, **kw: str(tmp_path))
|
|
_make_fake_repo_cache(tmp_path, "https://example.com/org/repo")
|
|
_make_fake_repo_cache(tmp_path, "https://github.com/org/other")
|
|
entries = _repo_cache_entries()
|
|
urls = {e[3] for e in entries}
|
|
assert "https://example.com/org/repo" in urls
|
|
assert "https://github.com/org/other" in urls
|
|
|
|
def test_repo_cache_entries_ignores_non_repo_dirs(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("ifcurl.git.user_cache_dir", lambda *a, **kw: str(tmp_path))
|
|
(tmp_path / "not_a_repo").mkdir() # no repo.git subdir
|
|
entries = _repo_cache_entries()
|
|
assert len(entries) == 0
|
|
|
|
def test_evict_removes_oldest_over_limit(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("ifcurl.git.user_cache_dir", lambda *a, **kw: str(tmp_path))
|
|
monkeypatch.setenv("IFCURL_CACHE_MAX_GB", "0.000001") # ~1 KB limit
|
|
old = _make_fake_repo_cache(
|
|
tmp_path, "https://old.example.com/repo", size_bytes=512
|
|
)
|
|
time.sleep(0.05)
|
|
new = _make_fake_repo_cache(
|
|
tmp_path, "https://new.example.com/repo", size_bytes=512
|
|
)
|
|
_evict_if_needed()
|
|
assert not old.exists()
|
|
assert new.exists()
|
|
|
|
def test_evict_no_limit_does_nothing(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("ifcurl.git.user_cache_dir", lambda *a, **kw: str(tmp_path))
|
|
monkeypatch.delenv("IFCURL_CACHE_MAX_GB", raising=False)
|
|
entry = _make_fake_repo_cache(tmp_path, "https://example.com/repo")
|
|
_evict_if_needed()
|
|
assert entry.exists()
|