Issue/14 & issue/15 added tests to the dispatcher service #patch

.github/workflows/ci.yaml

@@ -29,9 +29,28 @@ jobs:
           TAG_CONTEXT: branch
           PRERELEASE: ${{ github.ref != 'refs/heads/main' }}
 
+  test:
+    runs-on: ubuntu-latest
+    env:
+      REDDIT_CLIENT_ID: ${{ secrets.REDDIT_CLIENT_ID }}
+      REDDIT_CLIENT_SECRET: ${{ secrets.REDDIT_CLIENT_SECRET }}
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.11"
+      - name: Install dependencies
+        run: pip install -r tests/requirements.txt
+      - name: Run linter
+        run: flake8
+      - name: Run tests
+        run: pytest
+
   build:
     needs:
       - version
+      - test
     permissions:
       packages: write
       contents: read

Makefile

@@ -0,0 +1,31 @@
+.DEFAULT_GOAL := help
+include .env
+export
+
+build: ## Build the image with a dev tag
+	@docker build -t dispatcher:dev .
+
+run: build ## Run the dispatcher with the default config
+	@docker run -it --env-file .env -v `pwd`/config.json:/src/config.json dispatcher
+
+dependencies: ## Install pip depencencies
+	@echo "..... Installing depencencies"
+	@pip install -r tests/requirements.txt --quiet
+
+lint: dependencies ## Fun flake8 linter
+	@echo "..... Linting"
+	@python -m flake8
+
+test: dependencies ## Run pytest
+	@echo "..... Running tests"
+	@python -m pytest && $(MAKE) lint
+
+
+.PHONY: help
+help: ## Display this help
+	@echo "\nUsage:\n  make \033[36m<target>\033[0m"
+	@awk 'BEGIN {FS = ":.*##"}; \
+		/^[a-zA-Z0-9_-]+:.*?##/ { printf "  \033[36m%-15s\033[0m %s\n", $$1, $$2 } \
+		/^##@/ { printf "\n\033[1m%s\033[0m\n", substr($$0, 5) } \
+		/^###@/ { printf "\n\033[1m%s\033[0m\n", substr($$0, 5) } \
+		/^###&/ { printf "\t\t  \033[33m%s\033[0m\n", substr($$0, 5) } ' $(MAKEFILE_LIST)

README.md

@@ -4,7 +4,7 @@
 ![GitHub tag (latest SemVer)](https://img.shields.io/github/v/tag/flam-flam/dispatcher-service?logo=Github&sort=semver&style=for-the-badge)
 
 Small service that streams reddit submissions and comments
-to respective endpoints
+to respective endpoints.
 
 ## Reddit credentials
 
@@ -24,20 +24,36 @@ to respective endpoints
 
 ## Local dev / docker
 
+Requires python>=3.9 or docker.
+
 Build and run using the environment variables in `.env` file
 and `config.json`:
 
 ```sh
-docker build -t dispatcher . && docker run -it --env-file .env -v $(pwd)/config.json:/src/config.json dispatcher
+make build
+```
+
+Run the image with
+
+```sh
+make run
 ```
 
+>Note: there are issues with running this on Windows
+>(see [this PR discussion](https://github.com/flam-flam/dispatcher-service/pull/17#issuecomment-1481356643)),
+>but it should be okay running in the docker container.
+
 >Note: if you're running the code outside the docker container,
 >you need to set `CONFIG_PATH` environment variable to your `config.json` path.
 
+### Tests
+
+Run the tests with `make test`.
+
 ## Data output
 
 The code sends a POST request to endpoints in `config.json` with
-JSON payload, same for both comments and posts, e.g.:
+JSON payload, same for both comments and submissions, e.g.:
 
 ```json
 {"id": "j643al2", "created_utc": "2023-03-10T13:12:18+00:00"}

app/__init__.py

@@ -0,0 +1,3 @@
+from .dispatcher import RedditDispatcher
+
+__all__ = ["RedditDispatcher"]

app/__main__.py

@@ -1,6 +1,7 @@
 import os
 import json
 import logging
+import asyncpraw
 from .dispatcher import RedditDispatcher
 
 logger = logging.getLogger("__main__")
@@ -10,13 +11,13 @@
     with open(config_path, "r") as config_file:
         config = json.load(config_file)
 
-    config["api_config"] = dict(
+    reddit_instance = asyncpraw.Reddit(
         client_id=os.environ.get("REDDIT_CLIENT_ID"),
         client_secret=os.environ.get("REDDIT_CLIENT_SECRET"),
-        user_agent=f"python:flam-flam-dispatcher-service (by /u/timberhilly)",
+        user_agent="python:flam-flam-dispatcher-service (by /u/timberhilly)",
         redirect_uri="http://flam-flam.github.io"
     )
 
-    RedditDispatcher(**config).start()
+    RedditDispatcher(reddit_instance, config).start()
 except Exception as e:
     logger.error(e)

app/dispatcher.py

@@ -1,15 +1,10 @@
-import sys
 import json
-import socket
 import logging
 import requests
 import time
-import asyncpraw
 import asyncio
-from datetime import datetime as dt
-
-from asyncpraw.models.reddit.submission import Submission
-from asyncpraw.models.reddit.comment import Comment
+import asyncpraw
+from datetime import datetime
 
 
 class RedditDispatcher:
@@ -18,20 +13,24 @@ class RedditDispatcher:
     sends it to the consumer endpoints.
     """
 
-    def __init__(self, **kwargs):
-        self._setup_logging(kwargs.get("debug", False))
-        api_config = kwargs.get("api_config", dict())
+    def __init__(self, reddit, config):
+        self.reddit = reddit
+        self._setup_logging(config.get("config", False))
 
-        self.submission_endpoint = kwargs.get("submission_endpoint", "http://localhost:8080")
-        self.logger.info(f"Set submission endpoint to {self.submission_endpoint}")
+        self.submission_endpoint = \
+            config.get("submission_endpoint", "http://localhost:8080")
+        self.logger.info(
+            f"Set submission endpoint to {self.submission_endpoint}")
 
-        self.comment_endpoint = kwargs.get("comment_endpoint", "http://localhost:8080")
-        self.logger.info(f"Set comment endpoint to {self.comment_endpoint}")
+        self.comment_endpoint = \
+            config.get("comment_endpoint", "http://localhost:8080")
+        self.logger.info(
+            f"Set comment endpoint to {self.comment_endpoint}")
 
-        self.subreddits = kwargs.get("subreddits", [])
+        self._subreddit_object = None
+        self.subreddits = config.get("subreddits", [])
         self.logger.info(f"Watching subreddits {self.subreddits}")
 
-        self.reddit = asyncpraw.Reddit(**api_config)
         self.headers = {
             'Content-type': 'application/json',
             'Accept': 'application/json'
@@ -53,42 +52,53 @@ def start(self) -> None:
         self.logger.info("Started reddit dispatcher")
         loop.run_forever()
 
+    async def _get_subreddit(self) -> "asyncpraw.models.Subreddit":
+        """Get the asyncpraw.models.Subreddit object.
+        Only the first call will fetch data from Reddit API.
+        """
+        subreddit_string = "+".join(self.subreddits)
+        if self._subreddit_object is None:
+            self.logger.info(
+                f"Getting the subreddit object: {subreddit_string}")
+            self._subreddit_object = \
+                await self.reddit.subreddit(subreddit_string)
+        return self._subreddit_object
+
     async def _stream_submissions(self) -> None:
         """Stream submissions asynchronously. Calls self._dispatch_submission()
         """
-        subreddit = await self.reddit.subreddit(
-            "+".join(self.subreddits))
+        subreddit = await self._get_subreddit()
         async for submission in subreddit.stream.submissions(pause_after=-1):
             if submission is None:
                 continue
             await self._dispatch(self.submission_endpoint, dict(
                 id=submission.id,
-                created_utc=str(dt.fromtimestamp(submission.created_utc))
+                created_utc=str(datetime.fromtimestamp(submission.created_utc))
             ))
 
     async def _stream_comments(self) -> None:
         """Stream comments asynchronously. Calls self._dispatch_comment()
         """
-        subreddit = await self.reddit.subreddit(
-            "+".join(self.subreddits))
+        subreddit = await self._get_subreddit()
         async for comment in subreddit.stream.comments(pause_after=-1):
             if comment is None:
                 continue
             await self._dispatch(self.comment_endpoint, dict(
                 id=comment.id,
-                created_utc=str(dt.fromtimestamp(comment.created_utc))
+                created_utc=str(datetime.fromtimestamp(comment.created_utc))
             ))
 
     async def _dispatch(self, endpoint: str, data: dict) -> None:
         """POST a json data object to an endpoint
         """
         try:
-            response = requests.post(endpoint,
-                data=json.dumps(data),
-                headers=self.headers
-            ).raise_for_status()
+            requests.post(endpoint,
+                          data=json.dumps(data),
+                          headers=self.headers
+                          ).raise_for_status()
             self.logger.debug(f"Dispatched comment with ID={data.get('id')}")
         except Exception as e:
             self.logger.error(f"Failed to dispatch to {endpoint}, {e}")
-            self.logger.error(f"DATA: {json.dumps(data)} HEADERS: {self.headers}")
-            time.sleep(1) # wait for a bit to not flood the logs
+            self.logger.error(
+                f"DATA: {json.dumps(data)} HEADERS: {self.headers}")
+            time.sleep(1)  # wait for a bit to not flood the logs

requirements.txt

@@ -1 +1 @@
-asyncpraw>=7.7.0
+asyncpraw>=7.7.0

tests/helpers/__init__.py

@@ -0,0 +1,3 @@
+from .subreddit_stream import SubredditStream
+
+__all__ = ["SubredditStream"]

tests/helpers/subreddit_stream.py

@@ -0,0 +1,29 @@
+import asyncio
+from unittest.mock import AsyncMock
+
+
+class SubredditStream(AsyncMock):
+    fake_comments = [
+        type('Comment', (object,), dict(
+                id=f"abcde{i}",
+                created_utc=i+1681654036
+            ))
+        for i in range(3)
+    ]
+    fake_submissions = [
+        type('Submission', (object,), dict(
+                id=f"abcde{i}",
+                created_utc=i+1681654036
+            ))
+        for i in range(3)
+    ]
+
+    async def submissions(self, pause_after=0):
+        for i in range(3):
+            yield self.fake_submissions[i]
+            await asyncio.sleep(0.1)
+
+    async def comments(self, pause_after=0):
+        for i in range(3):
+            yield self.fake_comments[i]
+            await asyncio.sleep(0.1)

tests/requirements.txt

@@ -0,0 +1,5 @@
+-r ../requirements.txt
+pytest>=7.2.0
+requests-mock>=1.10.0
+pytest-asyncio>=0.21.0
+flake8>=6.0.0