Mini Task - mAIgic Team 3 - Gmail Communication

requirements.txt

@@ -0,0 +1,4 @@
+google-api-python-client
+google-auth
+google-auth-oauthlib
+python-dotenv

src/gmail/api.py

@@ -0,0 +1,21 @@
+from googleapiclient.discovery import build
+from src.gmail.auth import get_credentials
+from src.gmail.models import Email, Header
+
+class GmailAPI:
+    """class to actually interact with the mails"""
+    def __init__(self):
+        self.service = build('gmail', 'v1', credentials=get_credentials())
+
+    def fetch_latest_email(self):
+        results = self.service.users().messages().list(userId='me', maxResults=1).execute()
+        messages = results.get('messages', [])
+        if messages:
+            email_data = self.service.users().messages().get(userId='me', id=messages[0]['id']).execute()
+            headers = email_data.get('payload', {}).get('headers', [])
+            sender = next((header['value'] for header in headers if header['name'] == 'From'), 'Unknown Sender')
+            subject = next((header['value'] for header in headers if header['name'] == 'Subject'), 'No Subject')
+            snippet = email_data.get('snippet', 'No Snippet Available')
+            header_objects = [Header(header['name'], header['value']) for header in headers]
+            return Email(email_id=email_data['id'], sender=sender, subject=subject, snippet=snippet, headers=header_objects)
+        return None

src/gmail/auth.py

@@ -0,0 +1,29 @@
+import os
+from google.oauth2.credentials import Credentials
+from google_auth_oauthlib.flow import InstalledAppFlow
+from google.auth.transport.requests import Request
+from dotenv import load_dotenv
+
+load_dotenv()
+
+def get_credentials():
+    """gmail api authentication of credentials"""
+    SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']
+    token_path = 'token.json'
+    creds_path = os.getenv('GOOGLE_CREDENTIALS_JSON')
+
+    if not creds_path:
+        raise FileNotFoundError("Environment variable 'GOOGLE_CREDENTIALS_JSON' is not set or missing.")
+
+    creds = None
+    if os.path.exists(token_path):
+        creds = Credentials.from_authorized_user_file(token_path, scopes=SCOPES)
+    if not creds or not creds.valid:
+        if creds and creds.expired and creds.refresh_token:
+            creds.refresh(Request())
+        else:
+            flow = InstalledAppFlow.from_client_secrets_file(creds_path, SCOPES)
+            creds = flow.run_local_server(port=0)
+            with open(token_path, 'w') as token_file:
+                token_file.write(creds.to_json())
+    return creds

src/gmail/main.py

@@ -0,0 +1,21 @@
+from src.gmail.api import GmailAPI
+
+def main():
+    """basic main function to see the working of the api"""
+    try:
+        gmail_api = GmailAPI()
+        latest_email = gmail_api.fetch_latest_email()
+
+        if latest_email:
+            print("Latest Email:")
+            print(f"ID: {latest_email.email_id}")
+            print(f"From: {latest_email.sender}")
+            print(f"Subject: {latest_email.subject}")
+            print(f"Snippet: {latest_email.snippet}")
+        else:
+            print("No new emails found.")
+    except Exception as e:
+        print(f"An error occurred: {e}")
+
+if __name__ == "__main__":
+    main()

src/gmail/models.py

@@ -0,0 +1,25 @@
+from typing import List, Optional
+
+"""models for api"""
+
+class Header:
+    """Represents an email header."""
+    def __init__(self, name: str, value: str):
+        self.name = name
+        self.value = value
+
+    def __repr__(self):
+        return f"{self.name}: {self.value}"
+
+
+class Email:
+    """Represents an email message."""
+    def __init__(self, email_id: str, sender: str, subject: str, snippet: Optional[str] = None, headers: Optional[List[Header]] = None):
+        self.email_id = email_id
+        self.sender = sender
+        self.subject = subject
+        self.snippet = snippet
+        self.headers = headers or []
+
+    def __repr__(self):
+        return f"Email(id={self.email_id}, sender={self.sender}, subject={self.subject}, snippet={self.snippet})"

test/gmail/test_api.py

@@ -0,0 +1,38 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from src.gmail.api import GmailAPI
+from src.gmail.models import Email
+
+class TestGmailAPI(unittest.TestCase):
+    """testing gmail api"""
+    @patch('src.gmail.api.build')
+    @patch('src.gmail.api.get_credentials')
+    def test_fetch_latest_email(self, mock_get_credentials, mock_build):
+        service_mock = MagicMock()
+        mock_build.return_value = service_mock
+        service_mock.users().messages().list().execute.return_value = {
+            'messages': [{'id': '123'}]
+        }
+        service_mock.users().messages().get().execute.return_value = {
+            'id': '123',
+            'snippet': 'Test snippet',
+            'payload': {
+                'headers': [
+                    {'name': 'From', 'value': 'sender@example.com'},
+                    {'name': 'Subject', 'value': 'Test Subject'}
+                ]
+            }
+        }
+
+        gmail_api = GmailAPI()
+        email = gmail_api.fetch_latest_email()
+
+        self.assertIsNotNone(email, "Email should not be None")
+        self.assertIsInstance(email, Email, "The returned object should be an instance of Email")
+        self.assertEqual(email.email_id, '123', "Email ID should match the mocked response")
+        self.assertEqual(email.sender, 'sender@example.com', "Email sender should match the mocked response")
+        self.assertEqual(email.subject, 'Test Subject', "Email subject should match the mocked response")
+        self.assertEqual(email.snippet, 'Test snippet', "Email snippet should match the mocked response")
+
+if __name__ == '__main__':
+    unittest.main()

test/gmail/test_auth.py

@@ -0,0 +1,22 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from src.gmail.auth import get_credentials
+
+class TestAuthManager(unittest.TestCase):
+    """testing auth manager"""
+    @patch('src.gmail.auth.Credentials.from_authorized_user_file')
+    @patch('src.gmail.auth.InstalledAppFlow')
+    @patch('os.path.exists')
+    def test_get_credentials(self, mock_exists, mock_flow, mock_from_file):
+        mock_exists.return_value = True
+        creds_instance = MagicMock()
+        mock_from_file.return_value = creds_instance
+
+        creds = get_credentials()
+
+        self.assertIsNotNone(creds, "Credentials should not be None")
+        mock_from_file.assert_called_once()  
+        mock_flow.assert_not_called()  
+
+if __name__ == '__main__':
+    unittest.main()

test/gmail/test_live_api.py

@@ -0,0 +1,17 @@
+import unittest
+from src.gmail.api import GmailAPI
+from src.gmail.models import Email
+
+class TestRealGmailAPI(unittest.TestCase):
+    """testing gmail api using real gmail account"""
+    def test_fetch_latest_email(self):
+        gmail_api = GmailAPI()
+        email = gmail_api.fetch_latest_email()
+
+        self.assertIsNotNone(email, "No email was fetched. Check your API connection.")
+        self.assertIsInstance(email, Email, "The returned object should be an instance of Email")
+        self.assertTrue(hasattr(email, 'email_id'), "Email object should have an ID")
+        self.assertTrue(hasattr(email, 'snippet'), "Email object should have a snippet")
+
+if __name__ == '__main__':
+    unittest.main()