"""Real-Postgres checks for family canon's schema (milestone 463 step 1). Two kinds of claim live here, and both need a database: 1. THE CONSTRAINTS. The agent assesses and promotes with no approval step, so the few things that must always hold are held by the schema rather than by prose a session might skip: a canon idea states when it applies, a variant or an exemption says why, a decision has a reason. Each is shown refusing the bad row, and each beside a good row so the refusal is not a fixture that fails for some other reason. 2. THE RESTORE REMAPS. Two seams can come back plausible and wrong: - platforms travel by SLUG and must land on the destination's own seeded rows, not create duplicates of them; - `precedent_ids` is a list of ids INSIDE JSON. A restore that copied it raw would leave each decision pointing at whatever took the old number — populated, plausible, and about the wrong decision. """ import pytest import pytest_asyncio from sqlalchemy import select from sqlalchemy.exc import IntegrityError from scribe.models import async_session 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.models.user import User from scribe.services import backup from tests.helpers import ensure_user pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")] OWNER_USERNAME = "family_canon_owner" RESTORED_USERNAME = "family_canon_restored" async def _purge(username: str) -> None: """user -> project / note is ON DELETE CASCADE, and every family table cascades from a project or a note, so dropping the user's rows clears everything this file made. Platforms are global and never created here beyond the migration's seed, so there is nothing of theirs to clear.""" async with async_session() as s: for user in (await s.execute( select(User).where(User.username == username) )).scalars().all(): for note in (await s.execute( select(Note).where(Note.user_id == user.id) )).scalars().all(): await s.delete(note) for project in (await s.execute( select(Project).where(Project.user_id == user.id) )).scalars().all(): await s.delete(project) if username == RESTORED_USERNAME: await s.delete(user) await s.commit() @pytest_asyncio.fixture(autouse=True) async def _no_leftovers(): """SETUP ONLY — a database call after a `yield` in an autouse fixture orphans a pooled connection (see the backup round-trip siblings).""" await _purge(RESTORED_USERNAME) await _purge(OWNER_USERNAME) async def _platform(slug: str) -> Platform: async with async_session() as s: return (await s.execute( select(Platform).where(Platform.slug == slug, Platform.deleted_at.is_(None)) )).scalars().one() async def _owner_project_and_note() -> tuple[int, int, int]: async with async_session() as s: owner = await ensure_user(s, OWNER_USERNAME) project = Project(user_id=owner.id, title="an app on a platform") note = Note(user_id=owner.id, title="an idea", body="the idea") s.add_all([project, note]) await s.commit() return owner.id, project.id, note.id async def _refused(*rows) -> bool: async with async_session() as s: s.add_all(list(rows)) try: await s.commit() except IntegrityError: await s.rollback() return True return False # --- the seed --------------------------------------------------------------- async def test_the_migration_seeds_generic_platforms_with_their_markers(): android = await _platform("android-app") assert "AndroidManifest.xml" in android.markers # Declare-only platforms carry an empty list, never NULL: detection reads # the list and must not have to special-case a missing one. assert (await _platform("postgresql")).markers == [] # --- the constraints ---------------------------------------------------------- async def test_a_canon_idea_must_say_when_it_applies(): _, _, note_id = await _owner_project_and_note() assert await _refused(FamilyIdea(note_id=note_id, status="canon", applies_when=" ")) # A candidate may be vague — that is what a candidate is. assert not await _refused(FamilyIdea(note_id=note_id, status="candidate")) async def test_a_variant_or_an_exemption_must_say_why(): _, project_id, note_id = await _owner_project_and_note() assert not await _refused(FamilyIdea( note_id=note_id, status="canon", applies_when="any app that installs its own updates", )) for status in ("variant", "exempt"): assert await _refused(FamilyAdoption( project_id=project_id, idea_id=note_id, status=status, reason=" ", )), f"{status} with a blank reason was accepted" # Owed and adopted need no reason: neither is a departure. assert not await _refused(FamilyAdoption( project_id=project_id, idea_id=note_id, status="owed", decided_via="agent", )) async def test_an_unknown_state_is_refused(): _, project_id, note_id = await _owner_project_and_note() assert not await _refused(FamilyIdea(note_id=note_id)) android = await _platform("android-app") assert await _refused(ProjectPlatform( project_id=project_id, platform_id=android.id, state="maybe", )) assert await _refused(FamilyAdoption( project_id=project_id, idea_id=note_id, status="declined", )) async def test_a_decision_must_carry_a_reason(): _, _, note_id = await _owner_project_and_note() assert not await _refused(FamilyIdea(note_id=note_id)) assert await _refused(FamilyDecision(idea_id=note_id, action="propose", reason=" ")) assert not await _refused(FamilyDecision( idea_id=note_id, action="propose", reason="built twice, in two projects", )) # --- the restore ------------------------------------------------------------ @pytest_asyncio.fixture async def source(): """An idea with everything family canon can hang off it: a platform scope, a reference snippet, a project that declares one platform and rejects another, a variant answer, and two decisions where the second cites the first as its precedent. `decoy` exists so the target database has an ADDITIONAL decision whose id could collide with a source precedent id — without it, a raw-copied precedent list could happen to point at nothing and read as dropped, rather than as pointing at the wrong decision. """ owner_id, project_id, idea_id = await _owner_project_and_note() android = await _platform("android-app") container = await _platform("container-image") async with async_session() as s: snippet = Note(user_id=owner_id, title="the reference", body="code", note_type="snippet") s.add(snippet) s.add(FamilyIdea(note_id=idea_id, status="canon", canon_version=2, applies_when="any app that installs its own updates")) await s.flush() s.add_all([ FamilyIdeaPlatform(note_id=idea_id, platform_id=android.id), FamilyIdeaReference(idea_id=idea_id, snippet_id=snippet.id), ProjectPlatform(project_id=project_id, platform_id=android.id, state="declared"), ProjectPlatform(project_id=project_id, platform_id=container.id, state="rejected"), FamilyAdoption(project_id=project_id, idea_id=idea_id, status="variant", reason="updates arrive through the store, not in-app", canon_version=1, decided_via="agent"), ]) first = FamilyDecision(idea_id=idea_id, action="promote", reason="proven in CI, stated in platform terms", after={"status": "canon"}) s.add(first) await s.flush() second = FamilyDecision(idea_id=idea_id, project_id=project_id, action="assess", reason="store-distributed, as in the promotion's terms", precedent_ids=[first.id], decided_via="agent") s.add(second) await s.commit() snippet_id = snippet.id async with async_session() as s: notes = (await s.execute( select(Note).where(Note.id.in_([idea_id, snippet_id])).order_by(Note.id) )).scalars().all() payload = { "version": backup.BACKUP_VERSION, "users": backup._user_rows([await s.get(User, owner_id)]), "projects": backup._project_rows([await s.get(Project, project_id)]), "notes": backup._note_rows(notes), **backup._family_sections( (await s.execute(select(Platform))).scalars().all(), (await s.execute(select(ProjectPlatform).where( ProjectPlatform.project_id == project_id))).scalars().all(), [await s.get(FamilyIdea, idea_id)], (await s.execute(select(FamilyIdeaPlatform).where( FamilyIdeaPlatform.note_id == idea_id))).scalars().all(), (await s.execute(select(FamilyIdeaReference).where( FamilyIdeaReference.idea_id == idea_id))).scalars().all(), (await s.execute(select(FamilyAdoption).where( FamilyAdoption.idea_id == idea_id))).scalars().all(), (await s.execute(select(FamilyDecision).where( FamilyDecision.idea_id == idea_id).order_by(FamilyDecision.id) )).scalars().all(), ), } payload["users"][0]["username"] = RESTORED_USERNAME return {"payload": payload} @pytest_asyncio.fixture async def restored(source): async with async_session() as s: before = len((await s.execute(select(Platform))).scalars().all()) stats = await backup.restore_full_backup(source["payload"]) async with async_session() as s: user = (await s.execute( select(User).where(User.username == RESTORED_USERNAME) )).scalars().one() project = (await s.execute( select(Project).where(Project.user_id == user.id) )).scalars().one() notes = (await s.execute( select(Note).where(Note.user_id == user.id) )).scalars().all() note_ids = [n.id for n in notes] return { "stats": stats, "platforms_before": before, "platforms_after": len((await s.execute(select(Platform))).scalars().all()), "project": project, "notes": {n.note_type: n for n in notes}, "idea": (await s.execute( select(FamilyIdea).where(FamilyIdea.note_id.in_(note_ids)) )).scalars().one(), "memberships": (await s.execute( select(ProjectPlatform).where(ProjectPlatform.project_id == project.id) )).scalars().all(), "adoption": (await s.execute( select(FamilyAdoption).where(FamilyAdoption.project_id == project.id) )).scalars().one(), "decisions": (await s.execute( select(FamilyDecision).where(FamilyDecision.idea_id.in_(note_ids)) .order_by(FamilyDecision.id) )).scalars().all(), "references": (await s.execute( select(FamilyIdeaReference).where(FamilyIdeaReference.idea_id.in_(note_ids)) )).scalars().all(), } async def test_platforms_match_by_slug_and_create_nothing(restored): assert restored["stats"]["platforms"] == 0 assert restored["platforms_after"] == restored["platforms_before"] async def test_a_rejected_platform_survives_so_detection_cannot_re_add_it(restored): states = {m.state for m in restored["memberships"]} assert states == {"declared", "rejected"} async def test_the_idea_and_its_answer_come_back_whole(restored): idea = restored["idea"] assert idea.status == "canon" assert idea.canon_version == 2 adoption = restored["adoption"] assert adoption.status == "variant" assert adoption.reason == "updates arrive through the store, not in-app" # Assessed against version 1 of a version-2 idea: the derived recheck # signal must survive the trip, which needs both numbers intact. assert adoption.canon_version < idea.canon_version async def test_the_reference_points_at_the_RESTORED_snippet(restored): [ref] = restored["references"] assert ref.snippet_id == restored["notes"]["snippet"].id async def test_a_precedent_is_remapped_to_the_RESTORED_decision(restored): """The trap: precedent_ids is a list of ids inside JSON. Copied raw, it would name the SOURCE decision's id — which in a shared database is a real row, the wrong one.""" first, second = restored["decisions"] assert first.action == "promote" and second.action == "assess" assert second.precedent_ids == [first.id] assert second.project_id == restored["project"].id