diff --git a/docs/operations.md b/docs/operations.md index 91985f5..0dd539a 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -94,6 +94,7 @@ in the bot would read it. | `STURNUS_WHISPER_DEFAULT_LANGUAGE` | `de` | no | Language reported when faster-whisper's own detection comes up empty. This is the floor under the per-guild `transcription_language` (section 4.1), not the usual setting to reach for — it is consulted only for a guild that asked for detection (`transcription_language auto`) and got nothing back. It still matters more than a fallback usually does: the first transcription for a speaker in such a session pins that speaker's language, and every later job for them reuses it. | | `STURNUS_MODEL_CACHE_DIR` | unset | no | Where model weights are cached. When set, the worker exports it as `HF_HOME` before loading the model, so the download lands on a persistent volume; left unset, every cold start re-downloads several gigabytes of weights. | | `STURNUS_WORK_DIR` | `/tmp` | no | Scratch directory the encrypted recording is downloaded and decrypted into before transcription. It must be large enough for the biggest single recording — the chart sizes the corresponding volume with `worker.tmpSizeLimit`. | +| `STURNUS_CONSOLE_ORIGIN` | `https://sturnus.onelitefeather.dev` | no | Where a protocol stored in the object store is served from. A guild that publishes to a `markdown` or `html` export target gets an artefact in the audio bucket, and the URL recorded for it — the one the bot posts into the channel — is this origin plus `/api/sessions/{id}/documents/{target_id}`, answered by `sturnus-api` under the same participant rule as the session's own page. Deliberately **not** a presigned S3 URL, which would keep working for anybody it was forwarded to and could not be revoked. It must be the same value `sturnus-api` runs with, or every such link points at a host that does not answer — and because the link is posted into Discord, it outlives any later correction. | | `STURNUS_MAX_JOB_ATTEMPTS` | `3` | no | How many failed attempts a job gets before `JobQueue.fail` marks it `dead`. See section 5 for what a `dead` job means for the rest of its session. | | `STURNUS_JOB_LEASE_SECONDS` | `1800.0` | no | How long a claimed job may stay `running` before `JobQueue.claim` reclaims it for another worker. It is generous on purpose: it must exceed the longest plausible transcription, or a still-running job gets picked up a second time. | | `STURNUS_HEALTH_PORT` | `8080` | no | Port the `/healthz`, `/readyz`, `/metrics`, `/version` HTTP endpoints listen on. `/metrics` answers **`501 Not Implemented`**: metrics are *pushed* over OTLP, not scraped — see section 7. | diff --git a/src/sturnus/application/documents.py b/src/sturnus/application/documents.py index 77fa5b4..074e053 100644 --- a/src/sturnus/application/documents.py +++ b/src/sturnus/application/documents.py @@ -59,6 +59,27 @@ def _build_environment() -> SandboxedEnvironment: return env +def _build_html_environment() -> SandboxedEnvironment: + """The same engine with the opposite escaping, for an HTML destination. + + Two things here are the whole point of having a second environment + rather than a second template on the first one. + + `autoescape=True`: every value drawn from the transcript is escaped as + HTML on its way into the output, so a display name of + `` reaches the page as text. + + **No `md` filter.** It is not merely unused here -- it is absent, so a + template that reaches for it fails to render rather than quietly + emitting `escape_markdown`'s backslashes into a document where nothing + ever removes them again. The registry + (`sturnus.application.export_formats`) is what pairs each template with + the environment that is right for it; this is the half of that pairing + that cannot be got wrong silently. + """ + return SandboxedEnvironment(autoescape=True, trim_blocks=True, lstrip_blocks=True) + + class _BlockContext(TypedDict): time: str speaker: SpeakerIdentity @@ -112,21 +133,19 @@ def label(self) -> str: return self.name if self.name else str(self.channel_id) -def render_transcript( +def transcript_context( transcript: Transcript, - template_source: str, tz: tzinfo, channel: ChannelRef | None = None, -) -> str: - """Renders `transcript` through `template_source`, localised to `tz`. - - The timezone is a parameter rather than a constant: a protocol read by - people in one place should carry local times, and hardcoding UTC would - make every timestamp subtly wrong for its readers -- wrong in a way that - does not look wrong, since any hour reads as a plausible meeting time. - - `channel` is optional so a caller that cannot resolve it still gets a - protocol; the template simply omits the heading. +) -> dict[str, object]: + """Everything a document template may read, localised to `tz`. + + Extracted from `render_transcript` when a second output *shape* -- HTML + -- arrived. The two differ in exactly one thing, the escaping, and + nothing else about a protocol changes with its destination: the same + participants, the same blocks, the same local times. Building the + context once is what keeps that true, rather than leaving a second copy + to drift into rendering a different meeting. """ blocks: list[_BlockContext] = [ { @@ -138,19 +157,70 @@ def render_transcript( ] local_start = transcript.session_started_at.astimezone(tz) duration = transcript.session_ended_at - transcript.session_started_at - template = _build_environment().from_string(template_source) - return template.render( - participants=transcript.participants, - blocks=blocks, - channel=channel, + return { + "participants": transcript.participants, + "blocks": blocks, + "channel": channel, # Outline renders `mention://date/` as a date chip # (MentionType.Date). Date only: the version deployed here parses no # time component, so an ISO datetime would degrade to a plain link. - date_iso=local_start.strftime("%Y-%m-%d"), - date_label=local_start.strftime("%d.%m.%Y"), - started=local_start.strftime("%H:%M"), - duration_minutes=max(1, round(duration.total_seconds() / 60)), - ) + "date_iso": local_start.strftime("%Y-%m-%d"), + "date_label": local_start.strftime("%d.%m.%Y"), + "started": local_start.strftime("%H:%M"), + "duration_minutes": max(1, round(duration.total_seconds() / 60)), + # Only a standalone document needs its own title inside its body; + # the Outline template does not read this, because Outline stores + # the title as a field of its own. It is in the context rather than + # a parameter of the HTML renderer alone so there is one definition + # of what a protocol is called (`document_title`) and one context + # every template is rendered against. + "title": document_title(transcript, tz), + } + + +def render_transcript( + transcript: Transcript, + template_source: str, + tz: tzinfo, + channel: ChannelRef | None = None, +) -> str: + """Renders `transcript` through `template_source` as Markdown. + + The timezone is a parameter rather than a constant: a protocol read by + people in one place should carry local times, and hardcoding UTC would + make every timestamp subtly wrong for its readers -- wrong in a way that + does not look wrong, since any hour reads as a plausible meeting time. + + `channel` is optional so a caller that cannot resolve it still gets a + protocol; the template simply omits the heading. + + **Markdown, and only Markdown.** The environment escapes through the + `md` filter and autoescapes nothing, so handing this function an HTML + template would produce a page that renders a hostile display name as + markup. `render_html` is the other half of that pair; which one a + destination gets is `sturnus.application.export_formats`' decision. + """ + template = _build_environment().from_string(template_source) + return template.render(**transcript_context(transcript, tz, channel)) + + +def render_html( + transcript: Transcript, + template_source: str, + tz: tzinfo, + channel: ChannelRef | None = None, +) -> str: + """The same protocol, rendered as HTML with HTML escaping. + + Deliberately not `render_transcript` with a different template. + `escape_markdown` answers `") + body = render("html", block(hostile, 0, "hello")) + assert "" not in body + assert "<script>" in body + + +def test_html_escapes_hostile_transcript_text_as_html() -> None: + body = render("html", block(GUEST, 0, "")) + assert "" not in body + assert "<img" in body + + +def test_html_does_not_carry_markdown_backslash_escapes() -> None: + """`escape_markdown` escapes `.` and `-` among others, so a sentence + that went through it arrives full of backslashes -- visible ones, in + HTML, where nothing ever removes them again.""" + body = render("html", block(GUEST, 0, "one. two-three!")) + assert "one. two-three!" in body + assert "\\." not in body + + +def test_html_is_a_whole_document_a_browser_can_open() -> None: + """It is served from an object store on its own, not embedded in a + page somebody else wrote.""" + body = render("html", block(GUEST, 0, "hello")) + assert body.lstrip().lower().startswith("") + assert "" in body + assert 'charset="utf-8"' in body or "charset=utf-8" in body + + +def test_html_carries_no_mention_syntax() -> None: + body = render("html", block(LINKED, 0, "hello")) + assert "mention://" not in body + + +def test_html_goes_to_the_object_store_as_text_html() -> None: + entry = export_formats.format_named("html") + assert entry is not None + assert entry.sink == export_formats.OBJECT_STORE_SINK + assert entry.media_type == "text/html; charset=utf-8" + assert entry.file_extension == "html" + + +# --------------------------------------------------------------------------- +# What every renderer owes its reader, whichever one it is +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("name", ["outline", "markdown", "html"]) +def test_every_format_localises_times_to_the_configured_timezone(name: str) -> None: + """20:00 UTC is 22:00 in Berlin, and a protocol read in Berlin that + says 20:00 is wrong in a way that does not look wrong.""" + body = render(name, block(GUEST, 0, "hello"), tz=ZoneInfo("Europe/Berlin")) + assert "22:00" in body + + +@pytest.mark.parametrize("name", ["outline", "markdown", "html"]) +def test_every_format_renders_without_a_channel(name: str) -> None: + """A session recorded before the channel name was captured still gets + a protocol; the heading is simply absent.""" + assert render(name, block(GUEST, 0, "hello"), channel=None) + + +@pytest.mark.parametrize("name", ["outline", "markdown", "html"]) +def test_every_format_names_the_channel_when_there_is_one(name: str) -> None: + """A channel name with no metacharacter in it, so this asserts on the + heading being present rather than on how each format escapes -- which + is what the escaping tests above are for, one per format.""" + body = render(name, block(GUEST, 0, "hello"), channel=ChannelRef(1, 4711, "meetingraum")) + assert "meetingraum" in body + assert "https://discord.com/channels/1/4711" in body + + +# --------------------------------------------------------------------------- +# What a target string may say, per format +# --------------------------------------------------------------------------- + + +def test_an_object_store_target_may_not_climb_out_of_its_prefix() -> None: + """The target becomes part of an object key. `..` in one is not a + traversal in S3, but it is a key nobody meant to write and one no + listing groups where the administrator expects it.""" + entry = export_formats.format_named("markdown") + assert entry is not None + assert entry.accepts_target("protocols") + assert entry.accepts_target("team/protocols") + assert not entry.accepts_target("../secrets") + assert not entry.accepts_target("/absolute") + assert not entry.accepts_target("") + + +def test_an_outline_target_is_a_collection_id() -> None: + entry = export_formats.format_named("outline") + assert entry is not None + assert entry.accepts_target("c9a1b2e3-4f5a-4b3c-8d2e-1a2b3c4d5e6f") + assert not entry.accepts_target("") + assert not entry.accepts_target("a collection with spaces") diff --git a/tests/application/test_exporting.py b/tests/application/test_exporting.py new file mode 100644 index 0000000..5a7e3d1 --- /dev/null +++ b/tests/application/test_exporting.py @@ -0,0 +1,567 @@ +"""Publishing one protocol to several destinations, and surviving one of them. + +Two behaviours carry this file. + +**One failing destination must not lose the others.** A guild with Outline +and a second destination whose service is down still gets its Outline +document, and the failure is recorded as a failure rather than as the end of +the publish. + +**The retry sweep must retry only what actually failed.** `session_document` +is the record of what succeeded, and a destination already in it is skipped +on the next sweep. Without that, a flaky second destination reprints the +Outline document every five minutes for as long as it stays flaky. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest + +from sturnus.application import exporting +from sturnus.application.documents import CreatedDocument +from sturnus.application.export_formats import HTML, MARKDOWN, OUTLINE, format_named +from sturnus.domain.exports import ExportTarget, SessionDocument +from sturnus.domain.transcript import SpeakerIdentity, Transcript, TranscriptBlock + +T0 = datetime(2026, 8, 19, 20, 0, 0, tzinfo=UTC) +NOW = T0 + timedelta(hours=2) +SESSION = 77 + +TEMPLATE = ( + Path(__file__).parent.parent.parent + / "src/sturnus/infrastructure/documents/outline_template.md.j2" +).read_text(encoding="utf-8") + +SPEAKER = SpeakerIdentity(100, "speaker") + + +def transcript() -> Transcript: + return Transcript( + session_started_at=T0, + session_ended_at=T0 + timedelta(hours=1), + participants=(SPEAKER,), + blocks=(TranscriptBlock(speaker=SPEAKER, start=T0, text="hello"),), + ) + + +def request() -> exporting.RenderRequest: + return exporting.RenderRequest( + transcript=transcript(), tz=UTC, channel=None, outline_template=TEMPLATE + ) + + +def target( + target_id: int, + format: str = OUTLINE, + name: str = "wiki", + where: str = "col-1", + enabled: bool = True, +) -> ExportTarget: + return ExportTarget( + id=target_id, + guild_id=1, + format=format, + name=name, + target=where, + config={}, + has_secret=False, + enabled=enabled, + created_at=T0, + updated_at=T0, + ) + + +def destination( + target_id: int | None, format: str = OUTLINE, where: str = "col-1" +) -> exporting.Destination: + entry = format_named(format) + assert entry is not None + return exporting.Destination( + session_id=SESSION, target_id=target_id, format=entry, target=where, provider=format + ) + + +class PermanentDocumentError(Exception): + """Recognised by class name, exactly as the production type is. + + `sturnus.application` may not import + `sturnus.infrastructure.documents.outline`, so the production code + matches on `type(exc).__name__`. A local class of the same name is the + honest way to exercise that -- and it is also what proves the match is + by name rather than by identity. + """ + + +class FakeSink: + def __init__(self, url: str = "https://example/doc", fail: Exception | None = None) -> None: + self.url = url + self.fail = fail + self.created: list[tuple[str, str, str]] = [] + + async def create(self, title: str, body: str, where: str) -> CreatedDocument: + if self.fail is not None: + raise self.fail + self.created.append((title, body, where)) + return CreatedDocument(id=f"doc-{len(self.created)}", url=self.url) + + +class FakeSinks: + """Resolves a destination to a sink, one per destination it is given.""" + + def __init__(self, sinks: dict[int | None, FakeSink] | None = None) -> None: + self.sinks = sinks or {} + self.asked: list[exporting.Destination] = [] + + def sink_for(self, place: exporting.Destination) -> FakeSink | None: + self.asked.append(place) + return self.sinks.get(place.target_id) + + +class FakeTargets: + def __init__(self, targets: list[ExportTarget] | None = None) -> None: + self._targets = targets or [] + + async def enabled_for(self, _guild_id: int) -> list[ExportTarget]: + return [t for t in self._targets if t.enabled] + + +class FakeRecords: + def __init__(self, existing: list[SessionDocument] | None = None) -> None: + self.rows = list(existing or []) + self.recorded: list[tuple[int, int, str, str, str]] = [] + + async def for_session(self, session_id: int) -> list[SessionDocument]: + return [row for row in self.rows if row.session_id == session_id] + + async def record( + self, + session_id: int, + *, + target_id: int, + provider: str, + document_id: str, + url: str, + now: datetime, + ) -> None: + assert now == NOW + self.recorded.append((session_id, target_id, provider, document_id, url)) + + +def ports( + sinks: FakeSinks | None = None, + targets: FakeTargets | None = None, + records: FakeRecords | None = None, +) -> exporting.ExportPorts: + return exporting.ExportPorts( + sinks=sinks or FakeSinks(), + targets=targets or FakeTargets(), + documents=records or FakeRecords(), + ) + + +# --------------------------------------------------------------------------- +# Choosing the destinations: pure, and therefore where the rules live +# --------------------------------------------------------------------------- + + +def test_a_guild_with_no_configured_targets_publishes_where_it_always_did() -> None: + """The fallback is the point. `document_target` is what every guild + running today is configured with, there is no migration that moves it + into the new table, and a release that quietly stopped publishing for + all of them would be the worst possible outcome of this change.""" + legacy = destination(None, OUTLINE, "col-legacy") + chosen = exporting.destinations_for(SESSION, [], legacy) + assert chosen == (legacy,) + + +def test_a_guild_with_no_targets_and_no_legacy_setting_publishes_nowhere() -> None: + assert exporting.destinations_for(SESSION, [], None) == () + + +def test_a_configured_target_replaces_the_legacy_setting() -> None: + """Not "as well as". A guild that configured a destination in the table + said where its protocols go; publishing to the old collection too would + be a document nobody asked for and nobody knows about.""" + chosen = exporting.destinations_for(SESSION, [target(4)], destination(None)) + assert [place.target_id for place in chosen] == [4] + + +def test_several_targets_are_ordered_oldest_first() -> None: + """The order is the order they were configured in, which is what makes + the primary stable -- see `test_the_primary_...` below.""" + chosen = exporting.destinations_for( + SESSION, [target(9, name="a"), target(2, name="z"), target(5, name="m")], None + ) + assert [place.target_id for place in chosen] == [2, 5, 9] + + +def test_a_target_naming_a_format_this_deployment_cannot_publish_is_ignored() -> None: + """And ignored *alone*: `guild_export_target.format` is a plain string + precisely so one unrecognised row does not take the guild's other + destinations with it.""" + chosen = exporting.destinations_for( + SESSION, [target(1, format="pdf"), target(2, format=MARKDOWN)], None + ) + assert [place.target_id for place in chosen] == [2] + + +def test_a_destination_carries_the_renderer_its_format_names() -> None: + chosen = exporting.destinations_for(SESSION, [target(1, format=HTML)], None) + assert chosen[0].format.name == HTML + + +def test_the_provider_recorded_for_a_configured_target_is_its_format() -> None: + """`session_document.provider` has to outlive the target it names: once + a destination is removed `target_id` goes null, and a row that could not + say what kind of document it points at would be a URL with no context.""" + chosen = exporting.destinations_for(SESSION, [target(1, format=HTML)], None) + assert chosen[0].provider == HTML + + +# --------------------------------------------------------------------------- +# Publishing to each of them +# --------------------------------------------------------------------------- + + +async def test_every_enabled_destination_receives_the_protocol() -> None: + outline, markdown = FakeSink("https://outline/1"), FakeSink("https://console/2") + sinks = FakeSinks({1: outline, 2: markdown}) + places = exporting.destinations_for( + SESSION, [target(1, OUTLINE), target(2, MARKDOWN, name="archive", where="protocols")], None + ) + report = await exporting.publish_session(SESSION, places, request(), ports(sinks), NOW) + + assert len(outline.created) == 1 + assert len(markdown.created) == 1 + assert len(report.published) == 2 + + +async def test_each_destination_gets_the_body_its_own_format_renders() -> None: + """The whole reason a format is a pair. Two destinations, one + transcript, two different strings -- one with Outline's mention chips + in it and one with none.""" + outline, html = FakeSink(), FakeSink() + places = exporting.destinations_for( + SESSION, [target(1, OUTLINE), target(2, HTML, name="page", where="protocols")], None + ) + await exporting.publish_session( + SESSION, places, request(), ports(FakeSinks({1: outline, 2: html})), NOW + ) + + assert "mention://" in outline.created[0][1] + assert "mention://" not in html.created[0][1] + assert html.created[0][1].lstrip().lower().startswith("<!doctype html>") + + +async def test_each_destination_is_addressed_by_its_own_target() -> None: + first, second = FakeSink(), FakeSink() + places = exporting.destinations_for( + SESSION, + [target(1, OUTLINE, where="col-a"), target(2, OUTLINE, name="b", where="col-b")], + None, + ) + await exporting.publish_session( + SESSION, places, request(), ports(FakeSinks({1: first, 2: second})), NOW + ) + + assert first.created[0][2] == "col-a" + assert second.created[0][2] == "col-b" + + +async def test_two_destinations_of_one_format_render_the_body_once() -> None: + """Rendering is a Jinja pass over every block of a meeting. Doing it + twice for two destinations that would receive the identical string is + work nobody asked for, on a worker whose transcription queue is + waiting on the same process.""" + first, second = FakeSink(), FakeSink() + places = exporting.destinations_for( + SESSION, + [ + target(1, MARKDOWN, name="a", where="one"), + target(2, MARKDOWN, name="b", where="two"), + ], + None, + ) + await exporting.publish_session( + SESSION, places, request(), ports(FakeSinks({1: first, 2: second})), NOW + ) + assert first.created[0][1] == second.created[0][1] + + +async def test_each_published_destination_is_recorded() -> None: + records = FakeRecords() + places = exporting.destinations_for( + SESSION, [target(1, OUTLINE), target(2, MARKDOWN, name="b", where="p")], None + ) + await exporting.publish_session( + SESSION, + places, + request(), + ports(FakeSinks({1: FakeSink(), 2: FakeSink()}), records=records), + NOW, + ) + assert [(row[1], row[2]) for row in records.recorded] == [(1, OUTLINE), (2, MARKDOWN)] + + +async def test_the_legacy_destination_is_not_recorded_as_a_session_document() -> None: + """It has no target row to point at, and `session_document.target_id` + is part of the unique key that makes a re-export overwrite itself. A + row with a null there would be appended on every sweep instead. What + records the legacy destination is `session.document_url`, exactly as it + always was.""" + records = FakeRecords() + report = await exporting.publish_session( + SESSION, + (destination(None),), + request(), + ports(FakeSinks({None: FakeSink()}), records=records), + NOW, + ) + assert records.recorded == [] + assert report.primary is not None + + +# --------------------------------------------------------------------------- +# One failing destination +# --------------------------------------------------------------------------- + + +async def test_one_failing_destination_does_not_lose_the_others() -> None: + working = FakeSink("https://outline/1") + broken = FakeSink(fail=RuntimeError("confluence is down")) + places = exporting.destinations_for( + SESSION, [target(1, MARKDOWN, name="a", where="p"), target(2, OUTLINE, name="b")], None + ) + report = await exporting.publish_session( + SESSION, places, request(), ports(FakeSinks({1: broken, 2: working})), NOW + ) + + assert len(working.created) == 1 + assert report.failed == 1 + assert len(report.published) == 1 + + +async def test_a_destination_that_failed_is_not_recorded() -> None: + """The record is what the retry sweep reads to decide what is left to + do. Recording a failure would mean never trying it again.""" + records = FakeRecords() + with pytest.raises(exporting.NothingPublished): + await exporting.publish_session( + SESSION, + (destination(1),), + request(), + ports(FakeSinks({1: FakeSink(fail=RuntimeError("nope"))}), records=records), + NOW, + ) + assert records.recorded == [] + + +async def test_a_publish_where_nothing_reached_anywhere_raises() -> None: + """So the caller's existing retry path -- `process_one`'s handler and + the sweep -- sees a failure rather than a quiet no-op. A session that + published nowhere must stay `closed` and be tried again.""" + with pytest.raises(exporting.NothingPublished): + await exporting.publish_session( + SESSION, + (destination(1),), + request(), + ports(FakeSinks({1: FakeSink(fail=RuntimeError("nope"))})), + NOW, + ) + + +async def test_a_publish_where_something_reached_somewhere_does_not_raise() -> None: + places = exporting.destinations_for( + SESSION, [target(1, OUTLINE), target(2, MARKDOWN, name="b", where="p")], None + ) + report = await exporting.publish_session( + SESSION, + places, + request(), + ports(FakeSinks({1: FakeSink(), 2: FakeSink(fail=RuntimeError("nope"))})), + NOW, + ) + assert report.failed == 1 + assert len(report.published) == 1 + + +async def test_a_permanent_rejection_is_not_a_failure_to_retry() -> None: + """A deleted collection or a revoked token will not be fixed by trying + again in five minutes. It is counted apart from a transient failure and + does not raise, which is the behaviour the single-destination path + already had.""" + report = await exporting.publish_session( + SESSION, + (destination(1),), + request(), + ports(FakeSinks({1: FakeSink(fail=PermanentDocumentError())})), + NOW, + ) + assert report.rejected == 1 + assert report.failed == 0 + + +async def test_a_destination_whose_sink_this_process_cannot_build_is_a_failure() -> None: + """An object-store destination on a deployment with no object store + configured. Not silently skipped: a guild that configured a destination + and never hears about it again has no way to find out.""" + with pytest.raises(exporting.NothingPublished): + await exporting.publish_session( + SESSION, (destination(1, MARKDOWN, "p"),), request(), ports(FakeSinks({})), NOW + ) + + +async def test_a_failure_beside_a_destination_already_published_does_not_raise() -> None: + """The session has a protocol -- last sweep wrote it -- and one + destination is still down. Reporting that as a failed publish would put + a warning in the log on every sweep for a fault that has already been + survived.""" + report = await exporting.publish_session( + SESSION, + (destination(1), destination(2, MARKDOWN, "p")), + request(), + ports( + FakeSinks({1: FakeSink(), 2: FakeSink(fail=RuntimeError("down"))}), + records=FakeRecords([recorded(1)]), + ), + NOW, + ) + assert report.skipped == 1 + assert report.failed == 1 + + +# --------------------------------------------------------------------------- +# The retry sweep, and the duplicates it must not make +# --------------------------------------------------------------------------- + + +def recorded(target_id: int, url: str = "https://outline/1") -> SessionDocument: + return SessionDocument( + session_id=SESSION, + target_id=target_id, + provider=OUTLINE, + document_id="doc-1", + url=url, + created_at=T0, + ) + + +async def test_a_destination_already_recorded_is_not_published_again() -> None: + """The duplicate this whole mechanism exists to prevent: a flaky + Confluence brings the session back to the sweep every five minutes, and + without this the Outline document is written again on every one of + them.""" + outline, broken = FakeSink(), FakeSink(fail=RuntimeError("still down")) + places = exporting.destinations_for( + SESSION, [target(1, OUTLINE), target(2, MARKDOWN, name="b", where="p")], None + ) + report = await exporting.publish_session( + SESSION, + places, + request(), + ports(FakeSinks({1: outline, 2: broken}), records=FakeRecords([recorded(1)])), + NOW, + ) + assert outline.created == [] + assert report.skipped == 1 + assert report.failed == 1 + + +async def test_only_the_destination_that_failed_is_retried() -> None: + second = FakeSink("https://console/2") + places = exporting.destinations_for( + SESSION, [target(1, OUTLINE), target(2, MARKDOWN, name="b", where="p")], None + ) + report = await exporting.publish_session( + SESSION, + places, + request(), + ports(FakeSinks({1: FakeSink(), 2: second}), records=FakeRecords([recorded(1)])), + NOW, + ) + assert len(second.created) == 1 + assert len(report.published) == 1 + + +async def test_a_sweep_that_had_nothing_left_to_do_does_not_raise() -> None: + """Every destination already recorded is a session that is finished, + not a session that failed everywhere.""" + report = await exporting.publish_session( + SESSION, + (destination(1),), + request(), + ports(FakeSinks({1: FakeSink()}), records=FakeRecords([recorded(1)])), + NOW, + ) + assert report.skipped == 1 + assert report.failed == 0 + + +# --------------------------------------------------------------------------- +# The primary +# --------------------------------------------------------------------------- + + +async def test_the_primary_is_the_destination_the_guild_configured_first() -> None: + """`session.document_url` is what the announcement posts, so exactly + one destination has to be it. The oldest enabled target: it does not + move when somebody renames a destination, and it is the one a guild + that has only ever had one destination has always had.""" + first, second = FakeSink("https://outline/first"), FakeSink("https://console/second") + places = exporting.destinations_for( + SESSION, + [target(9, MARKDOWN, name="added later", where="p"), target(2, OUTLINE, name="wiki")], + None, + ) + report = await exporting.publish_session( + SESSION, places, request(), ports(FakeSinks({2: first, 9: second})), NOW + ) + assert report.primary is not None + assert report.primary.destination.target_id == 2 + assert report.primary.document.url == "https://outline/first" + + +async def test_a_failed_primary_leaves_no_primary_to_announce() -> None: + """So `session.document_url` is not written, the session stays + undocumented, and the sweep brings it back -- with the destination that + already worked skipped.""" + places = exporting.destinations_for( + SESSION, [target(1, OUTLINE), target(2, MARKDOWN, name="b", where="p")], None + ) + report = await exporting.publish_session( + SESSION, + places, + request(), + ports(FakeSinks({1: FakeSink(fail=RuntimeError("down")), 2: FakeSink()})), + NOW, + ) + assert report.primary is None + assert len(report.published) == 1 + + +async def test_a_primary_already_recorded_is_still_reported_as_the_primary() -> None: + """The narrow window where the document was created and recorded but + stamping the session failed. Without this the sweep would skip the + primary for ever and never write `document_url`, so the announcement + would never go out for a session that has a perfectly good document.""" + report = await exporting.publish_session( + SESSION, + (destination(1),), + request(), + ports(FakeSinks({1: FakeSink()}), records=FakeRecords([recorded(1, "https://outline/x")])), + NOW, + ) + assert report.primary is not None + assert report.primary.document.url == "https://outline/x" + + +async def test_publishing_nowhere_at_all_is_not_an_error() -> None: + """A guild that has configured nothing. There is no document to make + and nothing failed; raising would fill the log with a sweep's worth of + warnings about guilds that never asked for a protocol.""" + report = await exporting.publish_session(SESSION, (), request(), ports(), NOW) + assert report.primary is None + assert report.failed == 0 diff --git a/tests/application/test_worker.py b/tests/application/test_worker.py index 37b16fc..5a53572 100644 --- a/tests/application/test_worker.py +++ b/tests/application/test_worker.py @@ -16,10 +16,12 @@ from pathlib import Path from typing import Any -from sturnus.application.documents import CreatedDocument +from sturnus.application.documents import CreatedDocument, DocumentSink +from sturnus.application.exporting import Destination, ExportPorts from sturnus.application.transcription import TranscribedSegment, TranscriptionResult from sturnus.application.worker import process_one, retry_pending_documents from sturnus.domain import settings as domain_settings +from sturnus.domain.exports import ExportTarget, SessionDocument from sturnus.domain.measurements import JobMeasurements, RecordedAudio from sturnus.infrastructure.db.queue import ClaimedJob from sturnus.infrastructure.documents.outline import PermanentDocumentError @@ -195,6 +197,10 @@ def __init__(self) -> None: #: What `closed_undocumented_sessions` reports -- empty by default, #: since most tests never exercise `retry_pending_documents`. self.pending_retry: list[int] = [] + #: The second candidate set the sweep reads -- sessions that + #: published somewhere but not everywhere. Empty by default for + #: the same reason `pending_retry` is. + self.pending_targets: list[int] = [] #: What `guild_id` reports for every session id -- matches #: `FakeConfig`'s default guild (module-level `GUILD`), so #: guild-scoped configuration resolves the same way in both fakes @@ -222,6 +228,9 @@ async def session_bounds(self, _session_id: int) -> tuple[datetime, datetime]: async def closed_undocumented_sessions(self) -> list[int]: return self.pending_retry + async def sessions_with_unpublished_targets(self) -> list[int]: + return self.pending_targets + async def channel_ref(self, _session_id: int) -> tuple[int, int, str | None]: return (self.guild, 4711, "meeting-raum") @@ -307,6 +316,67 @@ def guild_config(extra: dict[str, str] | None = None) -> FakeConfig: return FakeConfig(values) +class FakeExportTargets: + """A guild with nothing in `guild_export_target`. + + The default for every test in this file, and deliberately so: that is + the state of every guild running today, and what these tests pin is + that such a guild still publishes exactly where it always did -- + through `document_target`, to the one sink, recorded on the session + row. The several-destinations behaviour has a file of its own + (`tests/application/test_exporting.py`), where it can be tested without + a queue and a transcription engine in the way. + """ + + async def enabled_for(self, _guild_id: int) -> list[ExportTarget]: + return [] + + +class FakeSessionDocuments: + """`session_document`, which the legacy destination never writes to.""" + + def __init__(self) -> None: + self.recorded: list[tuple[int, int, str, str, str, datetime]] = [] + + async def for_session(self, _session_id: int) -> list[SessionDocument]: + return [] + + async def record( + self, + session_id: int, + *, + target_id: int, + provider: str, + document_id: str, + url: str, + now: datetime, + ) -> None: + self.recorded.append((session_id, target_id, provider, document_id, url, now)) + + +class OneSink: + """Answers the same sink for every destination. + + Which is what a guild with one destination has, and what `documents` + meant when it was a parameter of `process_one` in its own right. + """ + + def __init__(self, sink: DocumentSink) -> None: + self.sink = sink + + def sink_for(self, _destination: Destination) -> DocumentSink: + return self.sink + + +def exports(documents: DocumentSink) -> ExportPorts: + """The publishing collaborators, around one sink and an empty table.""" + return ExportPorts( + sinks=OneSink(documents), + targets=FakeExportTargets(), + documents=FakeSessionDocuments(), + ) + + def job(job_id: int = 1, session_id: int = 1, user_id: int = 100) -> ClaimedJob: return ClaimedJob( id=job_id, @@ -326,7 +396,7 @@ def run(tmp_path: Path, **kw: Any) -> dict[str, Any]: "engine": kw.get("engine") or FakeEngine(), "store": kw.get("store") or FakeStore(), "crypto": kw.get("crypto") or FakeCrypto(), - "documents": kw.get("documents") or FakeDocuments(), + "exports": kw.get("exports") or exports(kw.get("documents") or FakeDocuments()), "sessions": kw.get("sessions") or FakeSessions(), "jobs": kw.get("jobs") or FakeJobs(), "links": kw.get("links") or FakeLinks(), @@ -842,7 +912,7 @@ async def test_retry_pending_documents_retries_closed_undocumented_sessions() -> ) } ) - await retry_pending_documents(documents, sessions, jobs, FakeLinks(), FakeConfig()) + await retry_pending_documents(exports(documents), sessions, jobs, FakeLinks(), FakeConfig()) assert len(documents.created) == 1 assert "hello again" in documents.created[0][1] assert sessions.documented == [(1, "https://outline.example/doc/1", "outline")] @@ -859,13 +929,15 @@ async def test_retry_pending_documents_survives_one_sessions_failure() -> None: {100: TranscriptionResult(segments=(TranscribedSegment(0.0, 1.0, "hi"),), language="de")} ) # must not raise - await retry_pending_documents(documents, sessions, jobs, FakeLinks(), FakeConfig()) + await retry_pending_documents(exports(documents), sessions, jobs, FakeLinks(), FakeConfig()) assert documents.calls == 2 # both sessions were attempted despite failing async def test_retry_pending_documents_does_nothing_when_nothing_is_pending() -> None: documents = FakeDocuments() - await retry_pending_documents(documents, FakeSessions(), FakeJobs(), FakeLinks(), FakeConfig()) + await retry_pending_documents( + exports(documents), FakeSessions(), FakeJobs(), FakeLinks(), FakeConfig() + ) assert documents.created == [] @@ -990,7 +1062,7 @@ async def test_a_finished_job_records_what_its_recording_is(tmp_path: Path) -> N engine=FakeEngine(), store=FakeStore(), crypto=WritesARealWav(sample_rate=16_000, channels=1), - documents=FakeDocuments(), + exports=exports(FakeDocuments()), sessions=FakeSessions(), jobs=FakeJobs(), links=FakeLinks(), @@ -1017,7 +1089,7 @@ async def test_the_size_recorded_is_the_stored_object_and_not_the_plaintext(tmp_ engine=FakeEngine(), store=FakeStore(), crypto=WritesARealWav(frames=8_000), - documents=FakeDocuments(), + exports=exports(FakeDocuments()), sessions=FakeSessions(), jobs=FakeJobs(), links=FakeLinks(), @@ -1048,7 +1120,7 @@ async def test_a_recording_whose_header_cannot_be_read_records_nothing(tmp_path: store=FakeStore(), # Writes `b"RIFFdecoded"`, which is not a WAV. crypto=FakeCrypto(), - documents=FakeDocuments(), + exports=exports(FakeDocuments()), sessions=FakeSessions(), jobs=FakeJobs(), links=FakeLinks(), @@ -1072,7 +1144,7 @@ async def test_a_stereo_recording_is_recorded_as_stereo(tmp_path: Path) -> None: engine=FakeEngine(), store=FakeStore(), crypto=WritesARealWav(sample_rate=48_000, channels=2), - documents=FakeDocuments(), + exports=exports(FakeDocuments()), sessions=FakeSessions(), jobs=FakeJobs(), links=FakeLinks(), @@ -1096,7 +1168,7 @@ async def test_a_failed_transcription_records_nothing_about_the_file(tmp_path: P engine=FakeEngine(fail=True), store=FakeStore(), crypto=WritesARealWav(), - documents=FakeDocuments(), + exports=exports(FakeDocuments()), sessions=FakeSessions(), jobs=FakeJobs(), links=FakeLinks(), @@ -1107,3 +1179,145 @@ async def test_a_failed_transcription_records_nothing_about_the_file(tmp_path: P assert queue.failed assert queue.recorded == [] + + +# --------------------------------------------------------------------------- +# Several destinations, reached through the whole worker +# --------------------------------------------------------------------------- + + +class ConfiguredTargets: + """A guild with rows in `guild_export_target`.""" + + def __init__(self, targets: list[ExportTarget]) -> None: + self._targets = targets + + async def enabled_for(self, _guild_id: int) -> list[ExportTarget]: + return self._targets + + +class RecordingSinks: + """One sink per destination, so a test can see which got what.""" + + def __init__(self, sinks: dict[int | None, FakeDocuments]) -> None: + self._sinks = sinks + + def sink_for(self, place: Destination) -> DocumentSink | None: + return self._sinks.get(place.target_id) + + +def export_target(target_id: int, format: str, where: str, name: str) -> ExportTarget: + return ExportTarget( + id=target_id, + guild_id=GUILD, + format=format, + name=name, + target=where, + config={}, + has_secret=False, + enabled=True, + created_at=T0, + updated_at=T0, + ) + + +async def test_a_configured_target_is_published_to_instead_of_document_target( + tmp_path: Path, +) -> None: + """`guild_export_target` is what an administrator configured through + the console; `document_target` is what was there before it existed. A + guild that has said where its protocols go must not also get a document + in the old collection. + """ + queue = FakeQueue([job()]) + queue.last_is_final = True + configured = FakeDocuments() + records = FakeSessionDocuments() + await process_one( + **run( + tmp_path, + queue=queue, + exports=ExportPorts( + sinks=RecordingSinks({7: configured}), + targets=ConfiguredTargets([export_target(7, "outline", "col-configured", "wiki")]), + documents=records, + ), + ) + ) + assert configured.targets == ["col-configured"] + assert [(row[0], row[1], row[2]) for row in records.recorded] == [(1, 7, "outline")] + + +async def test_two_destinations_both_receive_the_session_protocol(tmp_path: Path) -> None: + queue = FakeQueue([job()]) + queue.last_is_final = True + wiki, archive = FakeDocuments(), FakeDocuments() + sessions = FakeSessions() + await process_one( + **run( + tmp_path, + queue=queue, + sessions=sessions, + exports=ExportPorts( + sinks=RecordingSinks({1: wiki, 2: archive}), + targets=ConfiguredTargets( + [ + export_target(1, "outline", "col-1", "wiki"), + export_target(2, "markdown", "protocols", "archive"), + ] + ), + documents=FakeSessionDocuments(), + ), + ) + ) + assert len(wiki.created) == 1 + assert len(archive.created) == 1 + # The oldest target is the primary, and it is what the announcement + # path reads off the session row. + assert sessions.documented == [(1, "https://outline.example/doc/1", "outline")] + + +async def test_a_session_publishing_nowhere_at_all_is_left_for_the_sweep( + tmp_path: Path, +) -> None: + """A guild with no target row and no `document_target`. The job itself + is done and must not be requeued; the session stays undocumented, which + is exactly what the sweep looks for. + """ + queue = FakeQueue([job()]) + queue.last_is_final = True + sessions = FakeSessions() + config = FakeConfig({(GUILD, domain_settings.DOCUMENT_PROVIDER): "outline"}) + done = await process_one(**run(tmp_path, queue=queue, sessions=sessions, config=config)) + assert done is True + assert queue.failed == [] + assert sessions.documented == [] + + +async def test_the_retry_sweep_also_picks_up_a_session_that_published_only_partly() -> None: + """Outline succeeded, so the session is `documented` and invisible to + `closed_undocumented_sessions`. Without the second candidate set the + failed Markdown export would never be retried at all. + """ + sessions = FakeSessions() + sessions.pending_targets = [3] + documents = FakeDocuments() + await retry_pending_documents( + exports(documents), sessions, FakeJobs(), FakeLinks(), FakeConfig() + ) + assert len(documents.created) == 1 + + +async def test_a_session_in_both_candidate_sets_is_published_once() -> None: + """A session can answer both questions at once, and publishing it twice + in one sweep would be the duplicate this whole mechanism exists to + prevent. + """ + sessions = FakeSessions() + sessions.pending_retry = [3] + sessions.pending_targets = [3] + documents = FakeDocuments() + await retry_pending_documents( + exports(documents), sessions, FakeJobs(), FakeLinks(), FakeConfig() + ) + assert len(documents.created) == 1 diff --git a/tests/console/conftest.py b/tests/console/conftest.py index b9ce4fe..8336520 100644 --- a/tests/console/conftest.py +++ b/tests/console/conftest.py @@ -9,9 +9,18 @@ from __future__ import annotations -from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Sequence +from collections.abc import ( + AsyncGenerator, + AsyncIterator, + Awaitable, + Callable, + Mapping, + Sequence, +) +from dataclasses import replace from datetime import UTC, date, datetime, timedelta from pathlib import Path +from typing import Any import pytest from aiohttp import web @@ -29,7 +38,9 @@ ConsentDirectory, ConsentHolder, ConsentPage, + DocumentArtefacts, DownloadableTrack, + ExportTargets, GuildDirectory, GuildNames, GuildQueue, @@ -48,6 +59,7 @@ RequeueOutcome, RevocationOutcome, ScopeOutcome, + SessionDocumentDirectory, SessionNaming, SessionReads, SettingsStore, @@ -56,7 +68,7 @@ Track, TranscriptReader, ) -from sturnus.console.session import SessionCookie +from sturnus.console.session import SessionCookie, SignedSession from sturnus.console.statistics import ( AttendedSession, SessionName, @@ -65,6 +77,7 @@ TagUse, ) from sturnus.domain import preferences +from sturnus.domain.exports import ExportTarget, SessionDocument from sturnus.infrastructure.crypto import CHUNK_SIZE, encrypt_file from sturnus.infrastructure.documents.outline_oauth import ExternalIdentity, LinkExchangeError @@ -195,6 +208,19 @@ def now_at(moment: datetime = T0) -> Callable[[], datetime]: return lambda: moment +def signed_cookie(discord_user_id: int = ANNA, moment: datetime = T0) -> str: + """A session cookie for one person, signed with the test secret. + + Here rather than in each route test file: `build_test_api` defaults to + a `SessionCookie` built from `SECRET`, so every module that wants a + signed-in client needs exactly this and three of them had already + written their own. + """ + return SessionCookie(SECRET, timedelta(hours=12)).issue( + SignedSession(discord_user_id), now=moment + ) + + class FakeReads: """The console's reads, in memory. @@ -880,6 +906,123 @@ async def rename( return self.stored[session_id] +class FakeExportTargets: + """A guild's destinations, in memory, keyed the way the table is. + + Structurally an `ExportTargets`, and it keeps the store's two rules + that the API depends on: `save` upserts on `(guild_id, name)`, and the + secret is written only by `set_secret`. `secret_for` deliberately does + not exist -- the port has no such method, so a handler that reached for + one would not compile, and a double that offered one would be the one + place that stopped being true. + """ + + def __init__(self) -> None: + self.rows: dict[int, ExportTarget] = {} + #: What was actually stored, so a test can prove a credential + #: reached the store while never appearing in a response. + self.secrets: dict[int, str | None] = {} + self._next_id = 1 + + async def all_for(self, guild_id: int) -> tuple[ExportTarget, ...]: + return tuple( + sorted( + (row for row in self.rows.values() if row.guild_id == guild_id), + key=lambda row: (row.name, row.id), + ) + ) + + async def get(self, guild_id: int, target_id: int) -> ExportTarget | None: + row = self.rows.get(target_id) + return row if row is not None and row.guild_id == guild_id else None + + async def save( + self, + guild_id: int, + *, + format: str, + name: str, + target: str, + config: Mapping[str, Any], + enabled: bool = True, + now: datetime, + ) -> int: + existing = next( + (r for r in self.rows.values() if r.guild_id == guild_id and r.name == name), None + ) + target_id = existing.id if existing is not None else self._next_id + if existing is None: + self._next_id += 1 + self.rows[target_id] = ExportTarget( + id=target_id, + guild_id=guild_id, + format=format, + name=name, + target=target, + config=dict(config), + has_secret=self.secrets.get(target_id) is not None, + enabled=enabled, + created_at=existing.created_at if existing is not None else now, + updated_at=now, + ) + return target_id + + async def delete(self, guild_id: int, target_id: int) -> bool: + if await self.get(guild_id, target_id) is None: + return False + del self.rows[target_id] + self.secrets.pop(target_id, None) + return True + + async def set_secret( + self, guild_id: int, target_id: int, secret: str | None, now: datetime + ) -> bool: + row = await self.get(guild_id, target_id) + if row is None: + return False + self.secrets[target_id] = secret + self.rows[target_id] = replace(row, has_secret=secret is not None, updated_at=now) + return True + + +class FakeSessionDocuments: + """What a session published, in memory, keyed by session id. + + Deliberately unscoped, exactly as `FakeTranscripts` above is and for + the same reason: `SessionDocumentDirectory` carries no `requested_by` + because the participant rule is the handler's `session_for` call. What + the route tests use this for is the other half -- that the handler + asked the reads adapter first and only then came here, which is + visible in `asked` and in the 404 a fake with no session produces. + """ + + def __init__(self, documents: dict[int, list[SessionDocument]] | None = None) -> None: + self.documents = documents if documents is not None else {} + #: Every session this was asked about, in order. + self.asked: list[int] = [] + + async def documents_of(self, session_id: int) -> tuple[SessionDocument, ...] | None: + self.asked.append(session_id) + found = self.documents.get(session_id) + return None if found is None else tuple(found) + + async def document_of(self, session_id: int, target_id: int) -> SessionDocument | None: + found = await self.documents_of(session_id) + if found is None: + return None + return next((row for row in found if row.target_id == target_id), None) + + +class FakeArtefacts: + """The object store, as a dictionary.""" + + def __init__(self, objects: dict[str, bytes] | None = None) -> None: + self.objects = objects or {} + + async def get(self, key: str) -> bytes: + return self.objects[key] + + def build_test_api( *, oauth: OAuthClient | None = None, @@ -901,6 +1044,9 @@ def build_test_api( collections: CollectionNames | None = None, transcripts: TranscriptReader | None = None, naming: SessionNaming | None = None, + exports: ExportTargets | None = None, + documents: SessionDocumentDirectory | None = None, + artefacts: DocumentArtefacts | None = None, sessions: SessionCookie | None = None, now: Callable[[], datetime] | None = None, schema_ready: bool = True, @@ -951,6 +1097,9 @@ def build_test_api( collections=collections or FakeCollections(), transcripts=transcripts or FakeTranscripts(), naming=naming or FakeNaming(), + exports=exports or FakeExportTargets(), + documents=documents or FakeSessionDocuments(), + artefacts=artefacts or FakeArtefacts(), sessions=sessions or SessionCookie(SECRET, timedelta(hours=12)), now=now or now_at(), schema_ready=lambda: schema_ready, diff --git a/tests/console/test_adapters.py b/tests/console/test_adapters.py index 54f6459..f070bcf 100644 --- a/tests/console/test_adapters.py +++ b/tests/console/test_adapters.py @@ -16,6 +16,7 @@ from sturnus.console.adapters import ( ConsoleLinkDirectory, ConsoleProfileDirectory, + ConsoleSessionDocuments, ConsoleSessionNaming, ConsoleStateStore, ConsoleTagWriter, @@ -24,8 +25,10 @@ ) from sturnus.console.statistics import SessionName from sturnus.domain import settings +from sturnus.infrastructure.crypto import KeyWrapper from sturnus.infrastructure.db.admin_members import AdminMemberStore from sturnus.infrastructure.db.config_store import ConfigStore +from sturnus.infrastructure.db.export_targets import ExportTargetStore from sturnus.infrastructure.db.models import ( Base, Session, @@ -34,6 +37,7 @@ TranscriptionJob, ) from sturnus.infrastructure.db.repositories import AccountLinkRepository +from sturnus.infrastructure.db.session_documents import SessionDocumentStore T0 = datetime(2026, 8, 21, 12, 0, 0, tzinfo=UTC) ANNA, BEN = 100, 200 @@ -876,3 +880,93 @@ async def test_a_guilds_merge_gap_decides_where_the_paragraphs_break( # with each other anyway -- what this asserts is that a configured # value is read at all rather than silently defaulted. assert len(found.blocks) == 2 + + +# --------------------------------------------------------------------------- +# The protocols a session produced, and the rule they sit behind +# --------------------------------------------------------------------------- + + +async def _publish( + factory: async_sessionmaker[AsyncSession], + session_id: int, + target_id: int, + provider: str = "markdown", +) -> None: + await SessionDocumentStore(factory).record( + session_id, + target_id=target_id, + provider=provider, + document_id=f"protocols/{session_id}/{target_id}.md", + url=f"https://sturnus.example/api/sessions/{session_id}/documents/{target_id}", + now=T0, + ) + + +async def _target(factory: async_sessionmaker[AsyncSession], name: str = "archive") -> int: + return await ExportTargetStore(factory, KeyWrapper(b"m" * 32, "master-1")).save( + GUILD, format="markdown", name=name, target="protocols", config={}, now=T0 + ) + + +async def test_a_sessions_protocols_read_back_in_publication_order( + factory: async_sessionmaker[AsyncSession], +) -> None: + session_id = await seed_session(factory, participants=(ANNA, BEN)) + target_id = await _target(factory) + await _publish(factory, session_id, target_id) + + found = await ConsoleSessionDocuments(factory).documents_of(session_id) + + assert found is not None + assert [row.target_id for row in found] == [target_id] + + +async def test_a_session_that_published_nothing_reads_back_an_empty_list( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Empty is a real answer and `None` means "no such session", and the + two must not collapse: a meeting still being transcribed would + otherwise 404 for the people who were in it. The participant rule is + the caller's `session_for` and is not asked here -- see + `sturnus.console.ports.SessionDocumentDirectory`. + """ + session_id = await seed_session(factory, participants=(ANNA,)) + + assert await ConsoleSessionDocuments(factory).documents_of(session_id) == () + + +async def test_a_session_that_does_not_exist_produced_no_protocols( + factory: async_sessionmaker[AsyncSession], +) -> None: + assert await ConsoleSessionDocuments(factory).documents_of(999) is None + + +async def test_one_destination_is_reachable_by_its_own_id( + factory: async_sessionmaker[AsyncSession], +) -> None: + session_id = await seed_session(factory, participants=(ANNA,)) + first, second = await _target(factory, "one"), await _target(factory, "two") + await _publish(factory, session_id, first) + await _publish(factory, session_id, second) + + found = await ConsoleSessionDocuments(factory).document_of(session_id, second) + + assert found is not None + assert found.target_id == second + + +async def test_another_sessions_document_is_not_reachable_through_this_one( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The session is part of the lookup. A target id is a guild's, not a + session's, so the same one appears on every session that guild records + -- and the handler's `session_for` authorises the session in the path, + not the one the row happens to belong to. + """ + mine = await seed_session(factory, participants=(ANNA,)) + theirs = await seed_session(factory, participants=(BEN,)) + target_id = await _target(factory) + await _publish(factory, theirs, target_id) + + assert await ConsoleSessionDocuments(factory).document_of(mine, target_id) is None diff --git a/tests/console/test_document_routes.py b/tests/console/test_document_routes.py new file mode 100644 index 0000000..f04cdd1 --- /dev/null +++ b/tests/console/test_document_routes.py @@ -0,0 +1,296 @@ +"""The protocol endpoints, and the rule that makes an object-store URL safe. + +This route is the reason `CreatedDocument.url` for an object-store +destination is a console path rather than a presigned S3 URL: a presigned +URL is checked once when it is issued, this is checked on every request. The +tests that matter here are the refusals. + +The gate is `SessionReads.session_for`, the same call `/api/sessions/{id}` +and `/api/sessions/{id}/transcript` are served from, so `FakeReads` is what +decides who is in a session here -- not a participant set on the document +double, which would be a second implementation of the rule inside the tests +that are supposed to prove there is only one. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest +from aiohttp import web + +from sturnus.console.statistics import AttendedSession, Participant +from sturnus.domain.exports import SessionDocument +from tests.console.conftest import ( + ANNA, + BEN, + AiohttpClientFactory, + FakeArtefacts, + FakeReads, + FakeSessionDocuments, + build_test_api, + signed_cookie, +) + +SESSION_COOKIE = "sturnus_session" +SESSION = 42 +T0 = datetime(2026, 8, 21, 12, 0, 0, tzinfo=UTC) + +MARKDOWN_KEY = "protocols/42/7.md" +HTML_KEY = "protocols/42/8.html" + + +def document( + target_id: int | None, provider: str, document_id: str, url: str = "https://console/x" +) -> SessionDocument: + return SessionDocument( + session_id=SESSION, + target_id=target_id, + provider=provider, + document_id=document_id, + url=url, + created_at=T0, + ) + + +def attended() -> AttendedSession: + """The one session these tests are about, as the reads adapter sees it. + + `FakeReads` does not scope -- scoping is SQL and is tested against the + real database -- so "Ben was not in it" is expressed by signing Ben in + against a `FakeReads` that holds no session at all, which is exactly + what the real `session_for` answers him. + """ + return AttendedSession( + id=SESSION, + channel_id=555, + channel_name="meeting", + started_at=T0, + ended_at=None, + document_url=None, + participants=(Participant(ANNA, "anna"),), + tracks=(), + title=None, + description=None, + ) + + +def api( + documents: list[SessionDocument] | None = None, + objects: dict[str, bytes] | None = None, + reads: FakeReads | None = None, +) -> web.Application: + """The console with one session Anna may read. + + Who may read it is `FakeReads`' answer, because + `SessionReads.session_for` is the gate the route actually uses -- the + same call `/api/sessions/{id}` and the transcript endpoint are served + from. A participant set on the document double would be a second + implementation of the rule inside the tests meant to prove there is + only one. + """ + return build_test_api( + reads=reads if reads is not None else FakeReads(sessions=(attended(),)), + documents=FakeSessionDocuments({SESSION: list(documents or [])}), + artefacts=FakeArtefacts(objects or {}), + ) + + +@pytest.fixture +def cookies() -> dict[str, str]: + return {SESSION_COOKIE: signed_cookie(ANNA)} + + +# --------------------------------------------------------------------------- +# The listing +# --------------------------------------------------------------------------- + + +async def test_a_participant_sees_every_protocol_the_session_produced( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client( + api( + [ + document(3, "outline", "doc-1", "https://outline.example/doc/1"), + document(7, "markdown", MARKDOWN_KEY, "https://console/api/..."), + ] + ) + ) + response = await client.get(f"/api/sessions/{SESSION}/documents", cookies=cookies) + assert response.status == 200 + body = await response.json() + assert body["session_id"] == str(SESSION) + assert [d["provider"] for d in body["documents"]] == ["outline", "markdown"] + + +async def test_the_listing_says_which_protocols_this_process_can_serve( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """An Outline document's bytes are in Outline; its `url` is what the + console links to. A stored one is served from here.""" + client = await aiohttp_client( + api([document(3, "outline", "doc-1"), document(7, "markdown", MARKDOWN_KEY)]) + ) + body = await (await client.get(f"/api/sessions/{SESSION}/documents", cookies=cookies)).json() + assert [d["readable"] for d in body["documents"]] == [False, True] + + +async def test_a_document_whose_destination_was_removed_is_still_listed( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """`session_document.target_id` is `ON DELETE SET NULL`: removing a + destination is "stop publishing here", not "forget what was + published".""" + client = await aiohttp_client(api([document(None, "outline", "doc-1")])) + body = await (await client.get(f"/api/sessions/{SESSION}/documents", cookies=cookies)).json() + assert body["documents"][0]["target_id"] is None + + +async def test_a_session_that_published_nothing_lists_nothing( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """An empty list is a real answer -- a meeting still being transcribed + -- and is not the same as the 404 somebody outside the session gets.""" + client = await aiohttp_client(api([])) + response = await client.get(f"/api/sessions/{SESSION}/documents", cookies=cookies) + assert response.status == 200 + assert (await response.json())["documents"] == [] + + +async def test_somebody_outside_the_session_cannot_list_its_protocols( + aiohttp_client: AiohttpClientFactory, +) -> None: + client = await aiohttp_client(api([document(7, "markdown", MARKDOWN_KEY)], reads=FakeReads())) + response = await client.get( + f"/api/sessions/{SESSION}/documents", cookies={SESSION_COOKIE: signed_cookie(BEN)} + ) + assert response.status == 404 + assert await response.json() == {"error": "no such document"} + + +# --------------------------------------------------------------------------- +# The artefact +# --------------------------------------------------------------------------- + + +async def test_a_participant_reads_the_stored_protocol( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client( + api([document(7, "markdown", MARKDOWN_KEY)], {MARKDOWN_KEY: b"# Minutes\n"}) + ) + response = await client.get(f"/api/sessions/{SESSION}/documents/7", cookies=cookies) + assert response.status == 200 + assert await response.text() == "# Minutes\n" + assert response.headers["Content-Type"].startswith("text/markdown") + + +async def test_the_html_protocol_is_served_as_html( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client( + api([document(8, "html", HTML_KEY)], {HTML_KEY: b"<!doctype html><p>hi</p>"}) + ) + response = await client.get(f"/api/sessions/{SESSION}/documents/8", cookies=cookies) + assert response.headers["Content-Type"].startswith("text/html") + + +async def test_a_protocol_is_never_cached_by_a_shared_cache( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """The meeting written down, on its way to one named reader. A shared + cache holding a copy would hand it to the next person through the same + proxy -- exactly the audience the check exists to exclude.""" + client = await aiohttp_client( + api([document(7, "markdown", MARKDOWN_KEY)], {MARKDOWN_KEY: b"x"}) + ) + response = await client.get(f"/api/sessions/{SESSION}/documents/7", cookies=cookies) + assert response.headers["Cache-Control"] == "private, no-store" + + +async def test_the_html_protocol_is_served_under_a_policy_that_lets_it_do_nothing( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """It is a page served from the console's own origin. The template + fetches nothing and runs nothing, and this says so to the browser too.""" + client = await aiohttp_client(api([document(8, "html", HTML_KEY)], {HTML_KEY: b"<p>hi</p>"})) + response = await client.get(f"/api/sessions/{SESSION}/documents/8", cookies=cookies) + policy = response.headers["Content-Security-Policy"] + assert "default-src 'none'" in policy + assert "sandbox" in policy + assert response.headers["X-Content-Type-Options"] == "nosniff" + + +async def test_somebody_outside_the_session_cannot_read_its_protocol( + aiohttp_client: AiohttpClientFactory, +) -> None: + """The rule this route exists to enforce. A presigned S3 URL would + have answered this request.""" + client = await aiohttp_client( + api( + [document(7, "markdown", MARKDOWN_KEY)], + {MARKDOWN_KEY: b"secret minutes"}, + reads=FakeReads(), + ) + ) + response = await client.get( + f"/api/sessions/{SESSION}/documents/7", cookies={SESSION_COOKIE: signed_cookie(BEN)} + ) + assert response.status == 404 + assert "secret minutes" not in await response.text() + + +async def test_an_outline_document_is_not_served_from_here( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """Its bytes live in Outline. The listing carries its URL so the + console can link straight out.""" + client = await aiohttp_client(api([document(3, "outline", "doc-1")])) + response = await client.get(f"/api/sessions/{SESSION}/documents/3", cookies=cookies) + assert response.status == 404 + + +async def test_a_destination_this_session_never_reached_is_a_404( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client(api([document(7, "markdown", MARKDOWN_KEY)])) + response = await client.get(f"/api/sessions/{SESSION}/documents/99", cookies=cookies) + assert response.status == 404 + + +async def test_a_row_whose_object_is_gone_is_a_404_and_not_a_500( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """The row outlived its object. Nothing is broken -- the same reading + the audio route gives a recording the retention sweep erased.""" + client = await aiohttp_client(api([document(7, "markdown", MARKDOWN_KEY)], {})) + response = await client.get(f"/api/sessions/{SESSION}/documents/7", cookies=cookies) + assert response.status == 404 + + +async def test_a_document_of_a_format_this_deployment_cannot_read_is_a_404( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """A row written by a future release that knows `pdf`, read by one + that does not. It must not be served as some other media type, and it + must not be a 500.""" + client = await aiohttp_client(api([document(9, "pdf", "protocols/42/9.pdf")])) + response = await client.get(f"/api/sessions/{SESSION}/documents/9", cookies=cookies) + assert response.status == 404 + + +@pytest.mark.parametrize("path", ["/api/sessions/abc/documents", "/api/sessions/42/documents/xyz"]) +async def test_a_path_segment_that_is_not_a_number_is_a_404( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str], path: str +) -> None: + client = await aiohttp_client(api([])) + assert (await client.get(path, cookies=cookies)).status == 404 + + +@pytest.mark.parametrize("path", ["/api/sessions/42/documents", "/api/sessions/42/documents/7"]) +async def test_neither_route_answers_without_a_session( + aiohttp_client: AiohttpClientFactory, path: str +) -> None: + client = await aiohttp_client(api([document(7, "markdown", MARKDOWN_KEY)])) + assert (await client.get(path)).status == 401 diff --git a/tests/console/test_export_routes.py b/tests/console/test_export_routes.py new file mode 100644 index 0000000..773f8a3 --- /dev/null +++ b/tests/console/test_export_routes.py @@ -0,0 +1,523 @@ +"""The export-target endpoints, and the one thing they must never do. + +The assertion this file exists for is `test_no_response_anywhere_carries_the +_credential`. Everything else here is CRUD; that one is the reason export +destinations are a table of their own rather than `guild_config` keys, and +it is written so that a future field on `ExportTarget` cannot slip a token +back into a response without failing it. +""" + +from __future__ import annotations + +import pytest +from aiohttp import web + +from tests.console.conftest import ( + ANNA, + BEN, + GUILD, + AiohttpClientFactory, + FakeAdmins, + FakeExportTargets, + build_test_api, + signed_cookie, +) + +SESSION_COOKIE = "sturnus_session" + +#: A second guild, because one guild cannot demonstrate a per-guild rule. +OTHER_GUILD = 8822 + +#: A snowflake past 2^53, where a JSON number loses its last digits. +BIG_GUILD = 386950399101370374 + +TOKEN = "confluence-token-nobody-may-read" + + +def api(targets: FakeExportTargets | None = None) -> web.Application: + """The console, with Anna administering `GUILD` and Ben administering nothing.""" + return build_test_api(admins=FakeAdmins({ANNA}), exports=targets or FakeExportTargets()) + + +def outline_body(name: str = "wiki", **overrides: object) -> dict[str, object]: + body: dict[str, object] = { + "format": "outline", + "name": name, + "target": "c9a1b2e3-4f5a-4b3c-8d2e-1a2b3c4d5e6f", + } + body.update(overrides) + return body + + +@pytest.fixture +def cookies() -> dict[str, str]: + return {SESSION_COOKIE: signed_cookie(ANNA)} + + +# --------------------------------------------------------------------------- +# Reading and writing +# --------------------------------------------------------------------------- + + +async def test_a_guild_with_no_destinations_lists_none( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """An empty list rather than a refusal: that is what a guild looks like + before anybody has configured anything.""" + client = await aiohttp_client(api()) + response = await client.get(f"/api/guilds/{GUILD}/export-targets", cookies=cookies) + assert response.status == 200 + assert await response.json() == {"guild_id": str(GUILD), "targets": []} + + +async def test_creating_a_destination_answers_with_it( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client(api()) + response = await client.post( + f"/api/guilds/{GUILD}/export-targets", json=outline_body(), cookies=cookies + ) + assert response.status == 201 + body = await response.json() + assert body["format"] == "outline" + assert body["name"] == "wiki" + assert body["enabled"] is True + assert body["has_secret"] is False + assert body["config"] == {} + + +async def test_a_guild_id_is_a_string_in_every_response( + aiohttp_client: AiohttpClientFactory, +) -> None: + """A snowflake exceeds JavaScript's safe integer range, where a JSON + number silently loses its last digits and names nobody.""" + client = await aiohttp_client(build_test_api(admins=FakeAdmins(by_guild={BIG_GUILD: {ANNA}}))) + response = await client.post( + f"/api/guilds/{BIG_GUILD}/export-targets", + json=outline_body(), + cookies={SESSION_COOKIE: signed_cookie(ANNA)}, + ) + assert (await response.json())["guild_id"] == str(BIG_GUILD) + + +async def test_a_second_destination_of_the_same_name_is_refused( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """The store would upsert on `(guild_id, name)`, which is right for an + update and wrong for a create: a typo would redirect a guild's + protocols with nothing said.""" + client = await aiohttp_client(api()) + await client.post(f"/api/guilds/{GUILD}/export-targets", json=outline_body(), cookies=cookies) + again = await client.post( + f"/api/guilds/{GUILD}/export-targets", json=outline_body(), cookies=cookies + ) + assert again.status == 409 + + +async def test_updating_a_destination_keeps_its_name( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """A name is how an administrator refers to a destination. Changing one + silently under an id is how "publish to Wiki" stops meaning what the + person who set it up thought it meant.""" + client = await aiohttp_client(api()) + created = await ( + await client.post( + f"/api/guilds/{GUILD}/export-targets", json=outline_body(), cookies=cookies + ) + ).json() + updated = await client.put( + f"/api/guilds/{GUILD}/export-targets/{created['id']}", + json=outline_body(name="renamed", target="col-2", enabled=False), + cookies=cookies, + ) + body = await updated.json() + assert updated.status == 200 + assert body["name"] == "wiki" + assert body["target"] == "col-2" + assert body["enabled"] is False + + +async def test_updating_a_destination_that_does_not_exist_is_a_404( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client(api()) + response = await client.put( + f"/api/guilds/{GUILD}/export-targets/99", json=outline_body(), cookies=cookies + ) + assert response.status == 404 + + +async def test_deleting_a_destination_removes_it( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client(api()) + created = await ( + await client.post( + f"/api/guilds/{GUILD}/export-targets", json=outline_body(), cookies=cookies + ) + ).json() + removed = await client.delete( + f"/api/guilds/{GUILD}/export-targets/{created['id']}", cookies=cookies + ) + assert removed.status == 204 + listed = await (await client.get(f"/api/guilds/{GUILD}/export-targets", cookies=cookies)).json() + assert listed["targets"] == [] + + +async def test_deleting_a_destination_that_does_not_exist_is_a_404( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client(api()) + response = await client.delete(f"/api/guilds/{GUILD}/export-targets/99", cookies=cookies) + assert response.status == 404 + + +async def test_a_disabled_destination_is_still_listed( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """Switching a destination off is not the same as forgetting how it was + configured, and a page that hid it would leave no way to switch it back + on.""" + client = await aiohttp_client(api()) + await client.post( + f"/api/guilds/{GUILD}/export-targets", + json=outline_body(enabled=False), + cookies=cookies, + ) + listed = await (await client.get(f"/api/guilds/{GUILD}/export-targets", cookies=cookies)).json() + assert [t["enabled"] for t in listed["targets"]] == [False] + + +# --------------------------------------------------------------------------- +# The credential +# --------------------------------------------------------------------------- + + +async def test_a_credential_can_be_written( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + targets = FakeExportTargets() + client = await aiohttp_client(api(targets)) + created = await ( + await client.post( + f"/api/guilds/{GUILD}/export-targets", json=outline_body(), cookies=cookies + ) + ).json() + response = await client.put( + f"/api/guilds/{GUILD}/export-targets/{created['id']}/secret", + json={"secret": TOKEN}, + cookies=cookies, + ) + assert response.status == 200 + assert (await response.json())["has_secret"] is True + assert targets.secrets[created["id"]] == TOKEN + + +async def test_a_credential_can_be_cleared( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + targets = FakeExportTargets() + client = await aiohttp_client(api(targets)) + created = await ( + await client.post( + f"/api/guilds/{GUILD}/export-targets", json=outline_body(), cookies=cookies + ) + ).json() + await client.put( + f"/api/guilds/{GUILD}/export-targets/{created['id']}/secret", + json={"secret": TOKEN}, + cookies=cookies, + ) + cleared = await client.put( + f"/api/guilds/{GUILD}/export-targets/{created['id']}/secret", + json={"secret": None}, + cookies=cookies, + ) + assert (await cleared.json())["has_secret"] is False + assert targets.secrets[created["id"]] is None + + +async def test_no_response_anywhere_carries_the_credential( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """The whole reason export destinations are not `guild_config` keys. + + Every response this API can produce for a destination that has a + credential, checked against the credential itself -- not against a + field name, so a future field that happened to carry it would fail + here rather than ship. + """ + targets = FakeExportTargets() + client = await aiohttp_client(api(targets)) + created = await ( + await client.post( + f"/api/guilds/{GUILD}/export-targets", json=outline_body(), cookies=cookies + ) + ).json() + stored = await client.put( + f"/api/guilds/{GUILD}/export-targets/{created['id']}/secret", + json={"secret": TOKEN}, + cookies=cookies, + ) + listed = await client.get(f"/api/guilds/{GUILD}/export-targets", cookies=cookies) + updated = await client.put( + f"/api/guilds/{GUILD}/export-targets/{created['id']}", + json=outline_body(target="col-3"), + cookies=cookies, + ) + + for response in (stored, listed, updated): + text = await response.text() + assert TOKEN not in text + # Not even a prefix of it: "masked but recoverable" is recoverable. + assert TOKEN[:8] not in text + + +async def test_updating_a_destination_does_not_clear_its_credential( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """The edit form cannot render the token, so it cannot re-submit it + either -- and a `PUT` that also wrote the credential would therefore + clear it every time somebody changed a collection id.""" + targets = FakeExportTargets() + client = await aiohttp_client(api(targets)) + created = await ( + await client.post( + f"/api/guilds/{GUILD}/export-targets", json=outline_body(), cookies=cookies + ) + ).json() + await client.put( + f"/api/guilds/{GUILD}/export-targets/{created['id']}/secret", + json={"secret": TOKEN}, + cookies=cookies, + ) + updated = await client.put( + f"/api/guilds/{GUILD}/export-targets/{created['id']}", + json=outline_body(target="col-3"), + cookies=cookies, + ) + assert (await updated.json())["has_secret"] is True + assert targets.secrets[created["id"]] == TOKEN + + +async def test_an_empty_credential_is_refused( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """`""` is not a credential and is not "clear it" either -- `null` is. + Storing an empty string would leave `has_secret` true for a + destination that cannot authenticate.""" + client = await aiohttp_client(api()) + created = await ( + await client.post( + f"/api/guilds/{GUILD}/export-targets", json=outline_body(), cookies=cookies + ) + ).json() + response = await client.put( + f"/api/guilds/{GUILD}/export-targets/{created['id']}/secret", + json={"secret": ""}, + cookies=cookies, + ) + assert response.status == 400 + + +async def test_a_credential_for_a_destination_that_does_not_exist_is_a_404( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client(api()) + response = await client.put( + f"/api/guilds/{GUILD}/export-targets/99/secret", + json={"secret": TOKEN}, + cookies=cookies, + ) + assert response.status == 404 + + +# --------------------------------------------------------------------------- +# Authorisation: 404, and the same 404 every time +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "method,path,body", + [ + ("get", "", None), + ("post", "", {"format": "outline", "name": "n", "target": "t"}), + ("put", "/1", {"format": "outline", "name": "n", "target": "t"}), + ("delete", "/1", None), + ("put", "/1/secret", {"secret": "x"}), + ], +) +async def test_somebody_who_does_not_administer_the_guild_gets_a_404( + aiohttp_client: AiohttpClientFactory, + method: str, + path: str, + body: dict[str, object] | None, +) -> None: + """404 and not 403. A 403 confirms that the guild exists and that it + has a destination with that id, to somebody the system has just + decided has no business knowing either.""" + client = await aiohttp_client(api()) + call = getattr(client, method) + response = await call( + f"/api/guilds/{GUILD}/export-targets{path}", + json=body, + cookies={SESSION_COOKIE: signed_cookie(BEN)}, + ) + assert response.status == 404 + assert await response.json() == {"error": "no such export target"} + + +async def test_an_administrator_of_one_guild_is_nobody_in_another( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client(api()) + response = await client.get(f"/api/guilds/{OTHER_GUILD}/export-targets", cookies=cookies) + assert response.status == 404 + + +async def test_a_destination_of_another_guild_is_not_reachable_by_its_id( + aiohttp_client: AiohttpClientFactory, +) -> None: + """The guild is part of every lookup, so a target id belonging to + somebody else reads exactly like a target id belonging to nobody.""" + targets = FakeExportTargets() + client = await aiohttp_client( + build_test_api( + admins=FakeAdmins(by_guild={GUILD: {ANNA}, OTHER_GUILD: {BEN}}), exports=targets + ) + ) + created = await ( + await client.post( + f"/api/guilds/{OTHER_GUILD}/export-targets", + json=outline_body(), + cookies={SESSION_COOKIE: signed_cookie(BEN)}, + ) + ).json() + response = await client.get( + f"/api/guilds/{GUILD}/export-targets", cookies={SESSION_COOKIE: signed_cookie(ANNA)} + ) + assert (await response.json())["targets"] == [] + stolen = await client.delete( + f"/api/guilds/{GUILD}/export-targets/{created['id']}", + cookies={SESSION_COOKIE: signed_cookie(ANNA)}, + ) + assert stolen.status == 404 + + +async def test_a_guild_id_that_is_not_a_number_is_a_404( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client(api()) + response = await client.get("/api/guilds/not-a-guild/export-targets", cookies=cookies) + assert response.status == 404 + + +async def test_a_target_id_that_is_not_a_number_is_a_404( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client(api()) + response = await client.delete(f"/api/guilds/{GUILD}/export-targets/abc", cookies=cookies) + assert response.status == 404 + + +async def test_every_route_refuses_a_request_with_no_session( + aiohttp_client: AiohttpClientFactory, +) -> None: + client = await aiohttp_client(api()) + assert (await client.get(f"/api/guilds/{GUILD}/export-targets")).status == 401 + + +# --------------------------------------------------------------------------- +# What may be configured +# --------------------------------------------------------------------------- + + +async def test_a_format_this_deployment_cannot_publish_is_refused( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """`pdf` is specified and deliberately not built. Accepting a target + for it would produce a destination that is silently skipped after + every meeting, with nothing anywhere saying why.""" + client = await aiohttp_client(api()) + response = await client.post( + f"/api/guilds/{GUILD}/export-targets", + json=outline_body(format="pdf"), + cookies=cookies, + ) + assert response.status == 400 + body = await response.json() + assert set(body["supported"]) == {"outline", "markdown", "html"} + + +async def test_an_unknown_format_is_refused( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client(api()) + response = await client.post( + f"/api/guilds/{GUILD}/export-targets", + json=outline_body(format="smoke-signal"), + cookies=cookies, + ) + assert response.status == 400 + + +async def test_an_object_store_target_that_climbs_out_of_its_prefix_is_refused( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """The target becomes part of an object key, and the format is what + decides what a key may say -- so this refusal comes from the registry + entry rather than from a branch in the API.""" + client = await aiohttp_client(api()) + response = await client.post( + f"/api/guilds/{GUILD}/export-targets", + json={"format": "markdown", "name": "archive", "target": "../secrets"}, + cookies=cookies, + ) + assert response.status == 400 + + +@pytest.mark.parametrize( + "body", + [ + {"format": "outline", "name": "", "target": "col-1"}, + {"format": "outline", "name": "wiki", "target": ""}, + {"format": "outline", "name": "wiki", "target": "col-1", "config": []}, + {"format": "outline", "name": "wiki", "target": "col-1", "enabled": "yes"}, + {"name": "wiki", "target": "col-1"}, + ], +) +async def test_a_malformed_destination_is_refused( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str], body: dict[str, object] +) -> None: + client = await aiohttp_client(api()) + response = await client.post(f"/api/guilds/{GUILD}/export-targets", json=body, cookies=cookies) + assert response.status == 400 + + +async def test_a_body_that_is_not_an_object_is_refused( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + client = await aiohttp_client(api()) + response = await client.post( + f"/api/guilds/{GUILD}/export-targets", json=["nope"], cookies=cookies + ) + assert response.status == 400 + + +async def test_a_destinations_own_configuration_survives_a_round_trip( + aiohttp_client: AiohttpClientFactory, cookies: dict[str, str] +) -> None: + """`config` is whatever the format needs -- a base URL, a space key -- + and it is a mapping rather than five columns four formats would leave + null.""" + client = await aiohttp_client(api()) + response = await client.post( + f"/api/guilds/{GUILD}/export-targets", + json=outline_body(config={"base_url": "https://wiki.example", "space": "TEAM"}), + cookies=cookies, + ) + assert (await response.json())["config"] == { + "base_url": "https://wiki.example", + "space": "TEAM", + } diff --git a/tests/infrastructure/test_document_sinks.py b/tests/infrastructure/test_document_sinks.py new file mode 100644 index 0000000..0410db8 --- /dev/null +++ b/tests/infrastructure/test_document_sinks.py @@ -0,0 +1,191 @@ +"""The object-store sink, and the resolver that decides which sink runs. + +The load-bearing assertion in this file is the one about the URL. A +presigned S3 URL would satisfy `CreatedDocument` and would be wrong: it +works for anybody it is forwarded to, it keeps working after a participation +ends, and nothing can revoke it. The URL a protocol's link carries has to +point back at the console, where the rule is checked on every request. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from datetime import UTC, datetime + +import boto3 # type: ignore[import-untyped] +import pytest +from moto import mock_aws + +from sturnus.application.documents import CreatedDocument, DocumentSink +from sturnus.application.export_formats import HTML, MARKDOWN, OUTLINE, format_named +from sturnus.application.exporting import Destination +from sturnus.infrastructure.documents.sinks import DocumentSinks, ObjectStoreSink +from sturnus.infrastructure.objectstore import S3DocumentStore + +BUCKET = "sturnus-audio" +T0 = datetime(2026, 8, 19, 20, 0, 0, tzinfo=UTC) + + +@pytest.fixture +def store() -> Iterator[S3DocumentStore]: + with mock_aws(): + boto3.client("s3", region_name="us-east-1").create_bucket(Bucket=BUCKET) + yield S3DocumentStore(endpoint=None, bucket=BUCKET, access_key="ak", secret_key="sk") + + +def destination( + format: str = MARKDOWN, target_id: int | None = 7, where: str = "protocols" +) -> Destination: + entry = format_named(format) + assert entry is not None + return Destination( + session_id=42, target_id=target_id, format=entry, target=where, provider=format + ) + + +def sink(store: S3DocumentStore, format: str = MARKDOWN) -> ObjectStoreSink: + entry = format_named(format) + assert entry is not None + return ObjectStoreSink( + store, + console_origin="https://sturnus.example", + session_id=42, + target_id=7, + media_type=entry.media_type, + file_extension=entry.file_extension, + ) + + +# --------------------------------------------------------------------------- +# What the sink writes, and what it hands back +# --------------------------------------------------------------------------- + + +async def test_the_rendered_protocol_reaches_the_object_store(store: S3DocumentStore) -> None: + await sink(store).create("A meeting", "# Minutes\n", "protocols") + stored = boto3.client("s3", region_name="us-east-1").get_object( + Bucket=BUCKET, Key="protocols/42/7.md" + ) + assert stored["Body"].read() == b"# Minutes\n" + + +async def test_the_object_is_stored_under_its_own_media_type(store: S3DocumentStore) -> None: + """A browser handed `text/html` renders the protocol; handed + `binary/octet-stream` it offers to download it.""" + await sink(store, HTML).create("A meeting", "<p>hi</p>", "protocols") + stored = boto3.client("s3", region_name="us-east-1").get_object( + Bucket=BUCKET, Key="protocols/42/7.html" + ) + assert stored["ContentType"] == "text/html; charset=utf-8" + + +async def test_the_document_id_is_the_object_key(store: S3DocumentStore) -> None: + """`session_document.document_id` is what the console route reads back + to find the bytes, so it has to be a real identifier in the store that + holds them rather than a second, invented one.""" + created = await sink(store).create("A meeting", "body", "protocols") + assert created.id == "protocols/42/7.md" + + +async def test_the_url_points_at_the_console_and_not_at_the_object_store( + store: S3DocumentStore, +) -> None: + """The whole argument of §3.2. A presigned S3 URL outlives the access + rules that issued it and cannot be revoked; this one is answered by a + route that re-checks participation on every request.""" + created = await sink(store).create("A meeting", "body", "protocols") + assert created.url == "https://sturnus.example/api/sessions/42/documents/7" + assert "X-Amz-Signature" not in created.url + assert BUCKET not in created.url + + +async def test_two_destinations_of_one_format_get_two_objects(store: S3DocumentStore) -> None: + """A guild publishing Markdown to two prefixes wants two artefacts. + One key for both would have the second overwrite the first.""" + entry = format_named(MARKDOWN) + assert entry is not None + for target_id, prefix in ((7, "team"), (9, "archive")): + await ObjectStoreSink( + store, + console_origin="https://sturnus.example", + session_id=42, + target_id=target_id, + media_type=entry.media_type, + file_extension=entry.file_extension, + ).create("A meeting", f"body {target_id}", prefix) + assert await store.get("team/42/7.md") == b"body 7" + assert await store.get("archive/42/9.md") == b"body 9" + + +async def test_a_re_export_replaces_the_artefact_at_the_same_address( + store: S3DocumentStore, +) -> None: + """`SessionDocumentStore.record` upserts on `(session_id, target_id)`, + so a second artefact at a second key would leave a row pointing at one + of two objects with nothing saying which is current.""" + await sink(store).create("A meeting", "first", "protocols") + created = await sink(store).create("A meeting", "second", "protocols") + assert await store.get(created.id) == b"second" + + +async def test_reading_an_artefact_that_is_not_there_is_a_key_error( + store: S3DocumentStore, +) -> None: + with pytest.raises(KeyError): + await store.get("protocols/1/1.md") + + +# --------------------------------------------------------------------------- +# The resolver +# --------------------------------------------------------------------------- + + +class FakeOutline(DocumentSink): + async def create(self, title: str, body: str, target: str) -> CreatedDocument: + raise AssertionError(f"not called: {title} {body} {target}") + + +def test_an_outline_destination_resolves_to_the_outline_sink() -> None: + outline = FakeOutline() + sinks = DocumentSinks(outline=outline) + assert sinks.sink_for(destination(OUTLINE, 3, "col-1")) is outline + + +def test_an_object_store_destination_resolves_to_an_object_store_sink( + store: S3DocumentStore, +) -> None: + sinks = DocumentSinks(objects=store, console_origin="https://sturnus.example") + assert isinstance(sinks.sink_for(destination(MARKDOWN)), ObjectStoreSink) + + +def test_the_resolver_branches_on_the_family_and_not_on_the_format( + store: S3DocumentStore, +) -> None: + """`markdown` and `html` are two formats and one family, which is what + makes `pdf` an entry in the registry rather than a change here.""" + sinks = DocumentSinks(objects=store, console_origin="https://sturnus.example") + assert isinstance(sinks.sink_for(destination(MARKDOWN)), ObjectStoreSink) + assert isinstance(sinks.sink_for(destination(HTML)), ObjectStoreSink) + + +def test_a_destination_this_process_cannot_serve_resolves_to_nothing() -> None: + """A deployment with no object store configured. `None` rather than a + raise: one unbuildable destination must not take a guild's working + Outline document down with it.""" + assert DocumentSinks(outline=FakeOutline()).sink_for(destination(MARKDOWN)) is None + + +def test_a_deployment_with_no_outline_sink_serves_no_outline_destination() -> None: + assert DocumentSinks().sink_for(destination(OUTLINE, 3, "col-1")) is None + + +async def test_the_console_origins_trailing_slash_does_not_double_up( + store: S3DocumentStore, +) -> None: + """`STURNUS_CONSOLE_ORIGIN` is a Helm value somebody types, and a + doubled slash in a link posted to Discord is a broken link.""" + sinks = DocumentSinks(objects=store, console_origin="https://sturnus.example/") + resolved = sinks.sink_for(destination(MARKDOWN)) + assert resolved is not None + created = await resolved.create("A meeting", "body", "protocols") + assert created.url == "https://sturnus.example/api/sessions/42/documents/7" diff --git a/tests/infrastructure/test_repositories.py b/tests/infrastructure/test_repositories.py index d39769f..1006788 100644 --- a/tests/infrastructure/test_repositories.py +++ b/tests/infrastructure/test_repositories.py @@ -8,6 +8,8 @@ from sturnus.application.transcription import TranscribedSegment, TranscriptionResult from sturnus.domain.consent import ConsentScope from sturnus.entrypoints.worker import _WorkerSessionStore +from sturnus.infrastructure.crypto import KeyWrapper +from sturnus.infrastructure.db.export_targets import ExportTargetStore from sturnus.infrastructure.db.models import ( AccountLink, Base, @@ -22,6 +24,7 @@ JobRepository, SessionRepository, ) +from sturnus.infrastructure.db.session_documents import SessionDocumentStore T0 = datetime(2026, 8, 19, 20, 0, 0, tzinfo=UTC) GUILD, CHANNEL, ANNA, BEN = 1, 2, 100, 200 @@ -806,6 +809,150 @@ async def test_closed_undocumented_sessions_excludes_an_already_documented_sessi assert await sessions.closed_undocumented_sessions() == [] +# --------------------------------------------------------------------------- +# The second candidate set: a session that published somewhere but not everywhere +# --------------------------------------------------------------------------- + + +async def _configure_target( + factory: async_sessionmaker[AsyncSession], + name: str = "wiki", + enabled: bool = True, +) -> int: + """One `guild_export_target` row, through the real store.""" + store = ExportTargetStore(factory, KeyWrapper(b"m" * 32, "master-1")) + return await store.save( + GUILD, format="outline", name=name, target="col-1", config={}, enabled=enabled, now=T0 + ) + + +async def _publish( + factory: async_sessionmaker[AsyncSession], session_id: int, target_id: int +) -> None: + await SessionDocumentStore(factory).record( + session_id, + target_id=target_id, + provider="outline", + document_id="doc-1", + url="https://outline.example/doc/1", + now=T0, + ) + + +async def _finished_session(factory: async_sessionmaker[AsyncSession]) -> int: + """A closed session whose one job is done -- ready to be published.""" + sessions = SessionRepository(factory) + jobs = JobRepository(factory) + session_id = await sessions.open_session(GUILD, CHANNEL, "meeting-raum", T0) + job_id = await _enqueue_job(sessions, jobs, session_id, ANNA) + await sessions.close_session(session_id, T0 + timedelta(hours=1), "empty") + await JobQueue(factory).complete(job_id, "hello") + return session_id + + +async def test_a_documented_session_with_a_target_it_never_reached_is_a_candidate( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The case a second destination introduced. Outline succeeded, so the + session is `documented` and invisible to `closed_undocumented_sessions` + -- without this query the failed second export is never retried and the + guild is missing an artefact with nothing anywhere saying so. + """ + sessions = SessionRepository(factory) + session_id = await _finished_session(factory) + await _configure_target(factory) + await _mark_documented(factory, session_id) + + assert await sessions.sessions_with_unpublished_targets() == [session_id] + + +async def test_a_session_that_reached_every_target_is_not_a_candidate( + factory: async_sessionmaker[AsyncSession], +) -> None: + sessions = SessionRepository(factory) + session_id = await _finished_session(factory) + target_id = await _configure_target(factory) + await _publish(factory, session_id, target_id) + await _mark_documented(factory, session_id) + + assert await sessions.sessions_with_unpublished_targets() == [] + + +async def test_a_session_that_reached_one_of_two_targets_is_a_candidate( + factory: async_sessionmaker[AsyncSession], +) -> None: + sessions = SessionRepository(factory) + session_id = await _finished_session(factory) + first = await _configure_target(factory, "wiki") + await _configure_target(factory, "archive") + await _publish(factory, session_id, first) + await _mark_documented(factory, session_id) + + assert await sessions.sessions_with_unpublished_targets() == [session_id] + + +async def test_a_disabled_target_is_not_something_a_session_still_owes( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Switching a destination off is an administrator saying "stop + publishing here". A sweep that kept bringing the session back for it + would be re-deciding that every five minutes. + """ + sessions = SessionRepository(factory) + session_id = await _finished_session(factory) + await _configure_target(factory, "archive", enabled=False) + await _mark_documented(factory, session_id) + + assert await sessions.sessions_with_unpublished_targets() == [] + + +async def test_a_session_still_being_transcribed_is_not_a_candidate( + factory: async_sessionmaker[AsyncSession], +) -> None: + """There is no transcript to publish yet, and offering it here would + have the sweep assemble a partial one every five minutes for as long + as the meeting's last job takes. + """ + sessions = SessionRepository(factory) + jobs = JobRepository(factory) + session_id = await sessions.open_session(GUILD, CHANNEL, "meeting-raum", T0) + await _enqueue_job(sessions, jobs, session_id, ANNA) + await sessions.close_session(session_id, T0 + timedelta(hours=1), "empty") + await _configure_target(factory) + + assert await sessions.sessions_with_unpublished_targets() == [] + + +async def test_a_guild_with_no_configured_targets_owes_nothing( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Every guild running today. Their publishing is recorded on the + session row, and `closed_undocumented_sessions` is what retries it. + """ + sessions = SessionRepository(factory) + session_id = await _finished_session(factory) + await _mark_documented(factory, session_id) + + assert await sessions.sessions_with_unpublished_targets() == [] + + +async def test_another_guilds_targets_are_not_something_this_session_owes( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The join is on the session's own guild. Without that, one guild + configuring a destination would put every other guild's sessions back + in front of the sweep for ever. + """ + sessions = SessionRepository(factory) + session_id = await _finished_session(factory) + await ExportTargetStore(factory, KeyWrapper(b"m" * 32, "master-1")).save( + GUILD + 1, format="outline", name="theirs", target="col-9", config={}, now=T0 + ) + await _mark_documented(factory, session_id) + + assert await sessions.sessions_with_unpublished_targets() == [] + + async def test_candidates_for_retention_returns_undeleted_jobs( factory: async_sessionmaker[AsyncSession], ) -> None: diff --git a/tests/infrastructure/test_traced_ports.py b/tests/infrastructure/test_traced_ports.py index 1bff2d6..f9406b0 100644 --- a/tests/infrastructure/test_traced_ports.py +++ b/tests/infrastructure/test_traced_ports.py @@ -45,6 +45,7 @@ FakeQueue, FakeSessions, FakeStore, + exports, job, ) @@ -77,7 +78,7 @@ async def _run_one(tmp_path: Path, *, is_last: bool = True) -> tuple[FakeQueue, engine=TracedTranscriptionEngine(FakeEngine(CANARY)), store=TracedAudioDownloader(FakeStore()), crypto=TracedDecryptor(FakeCrypto()), - documents=TracedDocumentSink(documents), + exports=exports(TracedDocumentSink(documents)), sessions=sessions, jobs=FakeJobs(), links=FakeLinks(), @@ -175,7 +176,7 @@ async def test_a_failing_stage_marks_its_span_without_a_message( engine=TracedTranscriptionEngine(FakeEngine(CANARY, fail=True)), store=TracedAudioDownloader(FakeStore()), crypto=TracedDecryptor(FakeCrypto()), - documents=TracedDocumentSink(FakeDocuments()), + exports=exports(TracedDocumentSink(FakeDocuments())), sessions=FakeSessions(), jobs=FakeJobs(), links=FakeLinks(), @@ -212,7 +213,7 @@ async def test_the_root_span_of_a_failed_job_does_not_say_it_was_done( engine=TracedTranscriptionEngine(FakeEngine(CANARY, fail=True)), store=TracedAudioDownloader(FakeStore()), crypto=TracedDecryptor(FakeCrypto()), - documents=TracedDocumentSink(FakeDocuments()), + exports=exports(TracedDocumentSink(FakeDocuments())), sessions=FakeSessions(), jobs=FakeJobs(), links=FakeLinks(), diff --git a/tests/observability/test_no_payload_leaks.py b/tests/observability/test_no_payload_leaks.py index 87b05fb..6a020ee 100644 --- a/tests/observability/test_no_payload_leaks.py +++ b/tests/observability/test_no_payload_leaks.py @@ -42,6 +42,7 @@ FakeQueue, FakeSessions, FakeStore, + exports, job, ) @@ -101,7 +102,7 @@ async def test_the_worker_pipeline_logs_no_payload(tmp_path: Path, captured: io. engine=FakeEngine(TRANSCRIPT_CANARY), store=FakeStore(), crypto=FakeCrypto(), - documents=FakeDocuments(), + exports=exports(FakeDocuments()), sessions=sessions, jobs=FakeJobs(), links=FakeLinks(), @@ -141,7 +142,7 @@ async def transcribe( engine=ExplodingEngine(), # type: ignore[arg-type] store=FakeStore(), crypto=FakeCrypto(), - documents=FakeDocuments(), + exports=exports(FakeDocuments()), sessions=FakeSessions(), jobs=FakeJobs(), links=FakeLinks(),