Merge pull request #96 from MervinPraison/claude/issue-39-20260902-1212

feat: gated Reddit write actions + User-Agent template + tests

praisonai_tools/tools/reddit_tool.py

@@ -17,6 +17,17 @@ Environment Variables:
     REDDIT_CLIENT_ID: Reddit API client ID
     REDDIT_CLIENT_SECRET: Reddit API client secret
     REDDIT_USER_AGENT: User agent string (optional)
+    REDDIT_USERNAME: Reddit username (required for write actions)
+    REDDIT_PASSWORD: Reddit password (required for write actions)
+    REDDIT_ALLOW_WRITE: Set to "1" to enable write actions (post/comment)
+
+Write actions (submit_post, post_comment) are disabled by default to reduce
+spam/abuse surface. They are only permitted when ``REDDIT_ALLOW_WRITE=1`` (or
+``allow_write=True``) and Reddit account credentials are configured. You are
+responsible for complying with Reddit's API terms when enabling writes.
+
+Reddit rate-limits misconfigured clients aggressively. Use a descriptive
+User-Agent referencing your app id, e.g. ``script:my-app:v1.0 (by /u/you)``.
 """
 
 import os
@@ -32,13 +43,20 @@ class RedditTool(BaseTool):
     """Tool for interacting with Reddit."""
     
     name = "reddit"
-    description = "Search Reddit, get posts from subreddits, and read comments."
+    description = (
+        "Search Reddit, get posts from subreddits, and read comments. "
+        "Can also comment on posts and submit new posts when write actions "
+        "are enabled (REDDIT_ALLOW_WRITE=1 with account credentials)."
+    )
     
     def __init__(
         self,
         client_id: Optional[str] = None,
         client_secret: Optional[str] = None,
         user_agent: Optional[str] = None,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        allow_write: Optional[bool] = None,
     ):
         """Initialize RedditTool.
         
@@ -46,13 +64,30 @@ class RedditTool(BaseTool):
             client_id: Reddit API client ID
             client_secret: Reddit API client secret
             user_agent: User agent string
+            username: Reddit username (required for write actions)
+            password: Reddit password (required for write actions)
+            allow_write: Enable write actions. Defaults to the
+                ``REDDIT_ALLOW_WRITE`` env var ("1"/"true" enables).
         """
         self.client_id = client_id or os.getenv("REDDIT_CLIENT_ID")
         self.client_secret = client_secret or os.getenv("REDDIT_CLIENT_SECRET")
-        self.user_agent = user_agent or os.getenv("REDDIT_USER_AGENT", "PraisonAI/1.0")
+        self.user_agent = user_agent or os.getenv("REDDIT_USER_AGENT") or self._default_user_agent()
+        self.username = username or os.getenv("REDDIT_USERNAME")
+        self.password = password or os.getenv("REDDIT_PASSWORD")
+        if allow_write is None:
+            allow_write = os.getenv("REDDIT_ALLOW_WRITE", "").lower() in ("1", "true", "yes")
+        self.allow_write = allow_write
         self._reddit = None
         super().__init__()
     
+    def _default_user_agent(self) -> str:
+        """Build a compliant User-Agent referencing the app id.
+        
+        Reddit recommends ``<platform>:<app id>:<version> (by /u/<user>)``.
+        """
+        app_id = self.client_id or "unknown"
+        return f"script:praisonai-{app_id}:1.0 (by PraisonAI)"
+    
     @property
     def reddit(self):
         """Lazy-load Reddit client."""
@@ -65,13 +100,31 @@ class RedditTool(BaseTool):
             if not self.client_id or not self.client_secret:
                 raise ValueError("REDDIT_CLIENT_ID and REDDIT_CLIENT_SECRET required")
             
-            self._reddit = praw.Reddit(
-                client_id=self.client_id,
-                client_secret=self.client_secret,
-                user_agent=self.user_agent,
-            )
+            kwargs = {
+                "client_id": self.client_id,
+                "client_secret": self.client_secret,
+                "user_agent": self.user_agent,
+            }
+            if self.username and self.password:
+                kwargs["username"] = self.username
+                kwargs["password"] = self.password
+            
+            self._reddit = praw.Reddit(**kwargs)
         return self._reddit
     
+    def _check_write_allowed(self) -> Optional[Dict[str, Any]]:
+        """Return an error dict if write actions are not permitted, else None."""
+        if not self.allow_write:
+            return {
+                "error": "Write actions disabled. Set REDDIT_ALLOW_WRITE=1 to enable "
+                "posting/commenting.",
+            }
+        if not self.username or not self.password:
+            return {
+                "error": "REDDIT_USERNAME and REDDIT_PASSWORD required for write actions.",
+            }
+        return None
+    
     def run(
         self,
         action: str = "search",
@@ -98,6 +151,15 @@ class RedditTool(BaseTool):
             return self.get_comments(post_id=post_id, limit=limit)
         elif action == "subreddit_info":
             return self.get_subreddit_info(subreddit=subreddit)
+        elif action == "post_comment":
+            return self.post_comment(post_id=post_id, body=kwargs.get("body"))
+        elif action == "submit_post":
+            return self.submit_post(
+                subreddit=subreddit,
+                title=kwargs.get("title"),
+                body=kwargs.get("body"),
+                url=kwargs.get("url"),
+            )
         else:
             return {"error": f"Unknown action: {action}"}
     
@@ -284,6 +346,83 @@ class RedditTool(BaseTool):
             logger.error(f"Reddit get_subreddit_info error: {e}")
             return {"error": str(e)}
     
+    def post_comment(self, post_id: str, body: str) -> Dict[str, Any]:
+        """Reply with a comment to a post (write action).
+        
+        Gated behind ``REDDIT_ALLOW_WRITE=1`` and account credentials.
+        
+        Args:
+            post_id: Reddit post ID to comment on
+            body: Comment text
+            
+        Returns:
+            Result with the new comment id, or an error dict
+        """
+        denied = self._check_write_allowed()
+        if denied:
+            return denied
+        if not post_id:
+            return {"error": "post_id is required"}
+        if not body:
+            return {"error": "body is required"}
+        
+        try:
+            submission = self.reddit.submission(id=post_id)
+            comment = submission.reply(body)
+            return {
+                "success": True,
+                "id": comment.id,
+                "permalink": f"https://reddit.com{comment.permalink}",
+            }
+        except Exception as e:
+            logger.error(f"Reddit post_comment error: {e}")
+            return {"error": str(e)}
+    
+    def submit_post(
+        self,
+        subreddit: str,
+        title: str,
+        body: Optional[str] = None,
+        url: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Submit a new post to a subreddit (write action).
+        
+        Gated behind ``REDDIT_ALLOW_WRITE=1`` and account credentials.
+        
+        Args:
+            subreddit: Target subreddit name
+            title: Post title
+            body: Self-post text (mutually exclusive with url)
+            url: Link URL (mutually exclusive with body)
+            
+        Returns:
+            Result with the new post id, or an error dict
+        """
+        denied = self._check_write_allowed()
+        if denied:
+            return denied
+        if not subreddit:
+            return {"error": "subreddit is required"}
+        if not title:
+            return {"error": "title is required"}
+        if url and body:
+            return {"error": "provide either body or url, not both"}
+        
+        try:
+            sub = self.reddit.subreddit(subreddit)
+            if url:
+                submission = sub.submit(title, url=url)
+            else:
+                submission = sub.submit(title, selftext=body or "")
+            return {
+                "success": True,
+                "id": submission.id,
+                "permalink": f"https://reddit.com{submission.permalink}",
+            }
+        except Exception as e:
+            logger.error(f"Reddit submit_post error: {e}")
+            return {"error": str(e)}
+    
     def _format_post(self, post, include_body: bool = False) -> Dict[str, Any]:
         """Format a Reddit post."""
         result = {

tests/unit/tools/test_reddit_tool.py

@@ -0,0 +1,176 @@
+"""Unit tests for RedditTool."""
+
+import os
+from unittest.mock import MagicMock, patch
+
+from praisonai_tools.tools.reddit_tool import RedditTool
+
+
+# ── Import + schema (CI smoke) ──────────────────────────────────────
+
+
+class TestImportAndSchema:
+    def test_import_via_package(self):
+        from praisonai_tools import RedditTool as PkgRedditTool
+
+        assert PkgRedditTool is RedditTool
+
+    def test_schema_generation(self):
+        tool = RedditTool(client_id="cid", client_secret="secret")
+        schema = tool.get_schema()
+        assert schema["function"]["name"] == "reddit"
+        assert "description" in schema["function"]
+        assert "action" in schema["function"]["parameters"]["properties"]
+
+
+# ── User-Agent ──────────────────────────────────────────────────────
+
+
+class TestUserAgent:
+    def test_default_user_agent_references_app_id(self):
+        with patch.dict(os.environ, {}, clear=True):
+            tool = RedditTool(client_id="myappid", client_secret="secret")
+        assert "myappid" in tool.user_agent
+
+    def test_explicit_user_agent_wins(self):
+        tool = RedditTool(client_id="c", client_secret="s", user_agent="custom:ua:1.0")
+        assert tool.user_agent == "custom:ua:1.0"
+
+    def test_env_user_agent_used(self):
+        with patch.dict(os.environ, {"REDDIT_USER_AGENT": "env:ua:1.0"}, clear=True):
+            tool = RedditTool(client_id="c", client_secret="s")
+        assert tool.user_agent == "env:ua:1.0"
+
+
+# ── Write flag gating ───────────────────────────────────────────────
+
+
+class TestWriteGating:
+    def test_write_disabled_by_default(self):
+        with patch.dict(os.environ, {}, clear=True):
+            tool = RedditTool(client_id="c", client_secret="s")
+        assert tool.allow_write is False
+
+    def test_env_enables_write(self):
+        with patch.dict(os.environ, {"REDDIT_ALLOW_WRITE": "1"}, clear=True):
+            tool = RedditTool(client_id="c", client_secret="s")
+        assert tool.allow_write is True
+
+    def test_post_comment_blocked_when_write_disabled(self):
+        with patch.dict(os.environ, {}, clear=True):
+            tool = RedditTool(client_id="c", client_secret="s")
+        result = tool.post_comment(post_id="abc", body="hi")
+        assert "REDDIT_ALLOW_WRITE" in result["error"]
+
+    def test_submit_post_blocked_when_write_disabled(self):
+        with patch.dict(os.environ, {}, clear=True):
+            tool = RedditTool(client_id="c", client_secret="s")
+        result = tool.submit_post(subreddit="test", title="T", body="B")
+        assert "REDDIT_ALLOW_WRITE" in result["error"]
+
+    def test_write_requires_account_credentials(self):
+        tool = RedditTool(client_id="c", client_secret="s", allow_write=True)
+        result = tool.post_comment(post_id="abc", body="hi")
+        assert "REDDIT_USERNAME" in result["error"]
+
+
+# ── Write actions (mocked praw) ─────────────────────────────────────
+
+
+def _writable_tool():
+    return RedditTool(
+        client_id="c",
+        client_secret="s",
+        username="u",
+        password="p",
+        allow_write=True,
+    )
+
+
+class TestPostComment:
+    def test_requires_post_id(self):
+        tool = _writable_tool()
+        assert tool.post_comment(post_id="", body="hi") == {"error": "post_id is required"}
+
+    def test_requires_body(self):
+        tool = _writable_tool()
+        assert tool.post_comment(post_id="abc", body="") == {"error": "body is required"}
+
+    def test_success(self):
+        tool = _writable_tool()
+        comment = MagicMock()
+        comment.id = "c1"
+        comment.permalink = "/r/test/comments/abc/x/c1/"
+        submission = MagicMock()
+        submission.reply.return_value = comment
+        reddit = MagicMock()
+        reddit.submission.return_value = submission
+        with patch.object(RedditTool, "reddit", new=reddit):
+            result = tool.post_comment(post_id="abc", body="hello")
+        submission.reply.assert_called_once_with("hello")
+        assert result["success"] is True
+        assert result["id"] == "c1"
+
+
+class TestSubmitPost:
+    def test_requires_subreddit(self):
+        tool = _writable_tool()
+        assert tool.submit_post(subreddit="", title="T") == {"error": "subreddit is required"}
+
+    def test_requires_title(self):
+        tool = _writable_tool()
+        assert tool.submit_post(subreddit="test", title="") == {"error": "title is required"}
+
+    def test_body_and_url_mutually_exclusive(self):
+        tool = _writable_tool()
+        result = tool.submit_post(subreddit="test", title="T", body="B", url="http://x")
+        assert "not both" in result["error"]
+
+    def test_selftext_success(self):
+        tool = _writable_tool()
+        submission = MagicMock()
+        submission.id = "p1"
+        submission.permalink = "/r/test/comments/p1/"
+        sub = MagicMock()
+        sub.submit.return_value = submission
+        reddit = MagicMock()
+        reddit.subreddit.return_value = sub
+        with patch.object(RedditTool, "reddit", new=reddit):
+            result = tool.submit_post(subreddit="test", title="T", body="B")
+        sub.submit.assert_called_once_with("T", selftext="B")
+        assert result["id"] == "p1"
+
+    def test_link_success(self):
+        tool = _writable_tool()
+        submission = MagicMock()
+        submission.id = "p2"
+        submission.permalink = "/r/test/comments/p2/"
+        sub = MagicMock()
+        sub.submit.return_value = submission
+        reddit = MagicMock()
+        reddit.subreddit.return_value = sub
+        with patch.object(RedditTool, "reddit", new=reddit):
+            result = tool.submit_post(subreddit="test", title="T", url="http://x")
+        sub.submit.assert_called_once_with("T", url="http://x")
+        assert result["id"] == "p2"
+
+
+# ── run() dispatcher ────────────────────────────────────────────────
+
+
+class TestRunDispatcher:
+    def test_unknown_action(self):
+        tool = RedditTool(client_id="c", client_secret="s")
+        assert tool.run(action="bogus") == {"error": "Unknown action: bogus"}
+
+    def test_routes_post_comment(self):
+        tool = RedditTool(client_id="c", client_secret="s")
+        with patch.object(tool, "post_comment", return_value={"ok": True}) as m:
+            tool.run(action="post_comment", post_id="abc", body="hi")
+        m.assert_called_once_with(post_id="abc", body="hi")
+
+    def test_routes_submit_post(self):
+        tool = RedditTool(client_id="c", client_secret="s")
+        with patch.object(tool, "submit_post", return_value={"ok": True}) as m:
+            tool.run(action="submit-post", subreddit="test", title="T", body="B")
+        m.assert_called_once_with(subreddit="test", title="T", body="B", url=None)