wehub-resource-sync
337 lines
12 KiB
Python
337 lines
12 KiB
Python
"""Tests for multi-repo registry and connection pool."""
|
|
|
|
import sqlite3
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from code_review_graph.registry import ConnectionPool, Registry, resolve_repo
|
|
|
|
|
|
class TestRegistry:
|
|
def setup_method(self):
|
|
self.tmp_dir = tempfile.mkdtemp()
|
|
self.registry_path = Path(self.tmp_dir) / "registry.json"
|
|
self.registry = Registry(path=self.registry_path)
|
|
|
|
# Create fake repos
|
|
self.repo1 = Path(self.tmp_dir) / "repo1"
|
|
self.repo1.mkdir()
|
|
(self.repo1 / ".git").mkdir()
|
|
|
|
self.repo2 = Path(self.tmp_dir) / "repo2"
|
|
self.repo2.mkdir()
|
|
(self.repo2 / ".code-review-graph").mkdir()
|
|
|
|
def teardown_method(self):
|
|
import shutil
|
|
shutil.rmtree(self.tmp_dir, ignore_errors=True)
|
|
|
|
def test_register_and_list(self):
|
|
"""Register repos and list them back."""
|
|
self.registry.register(str(self.repo1), alias="r1")
|
|
self.registry.register(str(self.repo2), alias="r2")
|
|
|
|
repos = self.registry.list_repos()
|
|
assert len(repos) == 2
|
|
paths = [r["path"] for r in repos]
|
|
assert str(self.repo1.resolve()) in paths
|
|
assert str(self.repo2.resolve()) in paths
|
|
|
|
def test_register_duplicate_path(self):
|
|
"""Registering the same path twice updates alias."""
|
|
self.registry.register(str(self.repo1), alias="first")
|
|
self.registry.register(str(self.repo1), alias="second")
|
|
|
|
repos = self.registry.list_repos()
|
|
assert len(repos) == 1
|
|
assert repos[0]["alias"] == "second"
|
|
|
|
def test_register_invalid_path(self):
|
|
"""Registering a non-existent path raises ValueError."""
|
|
import pytest
|
|
with pytest.raises(ValueError, match="not a directory"):
|
|
self.registry.register("/nonexistent/path/repo")
|
|
|
|
def test_register_not_a_repo(self):
|
|
"""Registering a dir without .git or .code-review-graph raises ValueError."""
|
|
import pytest
|
|
bare_dir = Path(self.tmp_dir) / "bare"
|
|
bare_dir.mkdir()
|
|
with pytest.raises(ValueError, match="does not look like a repository"):
|
|
self.registry.register(str(bare_dir))
|
|
|
|
def test_unregister_by_path(self):
|
|
"""Unregister a repo by path."""
|
|
self.registry.register(str(self.repo1), alias="r1")
|
|
assert len(self.registry.list_repos()) == 1
|
|
|
|
result = self.registry.unregister(str(self.repo1))
|
|
assert result is True
|
|
assert len(self.registry.list_repos()) == 0
|
|
|
|
def test_unregister_by_alias(self):
|
|
"""Unregister a repo by alias."""
|
|
self.registry.register(str(self.repo1), alias="myalias")
|
|
assert len(self.registry.list_repos()) == 1
|
|
|
|
result = self.registry.unregister("myalias")
|
|
assert result is True
|
|
assert len(self.registry.list_repos()) == 0
|
|
|
|
def test_unregister_not_found(self):
|
|
"""Unregistering a non-registered repo returns False."""
|
|
result = self.registry.unregister("nonexistent")
|
|
assert result is False
|
|
|
|
def test_find_by_alias(self):
|
|
"""find_by_alias returns correct entry."""
|
|
self.registry.register(str(self.repo1), alias="myrepo")
|
|
entry = self.registry.find_by_alias("myrepo")
|
|
assert entry is not None
|
|
assert entry["alias"] == "myrepo"
|
|
assert entry["path"] == str(self.repo1.resolve())
|
|
|
|
def test_find_by_alias_not_found(self):
|
|
"""find_by_alias returns None for unknown alias."""
|
|
entry = self.registry.find_by_alias("nope")
|
|
assert entry is None
|
|
|
|
def test_find_by_path(self):
|
|
"""find_by_path returns correct entry."""
|
|
self.registry.register(str(self.repo1), alias="r1")
|
|
entry = self.registry.find_by_path(str(self.repo1))
|
|
assert entry is not None
|
|
assert entry["path"] == str(self.repo1.resolve())
|
|
|
|
def test_persistence(self):
|
|
"""Registry persists to disk and reloads correctly."""
|
|
self.registry.register(str(self.repo1), alias="persistent")
|
|
|
|
# Create a new registry from the same file
|
|
registry2 = Registry(path=self.registry_path)
|
|
repos = registry2.list_repos()
|
|
assert len(repos) == 1
|
|
assert repos[0]["alias"] == "persistent"
|
|
|
|
def test_resolve_by_alias(self):
|
|
"""resolve_repo resolves alias to path."""
|
|
self.registry.register(str(self.repo1), alias="r1")
|
|
result = resolve_repo(self.registry, "r1")
|
|
assert result == str(self.repo1.resolve())
|
|
|
|
def test_resolve_by_direct_path(self):
|
|
"""resolve_repo resolves direct path."""
|
|
result = resolve_repo(self.registry, str(self.repo1))
|
|
assert result == str(self.repo1.resolve())
|
|
|
|
def test_resolve_by_cwd(self):
|
|
"""resolve_repo falls back to cwd when repo is None."""
|
|
result = resolve_repo(self.registry, None, cwd=str(self.repo1))
|
|
assert result == str(self.repo1.resolve())
|
|
|
|
def test_resolve_returns_none(self):
|
|
"""resolve_repo returns None when nothing matches."""
|
|
result = resolve_repo(self.registry, None)
|
|
assert result is None
|
|
|
|
|
|
class TestConnectionPool:
|
|
def setup_method(self):
|
|
self.tmp_dir = tempfile.mkdtemp()
|
|
self.pool = ConnectionPool(max_size=3)
|
|
|
|
def teardown_method(self):
|
|
self.pool.close_all()
|
|
import shutil
|
|
shutil.rmtree(self.tmp_dir, ignore_errors=True)
|
|
|
|
def _make_db(self, name: str) -> str:
|
|
"""Create a temporary SQLite database file."""
|
|
db_path = str(Path(self.tmp_dir) / f"{name}.db")
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("CREATE TABLE IF NOT EXISTS test (id INTEGER)")
|
|
conn.close()
|
|
return db_path
|
|
|
|
def test_get_creates_connection(self):
|
|
"""get() creates a new connection."""
|
|
db_path = self._make_db("test1")
|
|
conn = self.pool.get(db_path)
|
|
assert conn is not None
|
|
assert self.pool.size == 1
|
|
|
|
def test_get_reuses_connection(self):
|
|
"""get() returns the same connection for the same path."""
|
|
db_path = self._make_db("test1")
|
|
conn1 = self.pool.get(db_path)
|
|
conn2 = self.pool.get(db_path)
|
|
assert conn1 is conn2
|
|
assert self.pool.size == 1
|
|
|
|
def test_eviction_on_full(self):
|
|
"""Pool evicts LRU connection when full."""
|
|
db1 = self._make_db("db1")
|
|
db2 = self._make_db("db2")
|
|
db3 = self._make_db("db3")
|
|
db4 = self._make_db("db4")
|
|
|
|
self.pool.get(db1)
|
|
self.pool.get(db2)
|
|
self.pool.get(db3)
|
|
assert self.pool.size == 3
|
|
|
|
# Adding 4th should evict db1 (LRU)
|
|
self.pool.get(db4)
|
|
assert self.pool.size == 3
|
|
|
|
def test_close_all(self):
|
|
"""close_all() clears all connections."""
|
|
db1 = self._make_db("db1")
|
|
db2 = self._make_db("db2")
|
|
|
|
self.pool.get(db1)
|
|
self.pool.get(db2)
|
|
assert self.pool.size == 2
|
|
|
|
self.pool.close_all()
|
|
assert self.pool.size == 0
|
|
|
|
def test_lru_ordering(self):
|
|
"""Recently used connections are kept over stale ones."""
|
|
db1 = self._make_db("db1")
|
|
db2 = self._make_db("db2")
|
|
db3 = self._make_db("db3")
|
|
db4 = self._make_db("db4")
|
|
|
|
conn1 = self.pool.get(db1)
|
|
self.pool.get(db2)
|
|
self.pool.get(db3)
|
|
|
|
# Access db1 again to make it recently used
|
|
self.pool.get(db1)
|
|
|
|
# Now add db4 — db2 should be evicted (LRU), not db1
|
|
self.pool.get(db4)
|
|
assert self.pool.size == 3
|
|
|
|
# db1 should still be in pool
|
|
conn1_again = self.pool.get(db1)
|
|
assert conn1_again is conn1
|
|
|
|
|
|
class TestCrossRepoSearch:
|
|
def test_cross_repo_search_no_repos(self):
|
|
"""cross_repo_search with empty registry returns empty results."""
|
|
from code_review_graph.tools import cross_repo_search_func
|
|
|
|
tmp_dir = tempfile.mkdtemp()
|
|
|
|
with patch("code_review_graph.registry.Registry") as mock_registry_cls:
|
|
mock_instance = MagicMock()
|
|
mock_instance.list_repos.return_value = []
|
|
mock_registry_cls.return_value = mock_instance
|
|
|
|
result = cross_repo_search_func(query="test")
|
|
assert result["status"] == "ok"
|
|
assert result["results"] == []
|
|
|
|
import shutil
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
|
|
|
|
class TestSetDataDir:
|
|
"""Tests for set_data_dir and get_data_dir_for_repo methods."""
|
|
|
|
def setup_method(self):
|
|
"""Set up isolated test registry."""
|
|
self.tmp_dir = tempfile.mkdtemp()
|
|
self.registry_path = Path(self.tmp_dir) / "registry.json"
|
|
self.registry = Registry(path=self.registry_path)
|
|
|
|
def teardown_method(self):
|
|
"""Clean up temporary directory."""
|
|
import shutil
|
|
shutil.rmtree(self.tmp_dir, ignore_errors=True)
|
|
|
|
def test_set_data_dir_new_repo(self):
|
|
"""set_data_dir should create new registry entry if repo not registered."""
|
|
repo = Path(self.tmp_dir) / "project"
|
|
repo.mkdir()
|
|
data_dir = Path(self.tmp_dir) / "data"
|
|
|
|
entry = self.registry.set_data_dir(str(repo), str(data_dir))
|
|
|
|
assert entry["path"] == str(repo.resolve())
|
|
assert entry["data_dir"] == str(data_dir.resolve())
|
|
|
|
# Verify it can be retrieved
|
|
retrieved = self.registry.get_data_dir_for_repo(str(repo))
|
|
assert retrieved == str(data_dir.resolve())
|
|
|
|
# Verify entry is in list
|
|
repos = self.registry.list_repos()
|
|
assert len(repos) == 1
|
|
assert repos[0]["path"] == str(repo.resolve())
|
|
|
|
def test_set_data_dir_existing_repo(self):
|
|
"""set_data_dir should update data_dir for already registered repo."""
|
|
repo = Path(self.tmp_dir) / "project"
|
|
repo.mkdir()
|
|
data_dir1 = Path(self.tmp_dir) / "data1"
|
|
data_dir2 = Path(self.tmp_dir) / "data2"
|
|
|
|
# Initial registration
|
|
entry1 = self.registry.set_data_dir(str(repo), str(data_dir1))
|
|
assert entry1["data_dir"] == str(data_dir1.resolve())
|
|
|
|
# Update with new data_dir
|
|
entry2 = self.registry.set_data_dir(str(repo), str(data_dir2))
|
|
assert entry2["data_dir"] == str(data_dir2.resolve())
|
|
|
|
# Verify only one entry exists
|
|
repos = self.registry.list_repos()
|
|
assert len(repos) == 1
|
|
|
|
def test_get_data_dir_for_repo_unknown(self):
|
|
"""get_data_dir_for_repo should return None for unknown repo."""
|
|
unknown_repo = Path(self.tmp_dir) / "unknown"
|
|
|
|
result = self.registry.get_data_dir_for_repo(str(unknown_repo))
|
|
assert result is None
|
|
|
|
def test_set_data_dir_with_alias(self):
|
|
"""register() with data_dir should store both."""
|
|
repo = Path(self.tmp_dir) / "project"
|
|
repo.mkdir()
|
|
(repo / ".git").mkdir()
|
|
data_dir = Path(self.tmp_dir) / "data"
|
|
alias = "my-project"
|
|
|
|
entry = self.registry.register(str(repo), alias=alias, data_dir=str(data_dir))
|
|
|
|
assert entry["path"] == str(repo.resolve())
|
|
assert entry["alias"] == alias
|
|
assert entry["data_dir"] == str(data_dir.resolve())
|
|
|
|
def test_backward_compatibility(self):
|
|
"""Old registry entries without data_dir should work."""
|
|
repo = Path(self.tmp_dir) / "project"
|
|
repo.mkdir()
|
|
|
|
# Create entry without data_dir (old format)
|
|
self.registry._repos.append({
|
|
"path": str(repo.resolve()),
|
|
"alias": "old-project"
|
|
})
|
|
self.registry._save()
|
|
|
|
# Should not crash
|
|
result = self.registry.get_data_dir_for_repo(str(repo))
|
|
assert result is None
|
|
|
|
# Should be able to add data_dir
|
|
data_dir = Path(self.tmp_dir) / "data"
|
|
entry = self.registry.set_data_dir(str(repo), str(data_dir))
|
|
assert entry["data_dir"] == str(data_dir.resolve())
|