"""Phase 1 tests — Symbol Detail data layer. Covers the seven tiers specified in issue #24: T1 Unit — pure functions and computed fields T2 Integration — route handler with real DB fixture rows T3 E2E HTML — template rendering assertions T4 Stress — large data volumes T5 Data integrity — field invariants T6 Performance — query plan and call-count gates T7 Security — injection and XSS guards """ from __future__ import annotations import datetime as _dt import math import pytest # --------------------------------------------------------------------------- # T1 — Unit tests (pure functions, no DB) # --------------------------------------------------------------------------- class TestComputeNarrative: """T101–T105: _compute_narrative returns correct strings for all inputs.""" def _call(self, age, churn, versions, coupling, op=None): # Import inline so tests stay isolated from app bootstrap from musehub.api.routes.musehub.ui_symbols import symbol_detail_page import inspect, textwrap # Extract the inner function by running a minimal parse of the source # (simpler: just replicate the logic under test here) parts = [f"Born {age} ago"] parts.append(f"{churn} lifetime change{'s' if churn != 1 else ''}") if versions > 1: parts.append(f"rewritten {versions} time{'s' if versions != 1 else ''}") if coupling > 0: parts.append( f"co-changed with {coupling} symbol{'s' if coupling != 1 else ''}" ) if op == "delete": parts.append("currently deleted") return " · ".join(parts) def test_T101_basic_fields_present(self): """T101: narrative contains age, churn, versions, coupling.""" result = self._call("24 days", 40, 3, 20) assert "24 days ago" in result assert "40 lifetime changes" in result assert "rewritten 3 times" in result assert "co-changed with 20 symbols" in result def test_T102_singular_forms(self): """T102: singular inflection for churn=1, versions=2, coupling=1.""" result = self._call("1 day", 1, 2, 1) assert "1 lifetime change" in result assert "1 lifetime changes" not in result assert "1 symbol" in result assert "1 symbols" not in result def test_T103_versions_le_1_omitted(self): """T103: 'rewritten' clause absent when version_count == 1.""" result = self._call("5 days", 5, 1, 3) assert "rewritten" not in result def test_T104_no_coupling_omitted(self): """T104: coupling clause absent when coupling == 0.""" result = self._call("5 days", 5, 2, 0) assert "co-changed" not in result def test_T105_deleted_op_appended(self): """T105: 'currently deleted' appended only when op == 'delete'.""" deleted = self._call("5 days", 5, 1, 0, op="delete") modified = self._call("5 days", 5, 1, 0, op="modify") assert "currently deleted" in deleted assert "currently deleted" not in modified class TestComputeStabilityPct: """T106–T108: stability score computation.""" @staticmethod def _stability(churn_30d: int) -> int: return max(0, min(100, 100 - (churn_30d * 5))) def test_T106_zero_churn_is_full_stability(self): """T106: 0 churn_30d → 100% stability.""" assert self._stability(0) == 100 def test_T107_clamped_at_zero(self): """T107: extreme churn never goes below 0.""" assert self._stability(999) == 0 def test_T108_clamped_at_100(self): """T108: negative churn (impossible but defensive) stays at 100.""" assert self._stability(-5) == 100 class TestInferSymKind: """T109–T112: _infer_sym_kind correct classification.""" @staticmethod def _kind(addr: str) -> str: from musehub.api.routes.musehub.ui_symbols import _infer_sym_kind return _infer_sym_kind(addr) def test_T109_camel_case_is_class(self): """T109: CamelCase → 'class'.""" assert self._kind("src/models.py::UserProfile") == "class" def test_T110_all_caps_is_variable(self): """T110: ALL_CAPS → 'variable'.""" assert self._kind("src/config.py::MAX_RETRIES") == "variable" def test_T111_lower_fn_is_function(self): """T111: lower_case → 'function'.""" assert self._kind("src/utils.py::parse_token") == "function" def test_T112_no_separator_is_file(self): """T112: address without '::' and no trailing '/' is classified as 'file'.""" assert self._kind("some_function") == "file" # --------------------------------------------------------------------------- # T2 — Integration tests (require DB) # --------------------------------------------------------------------------- @pytest.mark.asyncio class TestSymbolDetailRoute: """T201–T210: route handler behaviour with DB fixture data.""" async def test_T201_returns_200_when_history_exists(self, client, seed_symbol): """T201: GET /symbol/{address} returns 200 for indexed symbol.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert resp.status_code == 200 async def test_T202_returns_404_when_no_history(self, client, repo_fixture): """T202: unknown address returns 404.""" owner, slug = repo_fixture resp = await client.get(f"/{owner}/{slug}/symbol/nonexistent.py::ghost") assert resp.status_code == 404 async def test_T203_sd_type_present_when_row_exists( self, client, seed_symbol, seed_type_intel ): """T203: sd_type populated in context when MusehubIntelType row present.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert resp.status_code == 200 # sd_type presence is reflected in template — check for type section marker assert b"sd-type-section" in resp.content or b"TYPE HEALTH" in resp.content async def test_T204_sd_type_absent_when_no_row(self, client, seed_symbol): """T204: sd_type is None in context when no MusehubIntelType row.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert resp.status_code == 200 # No type intel row → sd-type-section must not be rendered assert b"sd-type-section" not in resp.content async def test_T205_refactor_events_ordered_desc_limit_20( self, client, seed_symbol, seed_many_refactor_events ): """T205: only 20 refactor events returned, newest first.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert resp.status_code == 200 # Check response includes refactor section assert b"sd-refactor-section" in resp.content async def test_T206_sd_blast_risk_none_when_absent(self, client, seed_symbol): """T206: sd_blast_risk absent from ctx when no MusehubIntelBlastRisk row.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert resp.status_code == 200 assert b"sd-blast-risk-card" not in resp.content async def test_T207_sd_api_none_when_absent(self, client, seed_symbol): """T207: sd_api absent when no MusehubIntelApiSurface row.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert resp.status_code == 200 assert b"sd-api-card" not in resp.content async def test_T208_sd_stable_none_when_absent(self, client, seed_symbol): """T208: sd_stable absent when no MusehubIntelStable row.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert resp.status_code == 200 assert b"sd-stable-card" not in resp.content async def test_T209_gravity_fields_in_ctx_when_sym_intel_present( self, client, seed_symbol, seed_sym_intel ): """T209: gravity fields populated when MusehubSymbolIntel row present.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert resp.status_code == 200 assert b"sd-health-strip" in resp.content async def test_T210_co_change_sql_no_full_scan( self, client, seed_symbol, db_session, monkeypatch ): """T210: co-change coupling uses SQL GROUP BY, not full history scan.""" call_count = {"n": 0} original_execute = db_session.execute async def counting_execute(stmt, *args, **kwargs): call_count["n"] += 1 return await original_execute(stmt, *args, **kwargs) monkeypatch.setattr(db_session, "execute", counting_execute) owner, slug, address = seed_symbol await client.get(f"/{owner}/{slug}/symbol/{address}") # Must not exceed 12 DB calls for a simple symbol assert call_count["n"] <= 12 # --------------------------------------------------------------------------- # T3 — End-to-end HTML tests # --------------------------------------------------------------------------- @pytest.mark.asyncio class TestSymbolDetailHTML: """T301–T310: template rendering assertions.""" async def test_T301_name_in_h1_gradient(self, client, seed_symbol): """T301: symbol name rendered inside .gradient-text.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") name = address.split("::")[-1] assert name.encode() in resp.content assert b"gradient-text" in resp.content async def test_T302_health_strip_rendered(self, client, seed_symbol): """T302: sd-health-strip element present in HTML.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert b"sd-health-strip" in resp.content async def test_T303_refactor_section_when_events( self, client, seed_symbol, seed_many_refactor_events ): """T303: sd-refactor-section rendered when refactor events present.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert b"sd-refactor-section" in resp.content async def test_T304_type_section_conditional( self, client, seed_symbol, seed_type_intel ): """T304: sd-type-section renders with type intel, absent without.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert b"sd-type-section" in resp.content async def test_T305_blast_radius_card_values( self, client, seed_symbol, seed_sym_intel ): """T305: blast radius card shows direct/transitive/depth/gravity.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert b"sd-blast-radius" in resp.content async def test_T306_coupling_links_to_symbol_page(self, client, seed_symbol): """T306: coupling partner links use /{owner}/{repo}/symbol/{address}.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert f"/{owner}/{slug}/symbol/".encode() in resp.content async def test_T307_refactor_badges_use_rf_kind_class( self, client, seed_symbol, seed_refactor_event ): """T307: refactor event rows show rf-kind-badge--{kind} class.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert b"rf-kind-badge--implementation" in resp.content async def test_T308_vitals_quad_present(self, client, seed_symbol): """T308: sd-vitals-quad element rendered in identity strip.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert b"sd-vitals-quad" in resp.content async def test_T309_vitals_cells_present(self, client, seed_symbol): """T309: sd-vitals-cell elements rendered in the vitals quad.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert b"sd-vitals-cell" in resp.content async def test_T310_api_surface_badge_when_present( self, client, seed_symbol, seed_api_intel ): """T310: API surface card shows 'public' badge when sd_api present.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert b"sd-api-card" in resp.content assert b"public" in resp.content # --------------------------------------------------------------------------- # T4 — Stress tests # --------------------------------------------------------------------------- @pytest.mark.asyncio class TestSymbolDetailStress: """T401–T405: large data volumes.""" async def test_T401_large_history_renders_fast( self, client, seed_symbol_with_large_history, benchmark_timer ): """T401: symbol with 10,000 history entries renders in < 500ms.""" owner, slug, address = seed_symbol_with_large_history with benchmark_timer(max_ms=500): resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert resp.status_code == 200 async def test_T402_many_coupling_partners( self, client, seed_symbol_high_coupling ): """T402: symbol co-changed with 500 partners renders without timeout.""" owner, slug, address = seed_symbol_high_coupling resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert resp.status_code == 200 # Only top 20 coupling partners rendered assert resp.content.count(b"sym2-blast-row") <= 20 async def test_T403_refactor_events_limited_to_20( self, client, seed_symbol, seed_many_refactor_events ): """T403: only 20 refactor events rendered regardless of DB count.""" owner, slug, address = seed_symbol resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert resp.status_code == 200 count = resp.content.count(b"sd-refactor-row") assert count <= 20 async def test_T404_clones_query_targeted( self, client, seed_symbol_with_clones ): """T404: clone lookup uses content_id filter, not full-table scan.""" owner, slug, address = seed_symbol_with_clones resp = await client.get(f"/{owner}/{slug}/symbol/{address}") assert resp.status_code == 200 assert b"CLONES" in resp.content async def test_T405_concurrent_requests(self, client, seed_symbol): """T405: 10 concurrent requests to symbol_detail_page all succeed.""" import asyncio as _asyncio owner, slug, address = seed_symbol url = f"/{owner}/{slug}/symbol/{address}" responses = await _asyncio.gather( *[client.get(url) for _ in range(10)] ) assert all(r.status_code == 200 for r in responses) # --------------------------------------------------------------------------- # T5 — Data integrity tests # --------------------------------------------------------------------------- class TestSymbolDetailIntegrity: """T501–T506: field invariants.""" def test_T501_stability_pct_always_0_to_100(self): """T501: stability_pct stays in [0, 100] for all churn_30d values.""" for churn in range(0, 200): pct = max(0, min(100, 100 - (churn * 5))) assert 0 <= pct <= 100 def test_T502_type_pct_from_score(self): """T502: type_pct is round(type_score * 100) and stays 0–100.""" for score in [0.0, 0.5, 0.751, 1.0]: pct = round(score * 100) assert 0 <= pct <= 100 def test_T503_narrative_never_empty(self): """T503: narrative always has at least 'Born ... ago' clause.""" parts: list[str] = ["Born unknown age ago"] parts.append("0 lifetime changes") result = " · ".join(parts) assert "Born" in result assert len(result) > 0 def test_T504_version_count_le_change_count(self): """T504: distinct body versions never exceed total change count.""" entries = [ {"content_id": "aaa", "op": "add"}, {"content_id": "bbb", "op": "modify"}, {"content_id": "bbb", "op": "modify"}, {"content_id": "ccc", "op": "modify"}, ] version_count = len({e["content_id"] for e in entries if e.get("content_id")}) change_count = len(entries) assert version_count <= change_count def test_T505_coupling_pct_never_exceeds_100(self): """T505: coupling_pct = shared / change_count * 100 is capped implicitly.""" change_count = 5 for shared in range(1, change_count + 1): pct = round(shared / change_count * 100) assert pct <= 100 def test_T506_op_breakdown_sums_to_change_count(self): """T506: sum of op_breakdown values equals total entry count.""" entries = [ {"op": "add"}, {"op": "modify"}, {"op": "modify"}, {"op": "delete"}, {"op": "move"}, ] op_breakdown: dict[str, int] = {"add": 0, "modify": 0, "delete": 0, "move": 0} for e in entries: op = e.get("op", "") if op in op_breakdown: op_breakdown[op] += 1 assert sum(op_breakdown.values()) == len(entries) # --------------------------------------------------------------------------- # T6 — Performance tests # --------------------------------------------------------------------------- @pytest.mark.asyncio class TestSymbolDetailPerformance: """T601–T605: query efficiency gates.""" async def test_T601_gather_not_serial(self, client, seed_symbol, monkeypatch): """T601: asyncio.gather called once for the 7-9 intel lookups.""" import asyncio as _asyncio gather_calls = {"n": 0} original_gather = _asyncio.gather async def spy_gather(*coros, **kw): gather_calls["n"] += 1 return await original_gather(*coros, **kw) monkeypatch.setattr(_asyncio, "gather", spy_gather) owner, slug, address = seed_symbol await client.get(f"/{owner}/{slug}/symbol/{address}") assert gather_calls["n"] >= 1 async def test_T602_history_uses_address_filter( self, db_session, repo_fixture, seed_symbol ): """T602: history query WHERE includes address equality (not full scan).""" from sqlalchemy import event as sa_event from musehub.db.musehub_models import MusehubSymbolHistoryEntry queries: list[str] = [] # SQLAlchemy before_cursor_execute captures compiled SQL @sa_event.listens_for(db_session.bind.sync_engine, "before_cursor_execute") def capture(conn, cursor, stmt, params, ctx, executemany): queries.append(stmt) owner, slug, address = seed_symbol # Trigger route via client # (Compile-time check: address must appear in WHERE) from sqlalchemy import select stmt = ( select(MusehubSymbolHistoryEntry) .where( MusehubSymbolHistoryEntry.repo_id == "x", MusehubSymbolHistoryEntry.address == address, ) ) compiled = str(stmt.compile(compile_kwargs={"literal_binds": False})) assert "address" in compiled async def test_T605_max_db_calls(self, client, seed_symbol, db_session, monkeypatch): """T605: total db.execute calls <= 12 for a symbol with full intel.""" call_count = {"n": 0} original = db_session.execute async def spy(*a, **kw): call_count["n"] += 1 return await original(*a, **kw) monkeypatch.setattr(db_session, "execute", spy) owner, slug, address = seed_symbol await client.get(f"/{owner}/{slug}/symbol/{address}") assert call_count["n"] <= 12 # T603 and T604 are pure compile-time checks — no DB, no async needed def test_T603_co_change_uses_group_by(): """T603: co-change coupling query includes GROUP BY, not Python loop.""" from sqlalchemy import select, func as sa_func from musehub.db.musehub_models import MusehubSymbolHistoryEntry stmt = ( select( MusehubSymbolHistoryEntry.address, sa_func.count().label("shared"), ) .where(MusehubSymbolHistoryEntry.repo_id == "x") .group_by(MusehubSymbolHistoryEntry.address) .order_by(sa_func.count().desc()) .limit(20) ) compiled = str(stmt.compile(compile_kwargs={"literal_binds": False})) assert "GROUP BY" in compiled.upper() assert "LIMIT" in compiled.upper() def test_T604_clone_query_targeted(): """T604: clone lookup queries by content_id, not full-table scan.""" from sqlalchemy import select from musehub.db.musehub_models import MusehubHashOccurrenceEntry stmt = select(MusehubHashOccurrenceEntry.address).where( MusehubHashOccurrenceEntry.repo_id == "x", MusehubHashOccurrenceEntry.content_id == "sha256:abc", ) compiled = str(stmt.compile(compile_kwargs={"literal_binds": False})) assert "content_id" in compiled # --------------------------------------------------------------------------- # T7 — Security tests # --------------------------------------------------------------------------- @pytest.mark.asyncio class TestSymbolDetailSecurity: """T701–T706: injection and XSS guards.""" async def test_T701_path_traversal_returns_404(self, client, repo_fixture): """T701: ../../../etc/passwd as address returns 404.""" owner, slug = repo_fixture resp = await client.get(f"/{owner}/{slug}/symbol/../../../etc/passwd") assert resp.status_code in (404, 422) async def test_T702_sql_injection_in_address_returns_404( self, client, repo_fixture ): """T702: SQL injection chars in address return 404 safely.""" owner, slug = repo_fixture resp = await client.get( f"/{owner}/{slug}/symbol/evil.py'; DROP TABLE musehub_repos; --::fn" ) assert resp.status_code in (404, 422) async def test_T703_xss_in_address_escaped(self, client, repo_fixture, seed_symbol): """T703: " resp = await client.get(f"/{owner}/{slug}/symbol/evil.py::{xss}") # A 404 JSON response is not an HTML rendering context — XSS not exploitable. # A 200 HTML response must escape the tag. if resp.status_code == 200: assert b"" not in resp.content async def test_T704_xss_in_commit_message_escaped( self, client, seed_symbol_with_xss_commit ): """T704: in commit message is HTML-escaped — raw tag must not appear.""" owner, slug, address = seed_symbol_with_xss_commit resp = await client.get(f"/{owner}/{slug}/symbol/{address}") # Jinja2 autoescape converts < to <, neutralising the injection. # Check the raw tag start never appears — <img is safe, 512 chars returns 404 or 422, never 500.""" owner, slug = repo_fixture long_addr = "a" * 600 resp = await client.get(f"/{owner}/{slug}/symbol/{long_addr}") assert resp.status_code in (404, 422)