"""ACL gating + field handling for services/systems.py (unit, mocked session).""" from unittest.mock import AsyncMock, MagicMock, patch import pytest from tests.helpers import make_mock_session @pytest.mark.asyncio async def test_create_system_denied_without_project_write(): with patch("scribe.services.systems.access") as acc: acc.can_write_project = AsyncMock(return_value=False) from scribe.services.systems import create_system result = await create_system(user_id=1, project_id=5, name="Reader") assert result is None @pytest.mark.asyncio async def test_create_system_sets_fields_when_authorized(): mock_session = make_mock_session() captured = {} def _capture_add(obj): captured["name"] = getattr(obj, "name", "MISSING") captured["project_id"] = getattr(obj, "project_id", "MISSING") mock_session.add = MagicMock(side_effect=_capture_add) with patch("scribe.services.systems.async_session") as mock_cls, \ patch("scribe.services.systems.access") as acc: acc.can_write_project = AsyncMock(return_value=True) mock_cls.return_value = mock_session from scribe.services.systems import create_system await create_system(user_id=1, project_id=5, name=" Reader ") assert captured["name"] == "Reader" # stripped assert captured["project_id"] == 5 @pytest.mark.asyncio async def test_set_record_systems_denied_without_note_write(): with patch("scribe.services.systems.access") as acc: acc.can_write_note = AsyncMock(return_value=False) from scribe.services.systems import set_record_systems result = await set_record_systems(user_id=1, note_id=9, system_ids=[1, 2]) assert result is None @pytest.mark.asyncio async def test_list_systems_denied_returns_empty(): with patch("scribe.services.systems.access") as acc: acc.can_read_project = AsyncMock(return_value=False) from scribe.services.systems import list_systems result = await list_systems(user_id=1, project_id=5) assert result == [] @pytest.mark.asyncio async def test_count_open_issues_denied_returns_zero(): with patch("scribe.services.systems.access") as acc: acc.can_read_project = AsyncMock(return_value=False) from scribe.services.systems import count_open_issues result = await count_open_issues(user_id=1, project_id=5) assert result == 0 # ── assess_system_name — the one gate both doors call (milestone 307) ── @pytest.mark.asyncio async def test_assess_matches_an_existing_system_on_case_and_spacing(): """The local gate compares the operator's own spelling: case and runs of whitespace are not a different System. Anything stronger belongs to the CANONICAL slug, which is a different question (is this the same AREA as another project's System) with a different answer.""" existing = MagicMock() existing.id, existing.name = 7, "Scrape Pipeline" with patch("scribe.services.systems.list_systems", AsyncMock(return_value=[existing])): from scribe.services.systems import assess_system_name out = await assess_system_name(1, 5, " scrape pipeline ") assert out["duplicate"] == {"id": 7, "name": "Scrape Pipeline"} # A duplicate short-circuits: there is nothing to file when nothing is created. assert out["canonical"] is None @pytest.mark.asyncio async def test_assess_prefers_an_exact_area_over_a_merely_similar_one(): from scribe.services.systems import assess_system_name exact = MagicMock() exact.id, exact.name = 3, "CI & Release" with patch("scribe.services.systems.list_systems", AsyncMock(return_value=[])), \ patch("scribe.services.systems.canonical_systems_svc") as canon: canon.find_by_name = AsyncMock(return_value=exact) canon.best_overlap = AsyncMock() out = await assess_system_name(1, 5, "CI and Release") assert out["canonical"] == {"id": 3, "name": "CI & Release", "basis": "exact"} # An exact hit is the answer — no need to go looking for a lesser one. canon.best_overlap.assert_not_awaited() @pytest.mark.asyncio async def test_assess_falls_back_to_overlap_and_never_invents_a_match(): from scribe.services.systems import assess_system_name near = {"id": 3, "name": "CI & Release", "basis": "overlap", "score": 0.33} with patch("scribe.services.systems.list_systems", AsyncMock(return_value=[])), \ patch("scribe.services.systems.canonical_systems_svc") as canon: canon.find_by_name = AsyncMock(return_value=None) canon.best_overlap = AsyncMock(return_value=near) assert (await assess_system_name(1, 5, "CI & runners"))["canonical"] == near canon.best_overlap = AsyncMock(return_value=None) # Nothing resembles it: a project-specific area, and silence is the # right answer — unmapped is a valid resting state. assert (await assess_system_name(1, 5, "Soundboard"))["canonical"] is None @pytest.mark.asyncio async def test_assess_fails_open_so_a_naming_aid_cannot_block_a_create(): """Both arms degrade to "no opinion" rather than raising. A create must never fail because the catalog was unreachable.""" from scribe.services.systems import assess_system_name with patch("scribe.services.systems.list_systems", AsyncMock(side_effect=RuntimeError("db down"))): assert await assess_system_name(1, 5, "Reader") == {"duplicate": None, "canonical": None} with patch("scribe.services.systems.list_systems", AsyncMock(return_value=[])), \ patch("scribe.services.systems.canonical_systems_svc") as canon: canon.find_by_name = AsyncMock(side_effect=RuntimeError("db down")) assert (await assess_system_name(1, 5, "Reader"))["canonical"] is None @pytest.mark.asyncio async def test_assess_says_nothing_about_a_nameless_system(): from scribe.services.systems import assess_system_name assert await assess_system_name(1, 5, " ") == {"duplicate": None, "canonical": None} # ── path patterns — the files that ARE the area (milestone 444, #4756) ── def test_patterns_are_tidied_to_repo_relative_and_deduplicated(): from scribe.services.systems import normalize_path_patterns out = normalize_path_patterns([ " ./src/billing/ ", "/src/billing", "", "src\\api\\*.py", " ", "frontend/src/**/Billing*.vue", ]) assert out == ["src/billing", "src/api/*.py", "frontend/src/**/Billing*.vue"] assert normalize_path_patterns(None) == [] assert normalize_path_patterns([]) == [] @pytest.mark.parametrize("bad", [ "src/billing", # a bare string, not a list ["../other-repo/src"], # leaves the repo ["src/../../etc"], [42], ["x" * 301], [f"src/f{i}.py" for i in range(51)], ]) def test_patterns_that_cannot_mean_an_area_are_refused(bad): from scribe.services.systems import normalize_path_patterns with pytest.raises(ValueError): normalize_path_patterns(bad) @pytest.mark.parametrize("pattern,path,expected", [ # A plain directory covers everything under it, at any depth. ("src/billing", "src/billing/invoice.py", True), ("src/billing", "src/billing/deep/er/x.py", True), ("src/billing", "src/billing", True), ("src/billing", "src/billingual/x.py", False), # `*` stays inside one segment; `**` spans any number, including none. ("src/*.py", "src/app.py", True), ("src/*.py", "src/pkg/app.py", False), ("src/**/*.py", "src/pkg/sub/app.py", True), ("src/**/*.py", "src/app.py", True), ("**/Billing*.vue", "frontend/src/components/BillingCard.vue", True), # Case counts, and the path is read the way the pattern was written. ("src/billing", "Src/billing/x.py", False), ("src/billing", "./src/billing/x.py", True), ("src/billing", "", False), ]) def test_path_matches(pattern, path, expected): from scribe.services.systems import path_matches assert path_matches(pattern, path) is expected @pytest.mark.asyncio async def test_systems_for_paths_returns_every_area_a_path_belongs_to(): """Areas overlap, and each one a path belongs to answers for it — the lookup never picks a winner. A System that named no files claims none.""" api = MagicMock(id=1, path_patterns=["src/scribe/routes", "src/scribe/mcp/tools"]) data = MagicMock(id=2, path_patterns=["src/scribe/models", "src/scribe/**/systems.py"]) unnamed = MagicMock(id=3, path_patterns=[]) with patch("scribe.services.systems.list_systems", AsyncMock(return_value=[api, data, unnamed])): from scribe.services.systems import systems_for_paths out = await systems_for_paths(1, 5, [ "src/scribe/routes/systems.py", "./README.md", "", ]) assert [(s.id, paths) for s, paths in out] == [ (1, ["src/scribe/routes/systems.py"]), (2, ["src/scribe/routes/systems.py"]), ] @pytest.mark.asyncio async def test_systems_for_paths_with_no_paths_reads_nothing(): lister = AsyncMock(return_value=[]) with patch("scribe.services.systems.list_systems", lister): from scribe.services.systems import systems_for_paths assert await systems_for_paths(1, 5, ["", " "]) == [] lister.assert_not_awaited() @pytest.mark.asyncio async def test_update_refuses_a_bad_pattern_before_touching_the_row(): with patch("scribe.services.systems.async_session") as mock_cls: from scribe.services.systems import update_system with pytest.raises(ValueError): await update_system(1, 9, path_patterns=["../elsewhere"]) mock_cls.assert_not_called() @pytest.mark.asyncio async def test_update_with_an_empty_list_clears_the_patterns(): """`[]` is a value, not "leave unchanged" — the service skips only None.""" system = MagicMock(deleted_at=None, project_id=5, path_patterns=["src/old"]) session = make_mock_session() session.get = AsyncMock(return_value=system) with patch("scribe.services.systems.async_session", return_value=session), \ patch("scribe.services.systems.access") as acc, \ patch("scribe.services.systems.embed_system"): acc.can_write_project = AsyncMock(return_value=True) from scribe.services.systems import update_system await update_system(1, 9, path_patterns=[]) assert system.path_patterns == [] @pytest.mark.asyncio async def test_create_stores_tidied_patterns(): mock_session = make_mock_session() captured = {} mock_session.add = MagicMock(side_effect=lambda obj: captured.update( patterns=getattr(obj, "path_patterns", "MISSING"))) with patch("scribe.services.systems.async_session", return_value=mock_session), \ patch("scribe.services.systems.access") as acc, \ patch("scribe.services.systems.embed_system"): acc.can_write_project = AsyncMock(return_value=True) from scribe.services.systems import create_system await create_system(1, 5, "Billing", path_patterns=["./src/billing/"]) assert captured["patterns"] == ["src/billing"] def test_system_to_dict_carries_its_patterns(): from scribe.models.system import System system = System(id=1, user_id=1, project_id=5, name="Billing", path_patterns=["src/billing"]) assert system.to_dict()["path_patterns"] == ["src/billing"] assert System(id=2, user_id=1, project_id=5, name="X").to_dict()["path_patterns"] == []