"""Tests for Signal 2: symbol anchor overlap via touched_symbols on proposals. Covers: - _symbols_from_delta extracts correct symbol addresses - touched_symbols populated at create_proposal time from existing branch commits - touched_symbols refreshed at merge_proposal time - find_proposals_by_symbol_overlap returns match when anchors intersect - find_proposals_by_symbol_overlap returns empty when no intersection - empty symbol_anchors returns empty list immediately - cross-repo isolation """ from __future__ import annotations import secrets from datetime import datetime, timezone from typing import TypedDict import pytest from sqlalchemy.ext.asyncio import AsyncSession from muse.core.types import fake_id, now_utc_iso from musehub.core.genesis import compute_branch_id, compute_identity_id, compute_proposal_id, compute_repo_id from musehub.db.musehub_repo_models import MusehubBranch, MusehubCommit, MusehubCommitRef, MusehubRepo from musehub.db.musehub_social_models import MusehubProposal from musehub.services import musehub_issues from musehub.services.musehub_proposals import ( _symbols_from_delta, _touched_symbols_for_branch, create_proposal, merge_proposal, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- class _ChildOp(TypedDict): op: str address: str content_summary: str class _DeltaOp(TypedDict): address: str child_ops: list[_ChildOp] class _Delta(TypedDict): ops: list[_DeltaOp] def _uid() -> str: return secrets.token_hex(16) def _commit_id() -> str: return fake_id(_uid()) def _delta(*symbol_addresses: str) -> _Delta: """Build a minimal structured_delta containing the given symbol addresses.""" return _Delta( ops=[ _DeltaOp( address=addr.split("::")[0], child_ops=[_ChildOp(op="update", address=addr, content_summary="function")], ) for addr in symbol_addresses ] ) async def _make_repo(db: AsyncSession, slug: str = "sym-test") -> str: created_at = datetime.now(tz=timezone.utc) owner_id = compute_identity_id(b"testuser") repo_id = compute_repo_id(owner_id, slug, "code", created_at.isoformat()) repo = MusehubRepo( repo_id=repo_id, name=slug, owner="testuser", slug=slug, visibility="public", owner_user_id=owner_id, created_at=created_at, updated_at=created_at, ) db.add(repo) await db.commit() await db.refresh(repo) return str(repo.repo_id) async def _make_commit( db: AsyncSession, repo_id: str, *, branch: str, symbol_addresses: list[str] | None = None, commit_id: str | None = None, ) -> str: """Seed a commit with an optional structured_delta and return its commit_id.""" cid = commit_id or _commit_id() row = MusehubCommit( commit_id=cid, branch=branch, parent_ids=[], message="test commit", author="tester", timestamp=datetime.now(timezone.utc), structured_delta=_delta(*symbol_addresses) if symbol_addresses else None, ) db.add(row) db.add(MusehubCommitRef(repo_id=repo_id, commit_id=cid)) await db.flush() return cid async def _make_branch( db: AsyncSession, repo_id: str, name: str, head_commit_id: str | None = None ) -> None: """Seed a branch record.""" branch = MusehubBranch( branch_id=compute_branch_id(repo_id, name), repo_id=repo_id, name=name, head_commit_id=head_commit_id, ) db.add(branch) await db.flush() # --------------------------------------------------------------------------- # Unit tests for _symbols_from_delta # --------------------------------------------------------------------------- def test_symbols_from_delta_extracts_symbol_addresses() -> None: delta = _delta( "musehub/services/musehub_issues.py::create_issue", "musehub/services/musehub_issues.py::get_issue", ) result = _symbols_from_delta(delta) assert "musehub/services/musehub_issues.py::create_issue" in result assert "musehub/services/musehub_issues.py::get_issue" in result assert len(result) == 2 def test_symbols_from_delta_skips_file_level_ops() -> None: """File-level ops without '::' in address must not appear in result.""" delta = { "ops": [ { "address": "musehub/services/musehub_issues.py", "child_ops": [], } ] } result = _symbols_from_delta(delta) assert result == [] def test_symbols_from_delta_handles_non_dict() -> None: assert _symbols_from_delta(None) == [] assert _symbols_from_delta("bad") == [] assert _symbols_from_delta({}) == [] def test_symbols_from_delta_deduplicates() -> None: delta = _delta( "musehub/services/foo.py::bar", "musehub/services/foo.py::bar", ) result = _symbols_from_delta(delta) assert result.count("musehub/services/foo.py::bar") == 1 # --------------------------------------------------------------------------- # Integration tests: touched_symbols populated at create / merge # --------------------------------------------------------------------------- async def test_touched_symbols_for_branch_extracts_from_commits( db_session: AsyncSession, ) -> None: repo_id = await _make_repo(db_session, "ts-branch-extract") await _make_commit( db_session, repo_id, branch="feat/s2", symbol_addresses=["a/b.py::foo", "a/b.py::bar"], ) await _make_commit( db_session, repo_id, branch="feat/s2", symbol_addresses=["a/c.py::baz"], ) await db_session.commit() result = await _touched_symbols_for_branch(db_session, repo_id, "feat/s2") assert "a/b.py::foo" in result assert "a/b.py::bar" in result assert "a/c.py::baz" in result assert len(result) == 3 async def test_create_proposal_populates_touched_symbols( db_session: AsyncSession, ) -> None: repo_id = await _make_repo(db_session, "ts-create") head_cid = await _make_commit( db_session, repo_id, branch="feat/create-signal", symbol_addresses=["musehub/services/x.py::MyFunc"], ) await _make_branch(db_session, repo_id, "feat/create-signal", head_cid) await _make_branch(db_session, repo_id, "main", head_cid) await db_session.commit() proposal = await create_proposal( db_session, repo_id=repo_id, title="Test proposal", from_branch="feat/create-signal", to_branch="main", ) await db_session.commit() # Fetch the raw ORM row to verify the column was written. from sqlalchemy import select as _select row = (await db_session.execute( _select(MusehubProposal).where(MusehubProposal.proposal_id == proposal.proposal_id) )).scalar_one() assert "musehub/services/x.py::MyFunc" in (row.touched_symbols or []) async def test_merge_proposal_refreshes_touched_symbols( db_session: AsyncSession, ) -> None: """touched_symbols at merge time includes any new commits added after create.""" repo_id = await _make_repo(db_session, "ts-merge") initial_cid = await _make_commit( db_session, repo_id, branch="feat/refresh", symbol_addresses=["svc/old.py::OldFunc"], ) await _make_branch(db_session, repo_id, "feat/refresh", initial_cid) to_cid = await _make_commit(db_session, repo_id, branch="main") await _make_branch(db_session, repo_id, "main", to_cid) await db_session.commit() proposal = await create_proposal( db_session, repo_id=repo_id, title="Refresh test", from_branch="feat/refresh", to_branch="main", ) await db_session.commit() # Push a new commit to the feature branch after proposal creation. new_cid = await _make_commit( db_session, repo_id, branch="feat/refresh", symbol_addresses=["svc/new.py::NewFunc"], ) # Update the branch head. from sqlalchemy import select as _select branch_row = (await db_session.execute( _select(MusehubBranch).where( MusehubBranch.repo_id == repo_id, MusehubBranch.name == "feat/refresh" ) )).scalar_one() branch_row.head_commit_id = new_cid await db_session.flush() await db_session.commit() await merge_proposal(db_session, repo_id, proposal.proposal_id) await db_session.commit() row = (await db_session.execute( _select(MusehubProposal).where(MusehubProposal.proposal_id == proposal.proposal_id) )).scalar_one() touched = row.touched_symbols or [] assert "svc/old.py::OldFunc" in touched assert "svc/new.py::NewFunc" in touched # --------------------------------------------------------------------------- # Integration tests: find_proposals_by_symbol_overlap # --------------------------------------------------------------------------- async def test_symbol_overlap_returns_match(db_session: AsyncSession) -> None: repo_id = await _make_repo(db_session, "overlap-match") # Manually seed a proposal with a known touched_symbols. author_id = compute_identity_id(b"tester") pid = compute_proposal_id(repo_id, author_id, "feat/fix", "main", now_utc_iso()) row = MusehubProposal( proposal_id=pid, repo_id=repo_id, proposal_number=1, title="Fix create_issue bug", body="", state="merged", from_branch="feat/fix", to_branch="main", author="tester", touched_symbols=["musehub/services/musehub_issues.py::create_issue"], ) db_session.add(row) await db_session.commit() results = await musehub_issues.find_proposals_by_symbol_overlap( db_session, repo_id, ["musehub/services/musehub_issues.py::create_issue"], ) assert len(results) == 1 assert results[0]["proposal_id"] == pid assert results[0]["state"] == "merged" assert results[0]["match_reason"] == "symbol_overlap" async def test_symbol_overlap_no_match(db_session: AsyncSession) -> None: repo_id = await _make_repo(db_session, "overlap-no-match") author_id = compute_identity_id(b"tester") pid = compute_proposal_id(repo_id, author_id, "feat/unrelated", "main", now_utc_iso()) row = MusehubProposal( proposal_id=pid, repo_id=repo_id, proposal_number=1, title="Unrelated proposal", body="", state="merged", from_branch="feat/unrelated", to_branch="main", author="tester", touched_symbols=["musehub/services/other.py::some_fn"], ) db_session.add(row) await db_session.commit() results = await musehub_issues.find_proposals_by_symbol_overlap( db_session, repo_id, ["musehub/services/musehub_issues.py::create_issue"], ) assert results == [] async def test_symbol_overlap_empty_anchors_returns_empty( db_session: AsyncSession, ) -> None: repo_id = await _make_repo(db_session, "overlap-empty") await db_session.commit() results = await musehub_issues.find_proposals_by_symbol_overlap( db_session, repo_id, [] ) assert results == [] async def test_symbol_overlap_cross_repo_isolation(db_session: AsyncSession) -> None: repo_a = await _make_repo(db_session, "overlap-repo-a") repo_b = await _make_repo(db_session, "overlap-repo-b") author_id = compute_identity_id(b"tester") pid = compute_proposal_id(repo_a, author_id, "feat/a", "main", now_utc_iso()) row = MusehubProposal( proposal_id=pid, repo_id=repo_a, proposal_number=1, title="Proposal in repo A", body="", state="merged", from_branch="feat/a", to_branch="main", author="tester", touched_symbols=["musehub/services/musehub_issues.py::create_issue"], ) db_session.add(row) await db_session.commit() # Query against repo_b — must return nothing. results = await musehub_issues.find_proposals_by_symbol_overlap( db_session, repo_b, ["musehub/services/musehub_issues.py::create_issue"], ) assert results == [] async def test_symbol_overlap_open_proposal_matched(db_session: AsyncSession) -> None: """Open proposals with matching touched_symbols are returned.""" repo_id = await _make_repo(db_session, "overlap-open") author_id = compute_identity_id(b"tester") pid = compute_proposal_id(repo_id, author_id, "feat/in-progress", "main", now_utc_iso()) row = MusehubProposal( proposal_id=pid, repo_id=repo_id, proposal_number=1, title="In-progress fix", body="", state="open", from_branch="feat/in-progress", to_branch="main", author="tester", touched_symbols=["musehub/api/routes/musehub/ui_issues.py::issue_detail_page"], ) db_session.add(row) await db_session.commit() results = await musehub_issues.find_proposals_by_symbol_overlap( db_session, repo_id, ["musehub/api/routes/musehub/ui_issues.py::issue_detail_page"], ) assert len(results) == 1 assert results[0]["proposal_id"] == pid assert results[0]["state"] == "open"