| 1 | """Tests for optional hosted HTML publishing.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import io |
| 6 | import json |
| 7 | import os |
| 8 | import sys |
| 9 | import tempfile |
| 10 | import unittest |
| 11 | from contextlib import redirect_stderr, redirect_stdout |
| 12 | from pathlib import Path |
| 13 | from unittest import mock |
| 14 | from urllib.error import HTTPError |
| 15 | |
| 16 | import last30days as cli |
| 17 | from lib import html_publish, schema |
| 18 | |
| 19 | |
| 20 | def _report(topic: str = "OpenClaw") -> schema.Report: |
| 21 | return schema.Report( |
| 22 | topic=topic, |
| 23 | range_from="2026-05-01", |
| 24 | range_to="2026-05-31", |
| 25 | generated_at="2026-05-31T00:00:00+00:00", |
| 26 | provider_runtime=schema.ProviderRuntime( |
| 27 | reasoning_provider="local", |
| 28 | planner_model="mock-planner", |
| 29 | rerank_model="mock-rerank", |
| 30 | ), |
| 31 | query_plan=schema.QueryPlan( |
| 32 | intent="concept", |
| 33 | freshness_mode="balanced_recent", |
| 34 | cluster_mode="none", |
| 35 | raw_topic=topic, |
| 36 | subqueries=[ |
| 37 | schema.SubQuery( |
| 38 | label="primary", |
| 39 | search_query=topic, |
| 40 | ranking_query=topic, |
| 41 | sources=["grounding"], |
| 42 | ) |
| 43 | ], |
| 44 | source_weights={"grounding": 1.0}, |
| 45 | ), |
| 46 | clusters=[], |
| 47 | ranked_candidates=[], |
| 48 | items_by_source={"grounding": []}, |
| 49 | errors_by_source={}, |
| 50 | ) |
| 51 | |
| 52 | |
| 53 | def _diag() -> dict[str, object]: |
| 54 | return { |
| 55 | "available_sources": ["grounding"], |
| 56 | "providers": {"google": True, "openai": False, "xai": False}, |
| 57 | "x_backend": None, |
| 58 | "bird_installed": True, |
| 59 | "bird_authenticated": False, |
| 60 | "bird_username": None, |
| 61 | "native_web_backend": "brave", |
| 62 | } |
| 63 | |
| 64 | |
| 65 | class _FakeResponse: |
| 66 | def __init__(self, payload: object): |
| 67 | self.payload = payload |
| 68 | |
| 69 | def __enter__(self): |
| 70 | return self |
| 71 | |
| 72 | def __exit__(self, *_args): |
| 73 | return False |
| 74 | |
| 75 | def read(self) -> bytes: |
| 76 | return json.dumps(self.payload).encode("utf-8") |
| 77 | |
| 78 | |
| 79 | class HtmlPublishModuleTests(unittest.TestCase): |
| 80 | def test_publish_posts_html_and_password(self): |
| 81 | captured = {} |
| 82 | |
| 83 | def opener(request, timeout): |
| 84 | captured["timeout"] = timeout |
| 85 | captured["url"] = request.full_url |
| 86 | captured["headers"] = dict(request.header_items()) |
| 87 | captured["body"] = json.loads(request.data.decode("utf-8")) |
| 88 | return _FakeResponse( |
| 89 | { |
| 90 | "url": "https://example.ht-ml.app", |
| 91 | "site_id": "site_123", |
| 92 | "status": "active", |
| 93 | "update_key": "secret-update-key", |
| 94 | } |
| 95 | ) |
| 96 | |
| 97 | result = html_publish.publish_html( |
| 98 | "<html>ok</html>", |
| 99 | password="share-pass", |
| 100 | opener=opener, |
| 101 | ) |
| 102 | |
| 103 | self.assertEqual("https://api.ht-ml.app/v1/sites", captured["url"]) |
| 104 | self.assertEqual({"html_content": "<html>ok</html>", "password": "share-pass"}, captured["body"]) |
| 105 | self.assertEqual("https://example.ht-ml.app", result["url"]) |
| 106 | self.assertEqual("secret-update-key", result["update_key"]) |
| 107 | |
| 108 | def test_publish_rejects_http_error_with_message(self): |
| 109 | def opener(_request, timeout): |
| 110 | raise HTTPError( |
| 111 | "https://api.ht-ml.app/v1/sites", |
| 112 | 400, |
| 113 | "Bad Request", |
| 114 | {}, |
| 115 | io.BytesIO(b'{"message":"HTML content is required"}'), |
| 116 | ) |
| 117 | |
| 118 | with self.assertRaisesRegex(html_publish.HtmlPublishError, "HTML content is required"): |
| 119 | html_publish.publish_html("<html></html>", opener=opener) |
| 120 | |
| 121 | def test_publish_rejects_json_response_that_is_not_an_object(self): |
| 122 | def opener(_request, timeout): |
| 123 | return _FakeResponse(["https://site.ht-ml.app"]) |
| 124 | |
| 125 | with self.assertRaisesRegex(html_publish.HtmlPublishError, "unexpected JSON response"): |
| 126 | html_publish.publish_html("<html></html>", opener=opener) |
| 127 | |
| 128 | def test_publish_html_documents_publishes_each_named_document(self): |
| 129 | with mock.patch.object( |
| 130 | html_publish, |
| 131 | "publish_html", |
| 132 | side_effect=[{"url": "https://one.ht-ml.app"}, {"url": "https://two.ht-ml.app"}], |
| 133 | ) as publish: |
| 134 | results = html_publish.publish_html_documents( |
| 135 | {"one": "<html>one</html>", "two": "<html>two</html>"}, |
| 136 | password="shared", |
| 137 | ) |
| 138 | |
| 139 | self.assertEqual( |
| 140 | {"one": {"url": "https://one.ht-ml.app"}, "two": {"url": "https://two.ht-ml.app"}}, |
| 141 | results, |
| 142 | ) |
| 143 | self.assertEqual(2, publish.call_count) |
| 144 | self.assertEqual("shared", publish.call_args_list[0].kwargs["password"]) |
| 145 | |
| 146 | |
| 147 | class HtmlPublishCliTests(unittest.TestCase): |
| 148 | def test_publish_requires_html_emit(self): |
| 149 | with mock.patch.object(sys, "argv", [ |
| 150 | "last30days.py", |
| 151 | "OpenClaw", |
| 152 | "--emit=md", |
| 153 | "--publish-html", |
| 154 | ]), mock.patch.object(cli.env, "get_config", return_value={}): |
| 155 | stderr = io.StringIO() |
| 156 | with redirect_stderr(stderr): |
| 157 | rc = cli.main() |
| 158 | self.assertEqual(2, rc) |
| 159 | self.assertIn("--publish-html requires --emit=html", stderr.getvalue()) |
| 160 | |
| 161 | def test_publish_writes_url_metadata_without_update_key(self): |
| 162 | with tempfile.TemporaryDirectory() as tmp: |
| 163 | output_path = Path(tmp) / "brief.html" |
| 164 | with mock.patch.object(cli.env, "get_config", return_value={}), \ |
| 165 | mock.patch.object(cli.pipeline, "diagnose", return_value=_diag()), \ |
| 166 | mock.patch.object(cli.pipeline, "run", return_value=_report()), \ |
| 167 | mock.patch.object(cli, "emit_output", return_value="<html>brief</html>"), \ |
| 168 | mock.patch.object(cli, "publish_rendered_html", wraps=cli.publish_rendered_html) as publish_wrapper, \ |
| 169 | mock.patch("lib.html_publish.publish_html", return_value={ |
| 170 | "url": "https://site.ht-ml.app", |
| 171 | "site_id": "site_123", |
| 172 | "status": "active", |
| 173 | "update_key": "secret-update-key", |
| 174 | }), \ |
| 175 | mock.patch.object(sys, "argv", [ |
| 176 | "last30days.py", |
| 177 | "OpenClaw", |
| 178 | "--emit=html", |
| 179 | "--output", |
| 180 | str(output_path), |
| 181 | "--publish-html", |
| 182 | ]), \ |
| 183 | mock.patch.dict(os.environ, {"LAST30DAYS_SKIP_PREFLIGHT": "1"}, clear=False): |
| 184 | stdout = io.StringIO() |
| 185 | stderr = io.StringIO() |
| 186 | with redirect_stdout(stdout), redirect_stderr(stderr): |
| 187 | rc = cli.main() |
| 188 | |
| 189 | self.assertEqual(0, rc) |
| 190 | publish_wrapper.assert_called_once() |
| 191 | self.assertEqual("<html>brief</html>", output_path.read_text(encoding="utf-8")) |
| 192 | self.assertIn("Published HTML to https://site.ht-ml.app", stderr.getvalue()) |
| 193 | self.assertNotIn("secret-update-key", stdout.getvalue()) |
| 194 | self.assertNotIn("secret-update-key", stderr.getvalue()) |
| 195 | metadata = json.loads((Path(str(output_path) + ".publish.json")).read_text(encoding="utf-8")) |
| 196 | self.assertEqual("https://site.ht-ml.app", metadata["url"]) |
| 197 | self.assertNotIn("update_key", metadata) |
| 198 | |
| 199 | def test_publish_uses_password_from_environment(self): |
| 200 | with tempfile.TemporaryDirectory() as tmp: |
| 201 | output_path = Path(tmp) / "brief.html" |
| 202 | with mock.patch.object(cli.env, "get_config", return_value={}), \ |
| 203 | mock.patch.object(cli.pipeline, "diagnose", return_value=_diag()), \ |
| 204 | mock.patch.object(cli.pipeline, "run", return_value=_report()), \ |
| 205 | mock.patch.object(cli, "emit_output", return_value="<html>brief</html>"), \ |
| 206 | mock.patch("lib.html_publish.publish_html", return_value={ |
| 207 | "url": "https://site.ht-ml.app", |
| 208 | }) as publish_mock, \ |
| 209 | mock.patch.object(sys, "argv", [ |
| 210 | "last30days.py", |
| 211 | "OpenClaw", |
| 212 | "--emit=html", |
| 213 | "--output", |
| 214 | str(output_path), |
| 215 | "--publish-html", |
| 216 | ]), \ |
| 217 | mock.patch.dict(os.environ, { |
| 218 | "LAST30DAYS_SKIP_PREFLIGHT": "1", |
| 219 | "LAST30DAYS_PUBLISH_PASSWORD": "share-pass", |
| 220 | }, clear=False): |
| 221 | stdout = io.StringIO() |
| 222 | stderr = io.StringIO() |
| 223 | with redirect_stdout(stdout), redirect_stderr(stderr): |
| 224 | rc = cli.main() |
| 225 | |
| 226 | self.assertEqual(0, rc) |
| 227 | self.assertEqual("share-pass", publish_mock.call_args.kwargs["password"]) |
| 228 | |
| 229 | def test_publish_metadata_failure_still_reports_url(self): |
| 230 | with tempfile.TemporaryDirectory() as tmp: |
| 231 | output_path = Path(tmp) / "brief.html" |
| 232 | with mock.patch.object(cli.env, "get_config", return_value={}), \ |
| 233 | mock.patch.object(cli.pipeline, "diagnose", return_value=_diag()), \ |
| 234 | mock.patch.object(cli.pipeline, "run", return_value=_report()), \ |
| 235 | mock.patch.object(cli, "emit_output", return_value="<html>brief</html>"), \ |
| 236 | mock.patch("lib.html_publish.publish_html", return_value={ |
| 237 | "url": "https://site.ht-ml.app", |
| 238 | "site_id": "site_123", |
| 239 | }), \ |
| 240 | mock.patch.object(cli, "_write_publish_metadata", side_effect=PermissionError("denied")), \ |
| 241 | mock.patch.object(sys, "argv", [ |
| 242 | "last30days.py", |
| 243 | "OpenClaw", |
| 244 | "--emit=html", |
| 245 | "--output", |
| 246 | str(output_path), |
| 247 | "--publish-html", |
| 248 | ]), \ |
| 249 | mock.patch.dict(os.environ, {"LAST30DAYS_SKIP_PREFLIGHT": "1"}, clear=False): |
| 250 | stdout = io.StringIO() |
| 251 | stderr = io.StringIO() |
| 252 | with redirect_stdout(stdout), redirect_stderr(stderr): |
| 253 | rc = cli.main() |
| 254 | |
| 255 | self.assertEqual(0, rc) |
| 256 | self.assertIn("Published HTML to https://site.ht-ml.app", stderr.getvalue()) |
| 257 | self.assertIn("Publish metadata warning", stderr.getvalue()) |
| 258 | |
| 259 | def test_publish_failure_preserves_local_output(self): |
| 260 | with tempfile.TemporaryDirectory() as tmp: |
| 261 | output_path = Path(tmp) / "brief.html" |
| 262 | with mock.patch.object(cli.env, "get_config", return_value={}), \ |
| 263 | mock.patch.object(cli.pipeline, "diagnose", return_value=_diag()), \ |
| 264 | mock.patch.object(cli.pipeline, "run", return_value=_report()), \ |
| 265 | mock.patch.object(cli, "emit_output", return_value="<html>brief</html>"), \ |
| 266 | mock.patch("lib.html_publish.publish_html", side_effect=html_publish.HtmlPublishError("timeout")), \ |
| 267 | mock.patch.object(sys, "argv", [ |
| 268 | "last30days.py", |
| 269 | "OpenClaw", |
| 270 | "--emit=html", |
| 271 | "--output", |
| 272 | str(output_path), |
| 273 | "--publish-html", |
| 274 | ]), \ |
| 275 | mock.patch.dict(os.environ, {"LAST30DAYS_SKIP_PREFLIGHT": "1"}, clear=False): |
| 276 | stdout = io.StringIO() |
| 277 | stderr = io.StringIO() |
| 278 | with redirect_stdout(stdout), redirect_stderr(stderr): |
| 279 | rc = cli.main() |
| 280 | |
| 281 | self.assertEqual(0, rc) |
| 282 | self.assertEqual("<html>brief</html>", output_path.read_text(encoding="utf-8")) |
| 283 | self.assertIn("HTML publish failed: timeout", stderr.getvalue()) |
| 284 | |
| 285 | |
| 286 | if __name__ == "__main__": |
| 287 | unittest.main() |
| 288 |