Harden briefing fixture validation

Author: Autumn-Atlas

examples/source_linked_briefing/README.md

@@ -25,7 +25,9 @@ python examples/source_linked_briefing/briefing.py \
   --output-dir briefing-output
 ```
 
-The fixture contains fictional `example.com` articles. It contains no customer data, publisher article bodies, or credentials.
+The fixture contains fictional `example.com` articles. It contains no customer data, publisher article bodies, or credentials. Its `_fixture_generated_at` value keeps both output files identical across runs.
+
+For another saved Search API response, add `_fixture_generated_at` at the top level or pass `--generated-at` explicitly.
 
 ## Run a live search
 

examples/source_linked_briefing/briefing.py

@@ -14,7 +14,10 @@ def parse_args():
     parser.add_argument("--keywords", default="artificial intelligence")
     parser.add_argument("--language", default="en")
     parser.add_argument("--output-dir", type=Path, default=Path("briefing-output"))
-    parser.add_argument("--generated-at", help=argparse.SUPPRESS)
+    parser.add_argument(
+        "--generated-at",
+        help="Set the output timestamp for a custom fixture.",
+    )
     return parser.parse_args()
 
 
@@ -55,6 +58,30 @@ def validate_response(response):
         raise ValueError("Search API response news must be a list")
     if any(not isinstance(article, dict) for article in response["news"]):
         raise ValueError("Every news item must be an object")
+    for article in response["news"]:
+        for field in ("title", "description", "url", "published", "language"):
+            value = article.get(field)
+            if value is not None and not isinstance(value, str):
+                raise ValueError("news item {} must be a string".format(field))
+        category = article.get("category")
+        if category is not None and (
+            not isinstance(category, list)
+            or any(not isinstance(item, str) for item in category)
+        ):
+            raise ValueError("news item category must be a list of strings")
+
+
+def resolve_generated_at(args, response):
+    if args.generated_at:
+        return args.generated_at
+    if args.fixture:
+        fixture_time = response.get("_fixture_generated_at")
+        if not isinstance(fixture_time, str) or not fixture_time:
+            raise ValueError(
+                "Fixture must include _fixture_generated_at or use --generated-at"
+            )
+        return fixture_time
+    return datetime.now(timezone.utc).isoformat()
 
 
 def build_output(response, generated_at):
@@ -86,7 +113,8 @@ def main():
             if args.fixture
             else load_live_response(args.keywords, args.language)
         )
-        generated_at = args.generated_at or datetime.now(timezone.utc).isoformat()
+        validate_response(response)
+        generated_at = resolve_generated_at(args, response)
         markdown, structured = build_output(response, generated_at)
     except (OSError, ValueError, json.JSONDecodeError) as exc:
         raise SystemExit("error: {}".format(exc))

examples/source_linked_briefing/fixtures/search_response.json

@@ -1,4 +1,5 @@
 {
+  "_fixture_generated_at": "2026-07-25T00:00:00Z",
   "status": "ok",
   "news": [
     {

tests/test_source_linked_briefing_example.py

@@ -133,6 +133,46 @@ def test_fixture_mode_rejects_malformed_search_response(tmp_path):
     assert not output_dir.exists()
 
 
+def test_fixture_mode_rejects_malformed_article_fields(tmp_path):
+    fixture = tmp_path / "search-response.json"
+    fixture.write_text(
+        json.dumps(
+            {
+                "status": "ok",
+                "news": [
+                    {
+                        "title": ["not", "a", "string"],
+                        "url": "https://news.example/invalid",
+                        "published": "2026-07-25T12:00:00Z",
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    output_dir = tmp_path / "output"
+
+    result = subprocess.run(
+        [
+            sys.executable,
+            str(EXAMPLE),
+            "--fixture",
+            str(fixture),
+            "--output-dir",
+            str(output_dir),
+            "--generated-at",
+            "2026-07-25T13:00:00Z",
+        ],
+        capture_output=True,
+        check=False,
+        text=True,
+    )
+
+    assert result.returncode != 0
+    assert "news item title must be a string" in result.stderr
+    assert not output_dir.exists()
+
+
 def test_live_mode_requires_api_key(tmp_path):
     env = os.environ.copy()
     env.pop("CURRENTS_API_KEY", None)
@@ -181,6 +221,34 @@ def test_checked_in_fixture_runs_without_network_or_credentials(tmp_path):
     assert "https://example.com/energy/storage-policy" in markdown
 
 
+def test_checked_in_fixture_produces_identical_output_on_every_run(tmp_path):
+    first_output = tmp_path / "first"
+    second_output = tmp_path / "second"
+
+    for output_dir in (first_output, second_output):
+        result = subprocess.run(
+            [
+                sys.executable,
+                str(EXAMPLE),
+                "--fixture",
+                str(CHECKED_IN_FIXTURE),
+                "--output-dir",
+                str(output_dir),
+            ],
+            capture_output=True,
+            check=False,
+            text=True,
+        )
+        assert result.returncode == 0, result.stderr
+
+    assert (first_output / "briefing.md").read_bytes() == (
+        second_output / "briefing.md"
+    ).read_bytes()
+    assert (first_output / "briefing.json").read_bytes() == (
+        second_output / "briefing.json"
+    ).read_bytes()
+
+
 def test_live_mode_uses_the_sdk_search_result(tmp_path):
     fake_package = tmp_path / "fake-sdk" / "currentsapi"
     fake_package.mkdir(parents=True)