bvandeusenandClaude Opus 5.5
b8efff43b4
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 13s
CI & Build / TypeScript typecheck (push) Successful in 56s
CI & Build / integration (push) Successful in 1m8s
CI & Build / Python tests (push) Successful in 2m4s
CI & Build / Build & push image (push) Successful in 34s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
179 lines
8.2 KiB
Python
179 lines
8.2 KiB
Python
"""A family idea reaches every project on its platform, and no other
|
|
(milestone 463 step 6).
|
|
|
|
WHY THIS IS AN INTEGRATION TEST — the lesson-reach reasoning (#3730) holds
|
|
here too: the reach is one `OR` inside the search's project filter, and only
|
|
the ROWS that come back prove it. Every note embeds identically to the query,
|
|
so scoping is the only thing that can separate them. The embedder is stubbed;
|
|
no similarity is asserted, only membership.
|
|
|
|
The corpus: an idea written in android one, canon for the Android platform,
|
|
with a Kotlin and a Python reference. Android two writes Python, so it should
|
|
reach the idea and the Python reference only. The Go service is on no shared
|
|
platform and should reach none of it.
|
|
"""
|
|
import uuid
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
|
|
from scribe.models import async_session
|
|
from scribe.models.embedding import EMBEDDING_DIM, NoteEmbedding
|
|
from scribe.models.family import (
|
|
FamilyAdoption, FamilyDecision, FamilyIdea, FamilyIdeaPlatform, FamilyIdeaReference,
|
|
Platform, ProjectPlatform,
|
|
)
|
|
from scribe.models.note import Note
|
|
from scribe.models.project import Project
|
|
from scribe.services import family_adoption as adoption_svc
|
|
from scribe.services.embeddings import CHUNKER_VERSION, EMBEDDING_MODEL, semantic_search_notes
|
|
from tests.helpers import ensure_user
|
|
|
|
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
|
|
|
|
QUERY_VEC = [1.0] + [0.0] * (EMBEDDING_DIM - 1)
|
|
|
|
|
|
async def _platform(s, slug: str) -> int:
|
|
return await s.scalar(select(Platform.id).where(
|
|
Platform.slug == slug, Platform.deleted_at.is_(None)))
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def corpus():
|
|
tag = uuid.uuid4().hex[:8]
|
|
async with async_session() as s:
|
|
owner = await ensure_user(s, f"family_reach_owner_{tag}")
|
|
await s.flush()
|
|
android, go = await _platform(s, "android-app"), await _platform(s, "go")
|
|
a = Project(user_id=owner.id, title="android one")
|
|
b = Project(user_id=owner.id, title="android two")
|
|
c = Project(user_id=owner.id, title="go service")
|
|
s.add_all([a, b, c])
|
|
await s.flush()
|
|
s.add_all([
|
|
ProjectPlatform(project_id=a.id, platform_id=android, state="declared"),
|
|
ProjectPlatform(project_id=b.id, platform_id=android, state="declared"),
|
|
ProjectPlatform(project_id=c.id, platform_id=go, state="declared"),
|
|
])
|
|
|
|
def snippet(project, title, language):
|
|
return Note(user_id=owner.id, project_id=project.id, note_type="snippet",
|
|
title=title, body=title, data={"language": language})
|
|
|
|
rows = {
|
|
"idea": Note(user_id=owner.id, project_id=a.id, note_type="note",
|
|
title="Signed APK lane", body="one keystore, in-place update"),
|
|
"kotlin_ref": snippet(a, "installUpdate (kotlin)", "kotlin"),
|
|
"python_ref": snippet(a, "install_update (python)", "python"),
|
|
"note_on_a": Note(user_id=owner.id, project_id=a.id, note_type="note",
|
|
title="An ordinary note", body="ordinary"),
|
|
"b_own": snippet(b, "b's own python helper", "python"),
|
|
}
|
|
s.add_all(rows.values())
|
|
await s.flush()
|
|
s.add(FamilyIdea(note_id=rows["idea"].id, status="canon",
|
|
applies_when="any Android app that ships its own APK"))
|
|
await s.flush()
|
|
s.add_all([
|
|
FamilyIdeaPlatform(note_id=rows["idea"].id, platform_id=android),
|
|
FamilyIdeaReference(idea_id=rows["idea"].id, snippet_id=rows["kotlin_ref"].id),
|
|
FamilyIdeaReference(idea_id=rows["idea"].id, snippet_id=rows["python_ref"].id),
|
|
])
|
|
for note in rows.values():
|
|
s.add(NoteEmbedding(
|
|
note_id=note.id, chunk_index=0, user_id=owner.id,
|
|
embedding=QUERY_VEC, chunk_text=note.title,
|
|
chunker_version=CHUNKER_VERSION, embedding_model=EMBEDDING_MODEL,
|
|
))
|
|
ids = {k: n.id for k, n in rows.items()}
|
|
ids.update(owner=owner.id, a=a.id, b=b.id, c=c.id)
|
|
await s.commit()
|
|
return ids
|
|
|
|
|
|
def _family(corpus) -> set[int]:
|
|
"""This corpus's family records. The reach is computed across every
|
|
user's canon (the search applies readability afterwards), and the
|
|
integration database holds other tests' Android ideas, so assertions
|
|
about it are made on this corpus's ids only."""
|
|
return {corpus["idea"], corpus["kotlin_ref"], corpus["python_ref"]}
|
|
|
|
|
|
async def _search(uid, **kw):
|
|
with patch("scribe.services.embeddings.get_embedding", AsyncMock(return_value=QUERY_VEC)):
|
|
hits = await semantic_search_notes(uid, "how does the app update itself", limit=20, **kw)
|
|
return {note.id for _score, note in hits}
|
|
|
|
|
|
async def test_an_idea_reaches_a_project_on_its_platform_with_the_reference_in_its_language(corpus):
|
|
found = await _search(corpus["owner"], project_id=corpus["b"], include_global_kinds=True)
|
|
assert {corpus["idea"], corpus["python_ref"], corpus["b_own"]} <= found
|
|
# Android two writes Python: the Kotlin reference stays with the idea.
|
|
assert corpus["kotlin_ref"] not in found
|
|
assert corpus["note_on_a"] not in found
|
|
|
|
|
|
async def test_an_idea_does_not_reach_a_project_on_no_shared_platform(corpus):
|
|
found = await _search(corpus["owner"], project_id=corpus["c"], include_global_kinds=True)
|
|
assert not found & {corpus["idea"], corpus["kotlin_ref"], corpus["python_ref"]}
|
|
|
|
|
|
async def test_the_reach_is_off_unless_the_search_widens_the_project(corpus):
|
|
"""The near-duplicate gate searches a project without widening it; a
|
|
family idea there would block a create on a project it was never
|
|
written in."""
|
|
found = await _search(corpus["owner"], project_id=corpus["b"])
|
|
assert found == {corpus["b_own"]}
|
|
|
|
|
|
async def test_with_no_reference_in_the_projects_language_every_reference_comes(corpus):
|
|
async with async_session() as s:
|
|
b_own = await s.get(Note, corpus["b_own"])
|
|
b_own.data = {"language": "dart"}
|
|
await s.commit()
|
|
reach = await adoption_svc.family_reach_ids(corpus["b"])
|
|
assert reach & _family(corpus) == _family(corpus)
|
|
|
|
|
|
async def test_a_retired_idea_reaches_nobody(corpus):
|
|
async with async_session() as s:
|
|
(await s.get(FamilyIdea, corpus["idea"])).status = "retired"
|
|
await s.commit()
|
|
assert not await adoption_svc.family_reach_ids(corpus["b"]) & _family(corpus)
|
|
|
|
|
|
# --- the entry readout and the owed answers a close hands back -------------------
|
|
|
|
async def test_the_entry_readout_counts_what_the_project_has_to_answer(corpus):
|
|
line = await adoption_svc.family_readout(corpus["owner"], corpus["b"])
|
|
assert (line["unassessed"], line["owed"], line["needs_recheck"]) == (1, 0, 0)
|
|
assert line["calls"]["unassessed"] == (
|
|
f'list_family_adoptions(project_id={corpus["b"]}, status="unassessed")')
|
|
# Nothing reaches the Go service, so it carries no line at all.
|
|
assert await adoption_svc.family_readout(corpus["owner"], corpus["c"]) is None
|
|
|
|
|
|
async def test_owed_since_lists_what_is_still_owed_and_only_since_then(corpus):
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
async with async_session() as s:
|
|
s.add(FamilyAdoption(project_id=corpus["b"], idea_id=corpus["idea"], status="owed",
|
|
reason="no in-place update yet", canon_version=1,
|
|
decided_via="agent"))
|
|
s.add(FamilyDecision(idea_id=corpus["idea"], project_id=corpus["b"], action="assess",
|
|
reason="no in-place update yet",
|
|
before={"status": "unassessed", "reason": "", "canon_version": None},
|
|
after={"status": "owed", "reason": "no in-place update yet",
|
|
"canon_version": 1},
|
|
evidence={}, precedent_ids=[], decided_via="agent",
|
|
user_id=corpus["owner"]))
|
|
await s.commit()
|
|
now = datetime.now(timezone.utc)
|
|
[owed] = await adoption_svc.owed_since(corpus["owner"], now - timedelta(minutes=5))
|
|
assert (owed["idea_id"], owed["project_id"], owed["project_title"]) == (
|
|
corpus["idea"], corpus["b"], "android two")
|
|
assert await adoption_svc.owed_since(corpus["owner"], now + timedelta(minutes=5)) == []
|