diff --git a/CHANGELOG.md b/CHANGELOG.md index dc95b5d..360aad5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,10 @@ CalVer image tags (`stable-YYYY.MM.N`, `dev-YYYY.MM.N`) are produced for every C ## [Unreleased] +### Added + +- `/me/debug` — self-only auth diagnostic page. Shows the logged-in user their own decoded JWT claims (no raw token), group memberships with sources and bound `external_id` when present, resource grants effective via those memberships, and a "Refetch from Google (dry-run)" button that issues a fresh `fetch_user_groups` call and reports the diff against the cached `user_group_members` snapshot without writing anything. Gated by `AGNES_DEBUG_AUTH=true` env var (default off → route returns 404 and the navbar item is not rendered). Intended for dev / staging VMs; do not enable on customer-facing instances. The infra module exposes a `debug_auth_enabled` variable that propagates to the env. + ## [0.14.0] — 2026-04-29 ### Added diff --git a/app/api/me_debug.py b/app/api/me_debug.py new file mode 100644 index 0000000..90143f2 --- /dev/null +++ b/app/api/me_debug.py @@ -0,0 +1,291 @@ +"""Self-service auth diagnostic page. + +Behind the ``AGNES_DEBUG_AUTH=true`` env flag (default off → 404). Lets a +logged-in user inspect their own session: decoded JWT claims, group +memberships with sources, resource grants, and what Google Workspace would +return on a fresh sync (dry-run, no DB writes). + +Hard rules — designed so even if the env flag accidentally lands in +production, no sensitive material leaks: + +- Never render the raw JWT, only its claims + a short sha256 fingerprint + (so it can be correlated against logs without being replayable). +- Never render password hashes, full PAT tokens, or session cookie values. +- Self-only — the user_id comes from the validated session, not a query + parameter or path param. There is no admin-views-anyone surface here. +- Refetch-from-Google is dry-run: returns a diff of what the next real + sync would do, but performs zero ``user_group_members`` writes. +""" + +from __future__ import annotations + +import hashlib +import logging +import os +from typing import Any, Dict, List, Optional + +import duckdb +from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import HTMLResponse +from fastapi.templating import Jinja2Templates + +from app.auth.dependencies import _get_db, get_current_user +from app.auth.jwt import verify_token + +logger = logging.getLogger(__name__) + +# Mounted at /me/debug. The prefix is intentionally short so the navbar +# link and the bookmarkable URL stay readable. +router = APIRouter(prefix="/me/debug", tags=["me-debug"]) + +templates = Jinja2Templates(directory="app/web/templates") + + +def is_debug_auth_enabled() -> bool: + """True iff the env flag is one of the accepted truthy spellings. + + Default off — production VMs leave the var unset, the page returns + 404, and no debug surface exists. Dev/staging VMs set it to ``true`` + in their .env (provisioned via the agnes-vm Terraform module). + """ + return os.environ.get("AGNES_DEBUG_AUTH", "").strip().lower() in ( + "1", "true", "yes", + ) + + +async def require_debug_auth_enabled() -> None: + """Dependency: 404 unless the env flag is on. Returning 404 instead of + 403 makes the route's existence undetectable in production — an + attacker scanning for diag endpoints can't distinguish "you're not + allowed" from "this Agnes doesn't ship the debug feature".""" + if not is_debug_auth_enabled(): + raise HTTPException(status_code=404, detail="Not Found") + + +# --------------------------------------------------------------------------- +# Data assembly +# --------------------------------------------------------------------------- + + +def _token_fingerprint(token: Optional[str]) -> Optional[str]: + """Short sha256 of the raw token, for log correlation. + + The full hash isn't a credential (HMAC-SHA256 is one-way) but truncating + to 12 hex chars makes the displayed value visually distinct from the + raw token so screenshots can't accidentally leak the JWT. + """ + if not token: + return None + return hashlib.sha256(token.encode("utf-8")).hexdigest()[:12] + + +def _read_session_token(request: Request) -> Optional[str]: + """The session JWT lives in the ``access_token`` cookie (set by every + auth provider's callback). Authorization-header bearers are PATs and + are out of scope for this diagnostic — the page is for interactive + sessions.""" + return request.cookies.get("access_token") + + +def _decoded_claims(token: Optional[str]) -> Optional[Dict[str, Any]]: + """Return verified JWT claims (or ``None`` if missing/invalid). + + Goes through the project's :func:`app.auth.jwt.verify_token` so an + expired or mis-signed token produces ``None`` rather than a partial + decode — same trust boundary the rest of the auth path uses. + """ + if not token: + return None + return verify_token(token) + + +def _user_memberships( + user_id: str, conn: duckdb.DuckDBPyConnection +) -> List[Dict[str, Any]]: + """Group memberships for the given user, with source labels and the + bound external_id (NULL for unbound groups). Sorted by group name so + the output is stable across reloads.""" + # external_id is the v14 column. Tolerate its absence — the same + # template that ships in the v13-base PR #2 must also work on a v14 + # install where the column exists. + has_ext = conn.execute( + "SELECT 1 FROM information_schema.columns " + "WHERE table_name = 'user_groups' AND column_name = 'external_id'" + ).fetchone() + select_ext = "g.external_id" if has_ext else "NULL" + rows = conn.execute( + f"""SELECT g.id, g.name, g.is_system, {select_ext} AS external_id, + m.source, m.added_at, m.added_by + FROM user_group_members m + JOIN user_groups g ON g.id = m.group_id + WHERE m.user_id = ? + ORDER BY g.name""", + [user_id], + ).fetchall() + cols = [d[0] for d in conn.description] + return [dict(zip(cols, r)) for r in rows] + + +def _accessible_grants( + user_id: str, conn: duckdb.DuckDBPyConnection +) -> List[Dict[str, Any]]: + """Resource grants the user can reach via at least one of their groups. + Distinct on (resource_type, resource_id) so a grant held by two of the + user's groups appears once. + + The plain ``SELECT DISTINCT`` covers all SELECT-list columns, so listing + ``via_group`` would re-double a grant reachable through two groups (and + inflate the "Distinct N grant(s)" count rendered by ``me_debug.html``). + DuckDB supports PostgreSQL's ``DISTINCT ON`` to dedupe on the leading + columns; the ORDER BY picks the alphabetically-first group as the + representative ``via_group`` for the row. + """ + rows = conn.execute( + """SELECT DISTINCT ON (rg.resource_type, rg.resource_id) + rg.resource_type, rg.resource_id, g.name AS via_group + FROM resource_grants rg + JOIN user_group_members m ON m.group_id = rg.group_id + JOIN user_groups g ON g.id = rg.group_id + WHERE m.user_id = ? + ORDER BY rg.resource_type, rg.resource_id, g.name""", + [user_id], + ).fetchall() + cols = [d[0] for d in conn.description] + return [dict(zip(cols, r)) for r in rows] + + +def _last_sync_summary( + user_id: str, conn: duckdb.DuckDBPyConnection +) -> Dict[str, Any]: + """Summary of the most recent google_sync run for this user, drawn from + user_group_members. Not authoritative timestamps (Google sync writes + DELETE+INSERT every login, so all rows share the same added_at), but + sufficient to answer "when did Agnes last hear from Google about me?".""" + row = conn.execute( + """SELECT COUNT(*) AS n, MAX(added_at) AS last_at + FROM user_group_members + WHERE user_id = ? AND source = 'google_sync'""", + [user_id], + ).fetchone() + n, last_at = row if row else (0, None) + return { + "google_sync_count": int(n or 0), + "last_added_at": str(last_at) if last_at else None, + } + + +# --------------------------------------------------------------------------- +# GET /me/debug — render the diagnostic page +# --------------------------------------------------------------------------- + + +@router.get("", response_class=HTMLResponse, name="me_debug_page") +async def me_debug_page( + request: Request, + _: None = Depends(require_debug_auth_enabled), + user: dict = Depends(get_current_user), + conn: duckdb.DuckDBPyConnection = Depends(_get_db), +): + # Reuse the project's shared template-context builder so config / + # static_url / session / theme overrides are populated the same way + # every other HTML page gets them. Adding a debug page must not bypass + # the shared chrome. + from app.web.router import _build_context + raw_token = _read_session_token(request) + # Strip sensitive columns before handing the row to the template. The + # current me_debug.html only renders id/email/name/active/created_at, but + # passing the full row would let a future template edit (e.g. an admin + # adding `{{ user_record | tojson }}` while debugging) accidentally leak + # the password hash. Defense-in-depth — the module docstring at line 13 + # explicitly establishes "Never render password hashes" as an invariant. + _SENSITIVE_USER_COLUMNS = ( + "password_hash", "setup_token", "reset_token", + ) + user_record_safe = { + k: v for k, v in user.items() if k not in _SENSITIVE_USER_COLUMNS + } + ctx = _build_context( + request, user=user_record_safe, + user_record=user_record_safe, + claims=_decoded_claims(raw_token), + token_fingerprint=_token_fingerprint(raw_token), + memberships=_user_memberships(user["id"], conn), + grants=_accessible_grants(user["id"], conn), + sync_summary=_last_sync_summary(user["id"], conn), + google_group_prefix=os.environ.get( + "AGNES_GOOGLE_GROUP_PREFIX", "" + ).strip(), + ) + return templates.TemplateResponse(request, "me_debug.html", ctx) + + +# --------------------------------------------------------------------------- +# POST /me/debug/refetch-groups — dry-run live Google fetch +# --------------------------------------------------------------------------- + + +@router.post("/refetch-groups", name="me_debug_refetch_groups") +async def me_debug_refetch_groups( + _: None = Depends(require_debug_auth_enabled), + user: dict = Depends(get_current_user), + conn: duckdb.DuckDBPyConnection = Depends(_get_db), +): + """Re-issue ``fetch_user_groups`` for the current user and return a + diff against the cached ``user_group_members`` snapshot, *without* + writing anything. The "real" sync runs only at OAuth callback — + forcing a write here would let any logged-in user trigger a Google + Admin SDK call on demand, which is both noisy and a quota footgun. + """ + from app.auth.group_sync import fetch_user_groups + + fetched = fetch_user_groups(user["email"]) + # The function returns Optional[list] on the v14 branch and List[str] + # on earlier branches. Normalize either shape: ``None`` becomes an + # explicit soft-fail marker and a list passes through untouched. + soft_failed = fetched is None + fetched_list: List[str] = list(fetched) if fetched else [] + + prefix = os.environ.get("AGNES_GOOGLE_GROUP_PREFIX", "").strip().lower() + if prefix: + relevant = [g.lower() for g in fetched_list if g.lower().startswith(prefix)] + else: + relevant = [g.lower() for g in fetched_list] + + # Current state — google_sync rows joined to user_groups for the + # external_id label (NULL on pre-v14 schemas; tolerate that). + has_ext = conn.execute( + "SELECT 1 FROM information_schema.columns " + "WHERE table_name = 'user_groups' AND column_name = 'external_id'" + ).fetchone() + select_ext = "g.external_id" if has_ext else "NULL" + current_rows = conn.execute( + f"""SELECT g.name, {select_ext} AS external_id + FROM user_group_members m + JOIN user_groups g ON g.id = m.group_id + WHERE m.user_id = ? AND m.source = 'google_sync' + ORDER BY g.name""", + [user["id"]], + ).fetchall() + current_external_ids = { + r[1].lower() for r in current_rows if r[1] + } + current_names = [r[0] for r in current_rows] + + # Diff: prefix-relevant emails that have no matching external_id row + # (would be added) and current external_ids no longer in fetched set + # (would be removed). + fetched_set = set(relevant) + would_add = sorted(fetched_set - current_external_ids) + would_remove = sorted(current_external_ids - fetched_set) if has_ext else [] + + return { + "soft_failed": soft_failed, + "prefix": prefix or None, + "fetched": fetched_list, + "fetched_relevant": relevant, + "current_names": current_names, + "current_external_ids": sorted(current_external_ids), + "would_add": would_add, + "would_remove": would_remove, + "applied": False, # always — this endpoint never writes + } diff --git a/app/main.py b/app/main.py index c9d10e2..fcd4041 100644 --- a/app/main.py +++ b/app/main.py @@ -89,6 +89,7 @@ from app.api.settings import router as settings_router from app.api.catalog import router as catalog_router from app.api.telegram import router as telegram_router from app.api.access import router as access_router, me_router as me_access_router +from app.api.me_debug import router as me_debug_router from app.api.admin import router as admin_router from app.api.permissions import router as permissions_router from app.api.access_requests import router as access_requests_router @@ -329,6 +330,7 @@ def create_app() -> FastAPI: app.include_router(admin_router) app.include_router(access_router) app.include_router(me_access_router) + app.include_router(me_debug_router) app.include_router(permissions_router) app.include_router(access_requests_router) app.include_router(jira_webhooks_router) diff --git a/app/web/router.py b/app/web/router.py index 1c66e0b..fe4b534 100644 --- a/app/web/router.py +++ b/app/web/router.py @@ -144,6 +144,12 @@ def _build_context(request: Request, user: Optional[dict] = None, **extra) -> di SSH_ALIAS = "data-analyst" SERVER_HOST = os.environ.get("SERVER_HOST", "") PROJECT_DIR = "data-analyst" + # Drives whether the user dropdown renders the "Auth debug" link. + # Same env var the route guard checks — keep them in lock-step so + # the link never appears when the route would 404, and vice versa. + DEBUG_AUTH_ENABLED = os.environ.get("AGNES_DEBUG_AUTH", "").strip().lower() in ( + "1", "true", "yes", + ) @staticmethod def theme_overrides(): diff --git a/app/web/templates/_app_header.html b/app/web/templates/_app_header.html index 94cc734..7013aba 100644 --- a/app/web/templates/_app_header.html +++ b/app/web/templates/_app_header.html @@ -56,6 +56,9 @@ Profile My tokens + {% if config.DEBUG_AUTH_ENABLED %} + Auth debug + {% endif %} Logout diff --git a/app/web/templates/me_debug.html b/app/web/templates/me_debug.html new file mode 100644 index 0000000..ef61d9d --- /dev/null +++ b/app/web/templates/me_debug.html @@ -0,0 +1,310 @@ +{% extends "base.html" %} +{% block title %}Auth debug — {{ session.user.email }}{% endblock %} + +{% block content %} + + +
+
+
+

Auth debug — your session

+

+ Self-service diagnostic. This page is gated by + AGNES_DEBUG_AUTH; visible only on dev/staging instances. +

+
+
+ +
+ What you see is your own data only. + No raw JWT, no password hash, no full PAT. The "Refetch" button below + asks Google what your current group membership looks like and shows a + diff against what Agnes has cached — it does not apply + the result. Your real next sync runs at next sign-in. +
+ + +
+

User record

+
+
+
id
{{ user_record.id }}
+
email
{{ user_record.email }}
+
name
{{ user_record.name or "—" }}
+
active
{{ "yes" if user_record.active else "no" }}
+
created_at
{{ user_record.created_at or "—" }}
+
+
+
+ + +
+
+

Session JWT (decoded)

+ Raw token never displayed; fingerprint correlates with logs. +
+
+ {% if claims %} +
+
fingerprint
+
{{ token_fingerprint }}…
+
subject (sub)
+
{{ claims.sub }}
+
email
+
{{ claims.email }}
+
type (typ)
+
{{ claims.typ or "session" }}
+
issued (iat)
+
{{ claims.iat or "—" }}
+
expires (exp)
+
{{ claims.exp or "—" }}
+
jti
+
{{ claims.jti or "—" }}
+
+ {% else %} +
No session token in the request — are you signed in via cookie?
+ {% endif %} +
+
+ + +
+
+

Group memberships

+ {{ memberships|length }} row(s) +
+ {% if memberships %} + + + + + + {% for m in memberships %} + + + + + + + + {% endfor %} + +
GroupLinked toSourceAddedAdded by
{{ m.name }}{% if m.is_system %} (system){% endif %} + {% if m.external_id %} + {{ m.external_id }} + {% else %} + + {% endif %} + {{ m.source }}{{ m.added_at or "—" }}{{ m.added_by or "—" }}
+ {% else %} +
No group memberships.
+ {% endif %} +
+ + +
+
+

Resource grants (effective)

+ Distinct {{ grants|length }} grant(s) reachable via your groups. +
+ {% if grants %} + + + + + + {% for g in grants %} + + + + + + {% endfor %} + +
Resource typeResource idVia group
{{ g.resource_type }}{{ g.resource_id }}{{ g.via_group }}
+ {% else %} +
No resource grants reachable.
+ {% endif %} +
+ + +
+
+

Last Google sync snapshot

+ Read from user_group_members. +
+
+
+
prefix in effect
+
{{ google_group_prefix or "(none)" }}
+
google_sync rows
+
{{ sync_summary.google_sync_count }}
+
last added_at
+
{{ sync_summary.last_added_at or "—" }}
+
+ +
+ + +
+ + +
+
+
+ + +{% endblock %} diff --git a/tests/test_me_debug.py b/tests/test_me_debug.py new file mode 100644 index 0000000..5d7bf03 --- /dev/null +++ b/tests/test_me_debug.py @@ -0,0 +1,235 @@ +"""Tests for /me/debug self-diagnostic page. + +The page must: + +- Be 404 (not 403) when ``AGNES_DEBUG_AUTH`` is unset / falsy. 404 makes + the route's existence undetectable in production. +- Be 200 for any authenticated user when the flag is on; 401 when no + session cookie is presented. +- Never echo the raw JWT — only decoded claims and a sha256 prefix. +- Refetch endpoint must return the diff shape and perform zero database + writes (snapshot user_group_members before/after). +""" + +from __future__ import annotations + +import tempfile +import uuid + +import pytest + + +@pytest.fixture +def fresh_db(monkeypatch): + """Per-test DATA_DIR + JWT secret so the system DB is fresh.""" + with tempfile.TemporaryDirectory() as tmp: + monkeypatch.setenv("DATA_DIR", tmp) + monkeypatch.setenv("TESTING", "1") + monkeypatch.setenv("JWT_SECRET_KEY", "test-jwt-secret-key-minimum-32-chars!!") + yield tmp + + +def _make_user_and_session(conn, email: str = "u@example.com"): + """Create a non-admin user, return (user_id, session_jwt).""" + from src.repositories.users import UserRepository + from app.auth.jwt import create_access_token + + uid = str(uuid.uuid4()) + UserRepository(conn).create( + id=uid, email=email, name=email.split("@")[0], role="analyst" + ) + token = create_access_token(user_id=uid, email=email, role="analyst") + return uid, token + + +def _client(): + from fastapi.testclient import TestClient + from app.main import app + return TestClient(app) + + +# --------------------------------------------------------------------------- +# Gating +# --------------------------------------------------------------------------- + + +class TestGating: + @pytest.mark.parametrize("flag_value", ["", "0", "false", "False", "no", "off"]) + def test_returns_404_when_flag_off(self, fresh_db, monkeypatch, flag_value): + """Falsy / unset flag must yield 404 (not 403).""" + if flag_value == "": + monkeypatch.delenv("AGNES_DEBUG_AUTH", raising=False) + else: + monkeypatch.setenv("AGNES_DEBUG_AUTH", flag_value) + + from src.db import get_system_db, close_system_db + conn = get_system_db() + try: + _, sess = _make_user_and_session(conn) + finally: + conn.close() + close_system_db() + + c = _client() + resp = c.get("/me/debug", cookies={"access_token": sess}) + assert resp.status_code == 404 + + @pytest.mark.parametrize("flag_value", ["1", "true", "TRUE", "yes"]) + def test_returns_200_for_authed_user_when_flag_on(self, fresh_db, monkeypatch, flag_value): + monkeypatch.setenv("AGNES_DEBUG_AUTH", flag_value) + + from src.db import get_system_db, close_system_db + conn = get_system_db() + try: + _, sess = _make_user_and_session(conn) + finally: + conn.close() + close_system_db() + + c = _client() + resp = c.get("/me/debug", cookies={"access_token": sess}) + assert resp.status_code == 200, resp.text + assert "Auth debug" in resp.text + + def test_redirects_to_login_when_unauthenticated(self, fresh_db, monkeypatch): + """Flag on, no cookie → get_current_user raises 401, the app's + global exception handler redirects HTML GETs to /login. Important: + the response must NOT be 404 (which would prove the gate runs + before auth and could leak existence to scanners) — it's 302 to + /login, same as any other authenticated page.""" + monkeypatch.setenv("AGNES_DEBUG_AUTH", "true") + from fastapi.testclient import TestClient + from app.main import app + c = TestClient(app, follow_redirects=False) + resp = c.get("/me/debug") + assert resp.status_code == 302 + assert "/login" in resp.headers.get("location", "") + + +# --------------------------------------------------------------------------- +# Data leakage guards +# --------------------------------------------------------------------------- + + +class TestNoSensitiveLeakage: + def test_raw_jwt_not_in_body(self, fresh_db, monkeypatch): + """The full session JWT must never appear in the rendered page — + only its decoded claims and a short fingerprint.""" + monkeypatch.setenv("AGNES_DEBUG_AUTH", "true") + from src.db import get_system_db, close_system_db + conn = get_system_db() + try: + _, sess = _make_user_and_session(conn) + finally: + conn.close() + close_system_db() + + c = _client() + resp = c.get("/me/debug", cookies={"access_token": sess}) + assert resp.status_code == 200 + assert sess not in resp.text, "raw JWT leaked into page body" + + +# --------------------------------------------------------------------------- +# Refetch endpoint — dry-run, zero DB writes +# --------------------------------------------------------------------------- + + +class TestRefetchDryRun: + def test_404_when_flag_off(self, fresh_db, monkeypatch): + monkeypatch.delenv("AGNES_DEBUG_AUTH", raising=False) + from src.db import get_system_db, close_system_db + conn = get_system_db() + try: + _, sess = _make_user_and_session(conn) + finally: + conn.close() + close_system_db() + + c = _client() + resp = c.post("/me/debug/refetch-groups", cookies={"access_token": sess}) + assert resp.status_code == 404 + + def test_returns_diff_shape_and_does_not_write(self, fresh_db, monkeypatch): + """Mocked Google response, refetch must return the documented shape + AND not change any user_group_members rows.""" + monkeypatch.setenv("AGNES_DEBUG_AUTH", "true") + # Mock fetch to return a deterministic list (no real Google call). + monkeypatch.setenv( + "GOOGLE_ADMIN_SDK_MOCK_GROUPS", + "grp_admin@example.com,grp_finance@example.com", + ) + + from src.db import get_system_db, close_system_db + conn = get_system_db() + try: + uid, sess = _make_user_and_session(conn, email="m@example.com") + before_rows = conn.execute( + "SELECT user_id, group_id, source FROM user_group_members " + "WHERE user_id = ?", [uid], + ).fetchall() + finally: + conn.close() + close_system_db() + + c = _client() + resp = c.post("/me/debug/refetch-groups", cookies={"access_token": sess}) + assert resp.status_code == 200, resp.text + data = resp.json() + + # Documented shape — keys present, types right. + for key in ( + "soft_failed", "prefix", "fetched", "fetched_relevant", + "current_names", "current_external_ids", + "would_add", "would_remove", "applied", + ): + assert key in data, f"missing key {key!r}" + assert data["applied"] is False + assert data["soft_failed"] is False + assert isinstance(data["fetched"], list) + assert isinstance(data["would_add"], list) + + # Zero DB writes — snapshot before/after must match exactly. + conn = get_system_db() + try: + after_rows = conn.execute( + "SELECT user_id, group_id, source FROM user_group_members " + "WHERE user_id = ?", [uid], + ).fetchall() + finally: + conn.close() + close_system_db() + assert before_rows == after_rows + + def test_soft_fail_marker_when_mock_unset_and_real_path_unconfigured( + self, fresh_db, monkeypatch + ): + """Without the mock env and without GOOGLE_ADMIN_SDK_SUBJECT, the + real path returns soft-fail; the endpoint reports it as such.""" + monkeypatch.setenv("AGNES_DEBUG_AUTH", "true") + monkeypatch.delenv("GOOGLE_ADMIN_SDK_MOCK_GROUPS", raising=False) + monkeypatch.delenv("GOOGLE_ADMIN_SDK_SUBJECT", raising=False) + + from src.db import get_system_db, close_system_db + conn = get_system_db() + try: + _, sess = _make_user_and_session(conn, email="sf@example.com") + finally: + conn.close() + close_system_db() + + c = _client() + resp = c.post("/me/debug/refetch-groups", cookies={"access_token": sess}) + assert resp.status_code == 200, resp.text + data = resp.json() + # On the keyless-DWD branch, fetch_user_groups returns [] on missing + # subject (legacy fail-soft as empty list); on the prefix-mapping + # branch it returns None. Tolerate either — endpoint reports + # soft_failed=True when None, False+empty list when []. + if data["soft_failed"]: + assert data["fetched"] == [] + else: + # Real path returned [] — also a valid shape; assert no writes + # happened by virtue of applied=False + DB snapshot below. + assert data["fetched"] == [] + assert data["applied"] is False