diff --git a/tests/test_ai_explainer.py b/tests/test_ai_explainer.py index dbf0ff6..322f91e 100644 --- a/tests/test_ai_explainer.py +++ b/tests/test_ai_explainer.py @@ -496,25 +496,26 @@ def test_multiline_detail(self) -> None: class TestFormatExplanationLine(unittest.TestCase): """Tests for format_explanation_line.""" - @patch("utils.llm.ai_explainer.upload_to_paste", return_value="https://dpaste.com/abc123") + @patch("utils.llm.ai_explainer.upload_to_paste", return_value="https://rentry.co/abc123") def test_format_with_detail(self, mock_paste: MagicMock) -> None: explanation = Explanation(summary="This pauses the protocol.", detail="Full detail here.") result = format_explanation_line(explanation) self.assertIn("AI Summary", result) self.assertIn("This pauses the protocol.", result) self.assertNotIn("Full detail here.", result) - self.assertIn("https://dpaste.com/abc123", result) + self.assertIn("https://rentry.co/abc123", result) self.assertIn("Full details", result) mock_paste.assert_called_once_with("Full detail here.", title="AI Transaction Analysis") @patch("utils.llm.ai_explainer.upload_to_paste", return_value="") def test_format_paste_failure(self, mock_paste: MagicMock) -> None: - """If paste upload fails, no link is included.""" + """If paste upload fails, surface a notice instead of a link.""" explanation = Explanation(summary="This pauses the protocol.", detail="Full detail here.") result = format_explanation_line(explanation) self.assertIn("AI Summary", result) self.assertIn("This pauses the protocol.", result) self.assertNotIn("Full details", result) + self.assertIn("Couldn't post full report", result) def test_format_no_detail(self) -> None: """If there's no detail, no paste upload is attempted.""" diff --git a/tests/test_paste.py b/tests/test_paste.py index d522e72..9394e62 100644 --- a/tests/test_paste.py +++ b/tests/test_paste.py @@ -3,61 +3,79 @@ import unittest from unittest.mock import MagicMock, patch +import requests + from utils.paste import upload_to_paste +def _mock_session(*, csrftoken: str = "tok", post_json: dict | None = None) -> MagicMock: + """Build a mock requests.Session for the rentry CSRF flow.""" + session = MagicMock() + session.__enter__.return_value = session + session.__exit__.return_value = False + session.cookies.get.return_value = csrftoken + + get_resp = MagicMock() + get_resp.raise_for_status.return_value = None + session.get.return_value = get_resp + + post_resp = MagicMock() + post_resp.raise_for_status.return_value = None + post_resp.json.return_value = post_json or {"status": "200", "url": "https://rentry.co/abc123"} + session.post.return_value = post_resp + return session + + class TestUploadToPaste(unittest.TestCase): """Tests for upload_to_paste.""" def test_empty_content_returns_empty(self) -> None: - result = upload_to_paste("") - self.assertEqual(result, "") - - @patch("utils.paste.requests.post") - def test_successful_upload(self, mock_post: MagicMock) -> None: - mock_response = MagicMock() - mock_response.text = '"https://dpaste.com/abc123"\n' - mock_response.raise_for_status.return_value = None - mock_post.return_value = mock_response - - result = upload_to_paste("Some content", title="Test") - self.assertEqual(result, "https://dpaste.com/abc123") - mock_post.assert_called_once() - call_kwargs = mock_post.call_args - self.assertEqual(call_kwargs[1]["data"]["content"], "Some content") - self.assertEqual(call_kwargs[1]["data"]["title"], "Test") - self.assertEqual(call_kwargs[1]["data"]["expiry_days"], 7) - - @patch("utils.paste.requests.post") - def test_custom_expiry(self, mock_post: MagicMock) -> None: - mock_response = MagicMock() - mock_response.text = "https://dpaste.com/xyz789" - mock_response.raise_for_status.return_value = None - mock_post.return_value = mock_response - - result = upload_to_paste("Content", expiry_days=15) - call_kwargs = mock_post.call_args - self.assertEqual(call_kwargs[1]["data"]["expiry_days"], 15) - self.assertEqual(result, "https://dpaste.com/xyz789") - - @patch("utils.paste.requests.post") - def test_request_failure_returns_empty(self, mock_post: MagicMock) -> None: - import requests - - mock_post.side_effect = requests.RequestException("Connection error") - result = upload_to_paste("Some content") - self.assertEqual(result, "") - - @patch("utils.paste.requests.post") - def test_no_title(self, mock_post: MagicMock) -> None: - mock_response = MagicMock() - mock_response.text = "https://dpaste.com/notitle" - mock_response.raise_for_status.return_value = None - mock_post.return_value = mock_response + self.assertEqual(upload_to_paste(""), "") + + @patch("utils.paste.requests.Session") + def test_successful_upload(self, mock_session_cls: MagicMock) -> None: + session = _mock_session() + mock_session_cls.return_value = session + + result = upload_to_paste("Some **markdown**", title="Test") + self.assertEqual(result, "https://rentry.co/abc123") + + # Title is prepended as a markdown heading, csrftoken echoed back. + data = session.post.call_args[1]["data"] + self.assertEqual(data["text"], "# Test\n\nSome **markdown**") + self.assertEqual(data["csrfmiddlewaretoken"], "tok") + self.assertEqual(session.post.call_args[1]["headers"]["Referer"], "https://rentry.co") + + @patch("utils.paste.requests.Session") + def test_no_title_sends_raw_content(self, mock_session_cls: MagicMock) -> None: + session = _mock_session() + mock_session_cls.return_value = session upload_to_paste("Content only") - call_kwargs = mock_post.call_args - self.assertNotIn("title", call_kwargs[1]["data"]) + self.assertEqual(session.post.call_args[1]["data"]["text"], "Content only") + + @patch("utils.paste.requests.Session") + def test_missing_csrftoken_returns_empty(self, mock_session_cls: MagicMock) -> None: + session = _mock_session(csrftoken=None) + mock_session_cls.return_value = session + + self.assertEqual(upload_to_paste("x"), "") + session.post.assert_not_called() + + @patch("utils.paste.requests.Session") + def test_non_200_status_returns_empty(self, mock_session_cls: MagicMock) -> None: + session = _mock_session(post_json={"status": "400", "errors": "bad request"}) + mock_session_cls.return_value = session + + self.assertEqual(upload_to_paste("x"), "") + + @patch("utils.paste.requests.Session") + def test_request_failure_returns_empty(self, mock_session_cls: MagicMock) -> None: + session = _mock_session() + session.post.side_effect = requests.RequestException("Connection error") + mock_session_cls.return_value = session + + self.assertEqual(upload_to_paste("Some content"), "") if __name__ == "__main__": diff --git a/utils/llm/ai_explainer.py b/utils/llm/ai_explainer.py index 122a821..1734042 100644 --- a/utils/llm/ai_explainer.py +++ b/utils/llm/ai_explainer.py @@ -1129,11 +1129,13 @@ def format_explanation_line(explanation: Explanation) -> str: """Format the AI explanation for inclusion in a Telegram alert message. Uses the short summary for the Telegram message. The detailed analysis - is uploaded to a paste service (dpaste.org) for easy access. + is uploaded to a paste service (rentry.co) for easy access. """ line = f"\n🤖 *AI Summary:*\n{escape_markdown(explanation.summary)}" if explanation.detail: paste_url = upload_to_paste(explanation.detail, title="AI Transaction Analysis") if paste_url: line += f"\n[Full details]({paste_url})" + else: + line += "\n⚠️ Couldn't post full report" return line diff --git a/utils/paste.py b/utils/paste.py index c44e248..f510b08 100644 --- a/utils/paste.py +++ b/utils/paste.py @@ -1,6 +1,8 @@ -"""Upload text to a paste service for temporary sharing. +"""Upload markdown text to rentry.co and return the URL. -Uses dpaste.com with configurable expiry (default 7 days). +rentry.co renders markdown and accepts anonymous pastes through a small CSRF +flow: fetch the ``csrftoken`` cookie from the site, then POST the content with +the matching ``csrfmiddlewaretoken`` field and a ``Referer`` header. """ import requests @@ -9,37 +11,52 @@ logger = get_logger("utils.paste") -DPASTE_API_URL = "https://dpaste.com/api/" -DEFAULT_EXPIRY_DAYS = 7 +RENTRY_BASE_URL = "https://rentry.co" +RENTRY_API_NEW_URL = f"{RENTRY_BASE_URL}/api/new" -def upload_to_paste(content: str, title: str = "", expiry_days: int = DEFAULT_EXPIRY_DAYS) -> str: - """Upload text content to dpaste.com and return the URL. +def upload_to_paste(content: str, title: str = "") -> str: + """Upload markdown ``content`` to rentry.co and return the rendered-page URL. Args: - content: The text content to upload. - title: Optional title for the paste. - expiry_days: Number of days before the paste expires (default 7). + content: The markdown text to upload. + title: Optional title, prepended as a top-level markdown heading. Returns: - The URL of the created paste, or empty string on failure. + The URL of the created paste, or an empty string on failure. """ if not content: return "" - payload: dict[str, str | int] = { - "content": content, - "expiry_days": expiry_days, - } - if title: - payload["title"] = title + text = f"# {title}\n\n{content}" if title else content try: - response = requests.post(DPASTE_API_URL, data=payload, timeout=10) - response.raise_for_status() - url = response.text.strip().strip('"') - logger.info("Uploaded paste to %s (expires in %d days)", url, expiry_days) - return url - except requests.RequestException as e: + with requests.Session() as session: + # Priming GET sets the csrftoken cookie that the POST must echo back. + session.get(RENTRY_BASE_URL, timeout=10).raise_for_status() + csrftoken = session.cookies.get("csrftoken") + if not csrftoken: + logger.warning("Failed to upload to paste service: no csrftoken cookie from rentry") + return "" + + response = session.post( + RENTRY_API_NEW_URL, + data={"csrfmiddlewaretoken": csrftoken, "text": text}, + headers={"Referer": RENTRY_BASE_URL}, + timeout=10, + ) + response.raise_for_status() + payload = response.json() + except (requests.RequestException, ValueError) as e: logger.warning("Failed to upload to paste service: %s", e) return "" + + # rentry signals success with status "200" in the JSON body; anything else + # carries the reason in "errors"/"content". + if str(payload.get("status")) != "200": + logger.warning("Paste service rejected upload: %s", payload.get("errors") or payload.get("content") or payload) + return "" + + url = payload.get("url", "") + logger.info("Uploaded paste to %s", url) + return url