Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions tests/test_ai_explainer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
110 changes: 64 additions & 46 deletions tests/test_paste.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__":
Expand Down
4 changes: 3 additions & 1 deletion utils/llm/ai_explainer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
61 changes: 39 additions & 22 deletions utils/paste.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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