Spaces:
Runtime error
Runtime error
| """Reports built around themes: the posts' stored ideas are grouped into weighted themes, a blueprint plans the whole report, each | |
| section is written from its own evidence. The model is always faked here (by signature name, as the other report tests do).""" | |
| import logging | |
| import uuid | |
| from types import SimpleNamespace | |
| import pytest | |
| from sqlalchemy import select | |
| from app.config import settings | |
| from app.corpus import Corpus | |
| from app.llm import run | |
| from app.models import Document, Workspace | |
| from app.pipeline import report_blueprint, report_evidence, report_themes | |
| from app.pipeline.generate import ideas_fresh | |
| from app.pipeline.report_blueprint import Blueprint, Section | |
| from app.pipeline.report_themes import Idea, Theme, ThemesUnavailable | |
| from tests.conftest import last_code | |
| from tests.test_idea_pool import _notebook, make_posts | |
| def owner(client, db_session): | |
| client.post("/api/auth/email/register/start", json={"email": "ada@example.com", "password": "stardust-42", "name": "Ada"}) | |
| r = client.post("/api/auth/email/register/verify", json={"email": "ada@example.com", "code": last_code()}) | |
| return {"Authorization": f"Bearer {r.json()['access_token']}"}, db_session.scalar(select(Workspace)) | |
| def fake_doc() -> SimpleNamespace: | |
| return SimpleNamespace(id=uuid.uuid4(), path="sources/x/a.md", title="A") | |
| def theme(tid: str, name: str, posts: int, *, asked: bool = False, tier: str = "minor", weight: float = 0.1) -> Theme: | |
| t = Theme(id=tid, name=name, summary=f"About {name}.", ideas=[Idea(f"{tid}.{n}", fake_doc(), f"{name} {n}", "n") for n in range(posts)], | |
| asked=asked) | |
| t.tier, t.weight, t.pinned = tier, weight, asked | |
| return t | |
| # Weights are counted in code | |
| def test_weights_come_from_counting_posts_and_ideas_and_a_priority_theme_moves_up(): | |
| a, b, c = theme("t1", "A", 8), theme("t2", "B", 4), theme("t3", "C", 1, asked=True) | |
| themes = [a, b, c] | |
| report_themes.weigh(themes, total_posts=13, total_ideas=13) | |
| assert round(sum(t.weight for t in themes), 3) == 1.0 | |
| assert (a.weight, b.weight, c.weight) == (0.6154, 0.3077, 0.0769) | |
| assert (a.tier, b.tier) == ("major", "major") | |
| assert c.tier == "medium" and c.pinned # minor by share, asked for: one tier up and pinned | |
| def test_the_heaviest_theme_is_major_even_when_the_weights_are_flat(): | |
| themes = [theme(f"t{n}", f"T{n}", 3) for n in range(1, 11)] # ten equal themes: 0.1 each, medium | |
| report_themes.weigh(themes, total_posts=30, total_ideas=30) | |
| assert sum(t.tier == "major" for t in themes) == 1 | |
| # Grouping: the model names themes, code files what it left out | |
| def _ideas(db_session, ws, n=8): | |
| docs = make_posts(db_session, ws, n, ideas_each=2) | |
| corpus = Corpus(ws.id) | |
| corpus.sync() | |
| return docs, report_themes.collect_ideas(docs, corpus) | |
| def test_ideas_the_model_left_out_are_filed_by_code_and_the_rest_are_dropped_or_caught(owner, db_session, monkeypatch): | |
| _, ws = owner | |
| docs, ideas = _ideas(db_session, ws) # post i has ideas p{i}.0 and p{i}.1; posts cycle Pricing, Growth, Craft | |
| def predict(sig, *, db, workspace_id, job=None, lm=None, **inputs): | |
| assert sig.__name__ == "GroupThemes" and "p0.0 | Pricing idea 0.0:" in inputs["ideas"][0] | |
| return {"themes": [ | |
| {"name": "Pricing", "summary": "s", "matches_request": False, | |
| "idea_ids": ["p0.0", "p0.1", "p3.0", "p3.1", "p6.0", "p6.1", "nope"]}, | |
| {"name": "Audience Growth", "summary": "s", "matches_request": True, | |
| "idea_ids": ["p1.0", "p1.1", "p4.0", "p4.1", "p0.0"]}, | |
| {"name": "", "summary": "no name", "idea_ids": ["p2.0"], "matches_request": False}], | |
| "relations": [{"a": "pricing", "b": "Audience Growth", "relation": "Price decides who stays."}, | |
| {"a": "Pricing", "b": "Ghost", "relation": "unknown theme"}]} | |
| monkeypatch.setattr(run, "predict", predict) | |
| themes, relations = report_themes.group_themes(db_session, None, ws.id, "Money", "a report", ideas) | |
| by_name = {t.name: t for t in themes} | |
| assert set(by_name) == {"Pricing", "Audience Growth", report_themes.CATCH_ALL} | |
| assert len(by_name["Pricing"].ideas) == 6 # the invented id was dropped | |
| assert "p0.0" not in [i.id for i in by_name["Audience Growth"].ideas] # an idea belongs to one theme only | |
| growth = {i.id for i in by_name["Audience Growth"].ideas} | |
| assert {"p7.0", "p7.1"} <= growth # filed by the words it shares with the theme | |
| assert {i.doc.path for i in by_name[report_themes.CATCH_ALL].ideas} == {docs[2].path, docs[5].path} # nothing fits them | |
| assert sum(len(t.ideas) for t in themes) == len(ideas) # every idea counts, so the weights cover all posts | |
| assert by_name["Audience Growth"].pinned and [t.id for t in themes] == ["t1", "t2", "t3"] | |
| assert relations == [{"a": "t1", "b": "t2", "relation": "Price decides who stays."}] | |
| def test_too_few_themes_means_the_posts_are_read_as_before(owner, db_session, monkeypatch): | |
| _, ws = owner | |
| _, ideas = _ideas(db_session, ws) | |
| monkeypatch.setattr(run, "predict", lambda *a, **k: {"themes": [{"name": "Only", "summary": "", "idea_ids": ["p0.0"]}], | |
| "relations": []}) | |
| with pytest.raises(ThemesUnavailable): | |
| report_themes.group_themes(db_session, None, ws.id, "Money", "r", ideas) | |
| def test_the_model_sees_at_most_300_ideas_one_round_of_every_post_at_a_time(owner, db_session): | |
| _, ws = owner | |
| docs = make_posts(db_session, ws, 120, ideas_each=8) | |
| corpus = Corpus(ws.id) | |
| corpus.sync() | |
| ideas = report_themes.collect_ideas(docs, corpus) | |
| listed = report_themes.listed_ideas(ideas) | |
| assert len(ideas) == 960 and len(listed) == report_themes.MAX_LISTED_IDEAS | |
| assert len({i.doc.id for i in listed}) == 120 # every post is in the sample | |
| def test_a_post_without_ideas_is_represented_by_its_opening(owner, db_session): | |
| _, ws = owner | |
| docs = make_posts(db_session, ws, 2, with_ideas=0) | |
| corpus = Corpus(ws.id) | |
| corpus.sync() | |
| ideas = report_themes.collect_ideas(docs, corpus) | |
| assert [i.bare for i in ideas] == [True, True] and ideas[0].label == docs[0].title | |
| assert ideas[0].sources[0]["path"] == docs[0].path and ideas[0].sources[0]["line_start"] >= 1 | |
| def test_ideas_are_extracted_together_for_at_most_the_inline_limit(owner, db_session, monkeypatch): | |
| _, ws = owner | |
| docs = make_posts(db_session, ws, 6, with_ideas=0) | |
| seen = {} | |
| def predict_many(sig, inputs_list, *, db, workspace_id, lm=None, workers=5): | |
| seen["n"] = len(inputs_list) | |
| return [{"ideas": [{"label": "Idea", "note": "n", "details": [], | |
| "sources": [{"path": docs[0].path, "line_start": 7, "line_end": 9}]}]} for _ in inputs_list] | |
| monkeypatch.setattr(run, "predict_many", predict_many) | |
| monkeypatch.setattr(run, "predict", lambda *a, **k: pytest.fail("together, not one by one")) | |
| monkeypatch.setattr(settings, "report_inline_ideas", 4) | |
| report_themes.ensure_ideas(db_session, None, ws.id, docs) | |
| assert seen["n"] == 4 and sum(ideas_fresh(d) for d in docs) == 4 | |
| assert report_themes.fresh_share(docs) == pytest.approx(4 / 6) | |
| # The blueprint is checked against the weights | |
| def _bp(*sections: Section, connections=None) -> Blueprint: | |
| return Blueprint(title="T", storyline="S", sections=list(sections), connections=connections or []) | |
| def _sec(heading, role="body", ids=(), depth="standard", must=()): | |
| return Section(heading=heading, role=role, brief=f"{heading} brief", theme_ids=list(ids), depth=depth, must_cover=list(must)) | |
| def test_the_plan_is_made_to_cover_every_major_and_asked_for_theme_in_its_own_section(): | |
| themes = [theme("t1", "Pricing", 5, tier="major", weight=0.5), theme("t2", "Hiring", 3, tier="major", weight=0.3), | |
| theme("t3", "Tools", 1, weight=0.1), theme("t4", "Remote", 1, asked=True, tier="medium", weight=0.1)] | |
| plan = _bp(_sec("Summary", "summary"), _sec("Pricing power", ids=["t1", "t1", "ghost"], depth="brief", must=["a"]), | |
| _sec("Also pricing", ids=["t1"]), _sec("Opening", "intro"), | |
| connections=[{"theme_ids": ["t1", "t2"], "relation": "linked", "section": "pricing POWER"}, | |
| {"theme_ids": ["t1", "zzz"], "relation": "unknown theme", "section": "x"}]) | |
| fixed = report_blueprint.repair(plan, themes) | |
| heads = [s.heading for s in fixed.sections] | |
| assert heads == ["Opening", "Pricing power", "Hiring", "Remote", "More to know", "Summary"] # intro first, summary last | |
| by = {s.heading: s for s in fixed.sections} | |
| assert by["Pricing power"].theme_ids == ["t1"] and by["Pricing power"].depth == "deep" # depth follows the weight | |
| assert by["Remote"].depth == "deep" and by["More to know"].theme_ids == ["t3"] and by["More to know"].depth == "brief" | |
| assert by["Summary"].theme_ids == ["t1", "t2", "t4"] and by["Summary"].depth == "brief" | |
| assert "Hiring: About Hiring." in by["Pricing power"].covered_elsewhere | |
| assert not any(c.startswith("Pricing power") for c in by["Pricing power"].covered_elsewhere) | |
| assert by["Opening"].covered_elsewhere == [] and by["Summary"].covered_elsewhere == [] | |
| assert fixed.connections == [{"theme_ids": ["t1", "t2"], "relation": "linked", "section": "Pricing power"}] | |
| def test_too_many_sections_merge_the_lightest_but_never_a_major_theme(): | |
| themes = [theme("t1", "Big", 5, tier="major", weight=0.4)] + \ | |
| [theme(f"t{n}", f"Small {n}", 1, weight=0.05 + n / 1000) for n in range(2, 12)] | |
| plan = _bp(_sec("Open", "intro"), _sec("Big one", ids=["t1"]), *[_sec(f"S{n}", ids=[f"t{n}"]) for n in range(2, 12)], | |
| _sec("End", "summary")) | |
| fixed = report_blueprint.repair(plan, themes) | |
| assert len(fixed.sections) == report_blueprint.MAX_SECTIONS | |
| assert [s for s in fixed.sections if s.heading == "Big one"][0].theme_ids == ["t1"] | |
| assert sorted(i for s in fixed.sections if s.role == "body" for i in s.theme_ids) == sorted(t.id for t in themes) # none lost | |
| def test_a_plan_with_one_section_is_not_usable(): | |
| with pytest.raises(report_blueprint.BlueprintUnavailable): | |
| report_blueprint.repair(_bp(_sec("Only", ids=["t1"])), [theme("t1", "A", 1, tier="major", weight=1.0), ]) | |
| # The evidence of a section | |
| def _themes_from(docs, ideas, groups): | |
| """Themes from an index range of posts each, weighed.""" | |
| themes = [Theme(id=f"t{n}", name=name, summary=f"About {name}.", ideas=[i for i in ideas if lo <= docs.index(i.doc) < hi]) | |
| for n, (name, lo, hi) in enumerate(groups, start=1)] | |
| report_themes.weigh(themes, len(docs), len(ideas)) | |
| return themes | |
| def test_a_deep_section_reads_several_posts_as_written_and_summarises_the_rest(owner, db_session): | |
| _, ws = owner | |
| docs, ideas = _ideas(db_session, ws, 10) | |
| corpus = Corpus(ws.id) | |
| t = _themes_from(docs, ideas, [("Everything", 0, 10), ("Other", 0, 0)][:1]) | |
| sec = _sec("Deep", ids=["t1"], depth="deep") | |
| ev = report_evidence.build(corpus, sec, {x.id: x for x in t}) | |
| raw = [b for b in ev.blocks if b.startswith("POST ")] | |
| assert 2 <= len(raw) == ev.raw_posts == 5 # a deep section reads five posts as written, never one | |
| assert any(b.startswith("SUMMARIES of other posts") for b in ev.blocks) | |
| assert len(ev.posts) == 10 # the other five are represented by their summaries | |
| assert "Sentence 1 about" in raw[0] and "[themes: Everything]" in raw[0] # the post's own words | |
| assert "7| " not in raw[0] and "sources/x" not in raw[0] # nothing is cited, so no line numbers and no paths | |
| assert sum(len(b) for b in ev.blocks) <= report_evidence.BUDGETS["deep"] | |
| def test_a_brief_section_gets_summaries_only_and_a_standard_one_three_posts(owner, db_session): | |
| _, ws = owner | |
| docs, ideas = _ideas(db_session, ws, 10) | |
| corpus = Corpus(ws.id) | |
| t = _themes_from(docs, ideas, [("Everything", 0, 10)]) | |
| by_id = {x.id: x for x in t} | |
| brief = report_evidence.build(corpus, _sec("Brief", ids=["t1"], depth="brief"), by_id) | |
| assert brief.raw_posts == 0 and not any(b.startswith("POST ") for b in brief.blocks) | |
| assert sum(len(b) for b in brief.blocks) <= report_evidence.BUDGETS["brief"] | |
| assert all(' (from "Post ' in line for line in brief.blocks[0].splitlines()[1:]) # each summary says which post it is from | |
| standard = report_evidence.build(corpus, _sec("Std", ids=["t1"], depth="standard"), by_id) | |
| assert standard.raw_posts == 3 | |
| def test_a_section_about_two_themes_starts_with_the_posts_that_speak_of_both(owner, db_session): | |
| _, ws = owner | |
| docs, ideas = _ideas(db_session, ws, 8) | |
| corpus = Corpus(ws.id) | |
| a, b = Theme("t1", "A", "", []), Theme("t2", "B", "", []) | |
| for d in docs: | |
| mine = [i for i in ideas if i.doc is d] | |
| a.ideas.append(mine[0]) | |
| if d in docs[5:7]: # only these two posts have an idea in each theme | |
| b.ideas.append(mine[1]) | |
| report_themes.weigh([a, b], 8, len(ideas)) | |
| ev = report_evidence.build(corpus, _sec("Bridge", "synthesis", ids=["t1", "t2"], depth="standard"), {"t1": a, "t2": b}) | |
| first = [b_ for b_ in ev.blocks if b_.startswith("POST ")][:2] | |
| assert {docs[5].title, docs[6].title} == {blk.split(" [themes")[0][5:] for blk in first} | |
| def test_an_unreadable_post_falls_back_to_its_summary(owner, db_session): | |
| _, ws = owner | |
| docs, ideas = _ideas(db_session, ws, 6) | |
| ideas[0].doc.path = "sources/x/missing.md" | |
| corpus = Corpus(ws.id) | |
| t = _themes_from(docs, ideas, [("All", 0, 6)]) | |
| ev = report_evidence.build(corpus, _sec("S", ids=["t1"], depth="deep"), {"t1": t[0]}) | |
| assert "sources/x/missing.md" in ev.posts and ev.raw_posts == 5 # the five others | |
| # The whole report | |
| def free_to_make(monkeypatch): | |
| monkeypatch.setattr("app.routers.artifacts.check_limit", lambda *a, **k: None) | |
| class Fake: | |
| """The model, by signature name. `calls` is every call in order; `sections` the inputs of every WriteThemedSection.""" | |
| def __init__(self, monkeypatch, docs, groups, ask=None, fail=(), quiz_themes=()): | |
| self.docs, self.calls, self.sections, self.fail = docs, [], [], set(fail) | |
| self.quiz_themes = list(quiz_themes) | |
| self.groups, self.ask, self.plan_inputs, self.report_inputs = groups, ask, None, None | |
| monkeypatch.setattr(run, "predict", self.predict) | |
| monkeypatch.setattr(run, "predict_many", self.predict_many) | |
| def predict(self, sig, *, db, workspace_id, job=None, lm=None, **inputs): | |
| name = sig.__name__ | |
| self.calls.append(name) | |
| if name == "GroupThemes": | |
| ids = [line.split(" | ")[0] for line in inputs["ideas"]] | |
| out = [] | |
| for title, lo, hi in self.groups: | |
| mine = [i for i in ids if lo <= int(i[1:].split(".")[0]) < hi] | |
| out.append({"name": title, "summary": f"About {title}.", "idea_ids": mine, "matches_request": title == self.ask}) | |
| rel = [{"a": self.groups[0][0], "b": self.groups[1][0], "relation": "One shapes the other."}] if len(self.groups) > 1 else [] | |
| return {"themes": out, "relations": rel} | |
| if name == "PlanReportBlueprint": | |
| self.plan_inputs = inputs | |
| return {"report_title": "Pricing and growth", "storyline": "Prices set expectations and growth follows.", "sections": [ | |
| {"heading": "Why this matters", "role": "intro", "brief": "Opens.", "theme_ids": [], "depth": "brief", | |
| "must_cover": ["scope"]}, | |
| {"heading": "Pricing power", "role": "body", "brief": "Pricing.", "theme_ids": ["t1"], "depth": "deep", | |
| "must_cover": ["raise slowly"], | |
| "embed": {"kind": "flashcards", "title": "Terms", "brief": "Price terms", "why": "Learn them"}}, | |
| {"heading": "Audience growth", "role": "body", "brief": "Growth.", "theme_ids": ["t2"], "depth": "standard", | |
| "must_cover": ["retention"]}, | |
| {"heading": "Quick quiz", "role": "body", "brief": "Test it.", "theme_ids": self.quiz_themes, "depth": "brief", | |
| "must_cover": []}, | |
| {"heading": "Wrap up", "role": "summary", "brief": "Closes.", "theme_ids": [], "depth": "brief", | |
| "must_cover": ["main points"]}], | |
| "connections": [{"theme_ids": ["t1", "t2"], "relation": "Price shapes who stays.", | |
| "section": "Audience growth"}]} | |
| if name == "WriteThemedReport": | |
| self.report_inputs = inputs | |
| text = "## Why this matters\nIntro [1].\n\n## Pricing power\nBody [2].\n\n## Wrap up\nEnd." | |
| return {"report_title": "One Piece", "markdown": text} | |
| if name == "WriteReport": | |
| return {"report_title": "Legacy", "markdown": "## One\nText."} | |
| raise AssertionError(f"unexpected {name}") | |
| def predict_many(self, sig, inputs_list, *, db, workspace_id, lm=None, workers=5): | |
| assert sig.__name__ == "WriteThemedSection" | |
| out = [] | |
| for inp in inputs_list: | |
| self.sections.append(inp) | |
| self.calls.append("WriteThemedSection") | |
| if inp["heading"] in self.fail: | |
| out.append(None) | |
| else: | |
| out.append({"markdown": f"{inp['heading']} text [1]."}) # a marker the model wrote anyway | |
| return out | |
| def _report(client, owner, docs, fmt, **extra): | |
| headers, _ = owner | |
| nb = _notebook(client, owner, docs) | |
| return client.post("/api/artifacts/generate", json={ | |
| "type": "report", "notebook_id": nb["id"], "document_ids": [str(d.id) for d in docs], "report_format": fmt, | |
| "instructions": "Write a report.", **extra}, headers=headers).json() | |
| def _content(client, owner, art): | |
| return client.get(f"/api/artifacts/{art['id']}", headers=owner[0]).json()["content"] | |
| GROUPS = [("Pricing", 0, 6), ("Audience Growth", 6, 8), ("Writing Craft", 8, 10)] # 60%, 20%, 20% of the posts | |
| def test_an_interactive_report_is_planned_then_written_section_by_section_from_its_own_evidence(client, owner, db_session, run_jobs, | |
| monkeypatch, free_to_make): | |
| docs = make_posts(db_session, owner[1], 10, ideas_each=2) | |
| fake = Fake(monkeypatch, docs, GROUPS, quiz_themes=["t3"]) # the plan wrongly makes a section of a quiz | |
| art = _report(client, owner, docs, "interactive") | |
| run_jobs() | |
| c = _content(client, owner, art) | |
| assert c["format"] == "interactive" and c["plan"]["planner"] == "themes" | |
| assert [b["text"].split("\n")[0] for b in c["blocks"]] == [ | |
| "## Why this matters", "## Pricing power", "## Audience growth", "## Wrap up"] | |
| assert "PlanInteractiveReport" not in fake.calls and "WriteReportSection" not in fake.calls and "WriteReport" not in fake.calls | |
| assert fake.calls[:2] == ["GroupThemes", "PlanReportBlueprint"] | |
| # The weights are counted: Pricing is in six of ten posts | |
| themes = {t["name"]: t for t in c["plan"]["themes"]} | |
| assert themes["Pricing"]["tier"] == "major" and themes["Pricing"]["weight"] == 0.6 and themes["Audience Growth"]["tier"] == "medium" | |
| assert c["plan"]["posts_total"] == 10 and c["source"]["total"] == 10 and len(c["source"]["document_ids"]) == 10 | |
| # Nothing is cited. The Sources list is every post the report was made from, shown in a drop-down with the prompt | |
| assert c["citations"] == [] and not any("[" in b["text"] for b in c["blocks"]) | |
| assert c["source"]["read"] == c["plan"]["posts_used"] == len(c["sources"]) == 10 | |
| assert {x["document_id"] for x in c["sources"]} == {str(d.id) for d in docs} and all("markers" not in x for x in c["sources"]) | |
| assert c["instructions"] == "Write a report." | |
| # The planner saw the allowed visuals and the themes, never the posts | |
| assert fake.plan_inputs["allowed_kinds"] and all("POST" not in line for line in fake.plan_inputs["themes"]) | |
| # Body sections first, then the introduction and the summary, written from what the sections say | |
| written = [s["heading"] for s in fake.sections] | |
| assert written[:2] == ["Pricing power", "Audience growth"] and written[2:] == ["Why this matters", "Wrap up"] | |
| pricing = fake.sections[0] | |
| assert pricing["storyline"].startswith("Prices set") and len(pricing["outline"]) == 4 | |
| assert pricing["covered_elsewhere"][0].startswith("Audience growth:") | |
| assert all(not x.startswith("Pricing power") for x in pricing["covered_elsewhere"]) | |
| assert sum(e.startswith("POST ") for e in pricing["evidence"]) >= 2 # several posts as written | |
| growth = fake.sections[1] | |
| assert growth["bridges"] == ["Pricing + Audience Growth: Price shapes who stays. (bridged in: Audience growth)"] | |
| assert fake.sections[2]["evidence"][0].startswith("Pricing power: Pricing power text") and fake.sections[2]["role"] == "intro" | |
| assert "Restate the main points about: Pricing" in fake.sections[3]["must_cover"][-1] | |
| assert growth["evidence"] and "Writing Craft" in growth["evidence"][-1] # the quiz section's theme stayed in the report | |
| # Visuals are only suggested; the quiz section became a suggestion | |
| # The plan suggested no infographic, so one is offered after the last section before the summary | |
| assert [(s["after_block_id"], s["kind"]) for s in c["suggestions"]] == [ | |
| ("b2", "flashcards"), ("b3", "quiz"), ("b3", "infographic")] | |
| assert "failed_sections" not in c | |
| def test_a_document_report_is_written_in_one_piece_from_the_plan_and_offers_no_visuals(client, owner, db_session, run_jobs, | |
| monkeypatch, free_to_make): | |
| docs = make_posts(db_session, owner[1], 10, ideas_each=2) | |
| fake = Fake(monkeypatch, docs, GROUPS) | |
| art = _report(client, owner, docs, "document") | |
| run_jobs() | |
| c = _content(client, owner, art) | |
| assert c["format"] == "document" and c["plan"]["planner"] == "themes" | |
| assert c["suggestions"] == [] # never in a document | |
| assert fake.calls == ["GroupThemes", "PlanReportBlueprint", "WriteThemedReport"] | |
| assert fake.plan_inputs["allowed_kinds"] == [] | |
| sections = fake.report_inputs["sections"] | |
| assert sections[0].startswith("SECTION 1: Why this matters") and all(s.startswith("SECTION") for s in sections) | |
| assert "Evidence:\nPOST " in sections[1] and fake.report_inputs["storyline"].startswith("Prices set") | |
| assert [b["text"].split("\n")[0] for b in c["blocks"]] == ["## Why this matters", "## Pricing power", "## Wrap up"] | |
| assert c["citations"] == [] and not any("[" in b["text"] for b in c["blocks"]) and len(c["sources"]) == 10 | |
| assert sum(len(s) for s in sections) < report_evidence.BUDGETS["deep"] * 4 # the evidence fits its budget | |
| def test_more_than_twenty_posts_are_all_used(client, owner, db_session, run_jobs, monkeypatch, free_to_make): | |
| docs = make_posts(db_session, owner[1], 30, ideas_each=2) | |
| fake = Fake(monkeypatch, docs, [("Pricing", 0, 20), ("Audience Growth", 20, 25), ("Writing Craft", 25, 30)]) | |
| art = _report(client, owner, docs, "interactive") | |
| run_jobs() | |
| c = _content(client, owner, art) | |
| assert c["plan"]["posts_total"] == 30 and len(c["source"]["document_ids"]) == 30 | |
| assert {t["name"]: t["posts"] for t in c["plan"]["themes"]} == {"Pricing": 20, "Audience Growth": 5, "Writing Craft": 5} | |
| assert fake.calls[0] == "GroupThemes" | |
| def test_a_theme_the_reader_asked_for_gets_its_own_deep_section(client, owner, db_session, run_jobs, monkeypatch, free_to_make): | |
| docs = make_posts(db_session, owner[1], 10, ideas_each=2) | |
| Fake(monkeypatch, docs, GROUPS, ask="Writing Craft") | |
| art = _report(client, owner, docs, "interactive", instructions="Focus on writing craft.") | |
| run_jobs() | |
| c = _content(client, owner, art) | |
| sections = {s["heading"]: s for s in c["plan"]["sections"]} | |
| assert sections["Writing Craft"]["depth"] == "deep" and sections["Writing Craft"]["theme_ids"] == ["t3"] | |
| assert {t["name"]: t["pinned"] for t in c["plan"]["themes"]}["Writing Craft"] is True | |
| def test_a_section_that_fails_is_tried_again_then_reported_not_hidden(client, owner, db_session, run_jobs, monkeypatch, free_to_make): | |
| docs = make_posts(db_session, owner[1], 10, ideas_each=2) | |
| seen: list[str] = [] | |
| class Flaky(Fake): | |
| def predict_many(self, sig, inputs_list, **kw): | |
| seen.extend(i["heading"] for i in inputs_list) | |
| return super().predict_many(sig, inputs_list, **kw) | |
| Flaky(monkeypatch, docs, GROUPS, fail=["Audience growth"]) | |
| art = _report(client, owner, docs, "interactive") | |
| run_jobs() | |
| c = _content(client, owner, art) | |
| assert seen.count("Audience growth") == 2 # once, then once more | |
| assert c["failed_sections"] == ["Audience growth"] | |
| assert [b["text"].split("\n")[0] for b in c["blocks"]] == ["## Why this matters", "## Pricing power", "## More to know", "## Wrap up"] | |
| def test_when_the_ideas_cannot_be_themed_the_posts_are_read_as_before(client, owner, db_session, run_jobs, monkeypatch, free_to_make): | |
| docs = make_posts(db_session, owner[1], 10, ideas_each=2) | |
| fake = Fake(monkeypatch, docs, [("Only", 0, 10)]) # a single theme is no plan | |
| art = _report(client, owner, docs, "document") | |
| run_jobs() | |
| c = _content(client, owner, art) | |
| assert fake.calls == ["GroupThemes", "WriteReport"] and "plan" not in c and c["title"] == "Legacy" | |
| def test_small_selections_chats_and_the_legacy_setting_keep_the_old_way(client, owner, db_session, run_jobs, monkeypatch, free_to_make): | |
| docs = make_posts(db_session, owner[1], 10, ideas_each=2) | |
| fake = Fake(monkeypatch, docs, GROUPS) | |
| small = _report(client, owner, docs[:5], "document") # five posts or fewer: read whole | |
| run_jobs() | |
| assert fake.calls == ["WriteReport"] and "plan" not in _content(client, owner, small) | |
| fake.calls.clear() | |
| monkeypatch.setattr(settings, "report_planner", "legacy") | |
| _report(client, owner, docs, "document") | |
| run_jobs() | |
| assert fake.calls == ["WriteReport"] | |
| def test_posts_without_ideas_use_the_old_way_until_half_have_them(client, owner, db_session, run_jobs, monkeypatch, free_to_make): | |
| docs = make_posts(db_session, owner[1], 10, with_ideas=3, ideas_each=2) | |
| monkeypatch.setattr(settings, "report_inline_ideas", 0) # nothing is extracted while the report is made | |
| fake = Fake(monkeypatch, docs, GROUPS) | |
| _report(client, owner, docs, "document") | |
| run_jobs() | |
| assert fake.calls == ["WriteReport"] | |
| assert [d for d in db_session.scalars(select(Document)) if ideas_fresh(d)].__len__() == 3 | |
| def test_every_step_of_a_themed_report_is_logged(client, owner, db_session, run_jobs, monkeypatch, free_to_make, caplog): | |
| caplog.set_level(logging.INFO, logger="app") | |
| docs = make_posts(db_session, owner[1], 10, ideas_each=2) | |
| Fake(monkeypatch, docs, GROUPS) | |
| _report(client, owner, docs, "interactive") | |
| run_jobs() | |
| text = "\n".join(r.getMessage() for r in caplog.records) | |
| for expected in ("report posts: ", "themed report starting: ", "report ideas collected: ", "report themes: asking the model", | |
| "report theme: id=t1 name='Pricing' weight=0.600 tier=major", "report blueprint checked: ", | |
| "report section 1: heading='Why this matters'", "report evidence: section='Pricing power' depth=deep", | |
| "report wave 1: ", "report wave 2: ", "report interactive written: ", "themed report done: ", | |
| "report: job="): | |
| assert expected in text, expected | |
| assert "way=themes" in text | |
| def test_a_fallback_says_why(client, owner, db_session, run_jobs, monkeypatch, free_to_make, caplog): | |
| caplog.set_level(logging.INFO, logger="app") | |
| docs = make_posts(db_session, owner[1], 10, ideas_each=2) | |
| Fake(monkeypatch, docs, [("Only", 0, 10)]) | |
| _report(client, owner, docs, "document") | |
| run_jobs() | |
| text = "\n".join(r.getMessage() for r in caplog.records) | |
| assert "report themes unusable: usable=1 needed=2" in text and "reading the posts instead" in text and "way=legacy" in text | |
| caplog.clear() | |
| _report(client, owner, docs[:5], "document") | |
| run_jobs() | |
| assert "report route: legacy" in "\n".join(r.getMessage() for r in caplog.records) | |
| def test_a_report_always_offers_visuals_and_never_repeats_what_the_plan_suggested(): | |
| from app.pipeline import report | |
| blocks = [{"id": f"b{n}", "type": "prose"} for n in range(1, 6)] | |
| kinds = list(report.EMBED_KINDS) | |
| nothing = report.fill_suggestions([], blocks, kinds) | |
| assert [(s["kind"], s["after_block_id"]) for s in nothing] == [("infographic", "b4"), ("flashcards", "b2"), ("quiz", "b3")] | |
| planned = [{"id": "s1", "after_block_id": "b1", "kind": "infographic", "brief": "x", "why": "y"}] | |
| kept = report.fill_suggestions(planned, blocks, kinds) | |
| assert kept[0] == planned[0] and [s["kind"] for s in kept] == ["infographic", "flashcards", "quiz"] # not offered twice | |
| assert [s["kind"] for s in report.fill_suggestions([], blocks, ["infographic", "mind_map"])] == ["infographic", "mind_map"] | |
| assert report.fill_suggestions([], blocks, []) == [] and report.fill_suggestions([], [], kinds) == [] | |
| chat = report.fill_suggestions([], blocks, [k for k in kinds if k != "mind_map"]) | |
| assert "mind_map" not in {s["kind"] for s in chat} | |
| def test_an_older_report_without_suggestions_gets_them_once_when_opened(client, owner, db_session, run_jobs, monkeypatch, free_to_make): | |
| from app.models import Artifact | |
| docs = make_posts(db_session, owner[1], 10, ideas_each=2) | |
| Fake(monkeypatch, docs, GROUPS) | |
| art = _report(client, owner, docs, "interactive") | |
| run_jobs() | |
| row = db_session.get(Artifact, uuid.UUID(art["id"])) | |
| row.content_json = {k: v for k, v in row.content_json.items() if k not in ("suggestions", "suggestions_offered")} # as an old one | |
| db_session.commit() | |
| first = _content(client, owner, art) | |
| assert {"infographic", "flashcards", "quiz"} <= {s["kind"] for s in first["suggestions"]} and first["suggestions_offered"] | |
| # Once: a reader who has added or taken everything away is not offered them again | |
| row = db_session.get(Artifact, uuid.UUID(art["id"])) | |
| db_session.refresh(row) | |
| row.content_json = {**row.content_json, "suggestions": []} | |
| db_session.commit() | |
| assert _content(client, owner, art)["suggestions"] == [] | |
| def test_a_document_report_is_never_given_suggestions_not_even_an_old_one_when_opened(client, owner, db_session, run_jobs, monkeypatch, | |
| free_to_make): | |
| from app.models import Artifact | |
| docs = make_posts(db_session, owner[1], 10, ideas_each=2) | |
| Fake(monkeypatch, docs, GROUPS) | |
| art = _report(client, owner, docs, "document") | |
| run_jobs() | |
| row = db_session.get(Artifact, uuid.UUID(art["id"])) | |
| row.content_json = {k: v for k, v in row.content_json.items() if k not in ("suggestions", "suggestions_offered")} | |
| db_session.commit() | |
| assert _content(client, owner, art).get("suggestions", []) == [] | |