Add pytest scripts and requirements.txt

.github/workflows/python-app.yml

@@ -0,0 +1,39 @@
+# This workflow will install Python dependencies, run tests and lint with a single version of Python
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
+
+name: Python application
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v3
+
+    - name: Set up Python 3.10
+      uses: actions/setup-python@v3
+      with:
+        python-version: "3.10"
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+
+    - name: Test with pytest
+      run: |
+        py -m pytest py_test.py
+      env:
+        TESTING: ${{ secrets.TESTING || 'true' }}
+        BEARER_TOKEN: ${{ secrets.BEARER_TOKEN }}

.gitignore

@@ -1,2 +1,3 @@
 .env
+__pycache__/
 output/

py_test.py

@@ -0,0 +1,67 @@
+import pytest
+from unittest.mock import Mock, patch, mock_open
+from stream_listener import fetch_tweets, get_user_id
+import datetime
+import re
+import unittest
+from freezegun import freeze_time
+
+author_name = 'LotsoFandom'
+author_id = '1742141344787857408'
+tweet_id = '1777555856705667262'
+
+# Test for get_user_id function
+def test_get_user_id():
+    with patch('stream_listener.client') as mocked_client:
+        mocked_client.get_user.return_value = Mock(data=Mock(id=author_id))
+        user_id = get_user_id(author_name)
+        assert user_id == author_id
+        mocked_client.get_user.assert_called_once_with(username='LotsoFandom')
+
+# Test for fetch_tweets function
+@freeze_time("2023-04-23 01:01:01")
+def test_fetch_tweets():
+    with patch('stream_listener.client') as mocked_client, \
+         patch('builtins.open', mock_open(), create=True) as mocked_file:  # Note the create=True for broader patching
+        # Setup the mock return value for tweets
+        mocked_client.get_users_tweets.return_value = Mock(
+            data=[Mock(id=tweet_id, author_id=author_id, created_at='2023-01-01', text='Sample Tweet')]
+        )
+
+        seen_tweets = set()
+
+        # Action
+        fetch_tweets(author_id, seen_tweets, 1)
+
+        # Assert
+        assert tweet_id in seen_tweets
+        mocked_client.get_users_tweets.assert_called_once_with(
+            id=author_id,
+            max_results=1,
+            exclude='replies',
+            tweet_fields=["created_at", "text", "author_id"]
+        )
+
+        # Assert that the file was attempted to be opened with the correct parameters
+        filename = f"output/{author_id}_{tweet_id}_20230423010101.md"
+        mocked_file.assert_called_once_with(filename, 'w', encoding='utf-8')
+
+        # Check the content written to the file
+        mocked_file().write.assert_called_once_with(
+            "## Tweet by 1742141344787857408\n\n"
+            "**Tweet ID:** 1777555856705667262\n"
+            "**Created at:** 2023-01-01\n"
+            "**Text:**\n\n"
+            "Sample Tweet\n"
+        )
+
+# Test for handling too many tweets requested
+def test_fetch_tweets_too_many():
+    with patch('stream_listener.client') as mocked_client:
+        fetch_tweets(author_id, set(), 150)
+        mocked_client.get_users_tweets.assert_called_with(
+            id=author_id,
+            max_results=100,
+            exclude='replies',
+            tweet_fields=["created_at", "text", "author_id"]
+        )

requirements.txt

@@ -0,0 +1,4 @@
+tweepy
+python-dotenv
+pytest
+freezegun

stream_listener.py

@@ -18,7 +18,7 @@
 
 # Function to fetch and process tweets
 def fetch_tweets(user_id, seen_tweets, num_tweets):
-    print("Fetching tweets...")
+    print(f"Fetching tweets for the user: {user_id}")
     if num_tweets > 100:
         print("Maximum number of tweets per request is 100. Setting num_tweets to 100.")
         num_tweets = 100
@@ -63,22 +63,24 @@ def fetch_tweets(user_id, seen_tweets, num_tweets):
 
 # Get user ID from username
 def get_user_id(username):
+    print(f"Getting user ID for username: {username}")
     try:
         user = client.get_user(username=username)
         return user.data.id
     except Exception as e:
         raise ValueError(f"Failed to get user ID for username {username}: {str(e)}")
 
-# Read username from command-line arguments
-if len(sys.argv) < 2:
-    raise ValueError("Please provide the Twitter username as a command-line argument.")
-username = sys.argv[1]
-user_id = get_user_id(username)
+if not os.getenv('TESTING'):
+    # Read username from command-line arguments
+    if len(sys.argv) < 2:
+        raise ValueError("Please provide the Twitter username as a command-line argument.")
+    username = sys.argv[1]
+    user_id = get_user_id(username)
 
-# Set to track seen tweets
-seen_tweets = set()
+    # Set to track seen tweets
+    seen_tweets = set()
 
-# Main loop to periodically check for tweets every minute
-while True:
-    fetch_tweets(user_id, seen_tweets, 5)
-    time.sleep(60)  # Delay for 1 minute before fetching again
+    # Main loop to periodically check for tweets every minute
+    while True:
+        fetch_tweets(user_id, seen_tweets, 5)
+        time.sleep(60)  # Delay for 1 minute before fetching again