109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
"""Tests for GmailConnector."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.models import CloudScoutConfig
|
|
from app.scouts.connectors.base import ItemRef
|
|
from app.scouts.connectors.gmail import GmailConnector
|
|
|
|
|
|
def _make_scout():
|
|
return CloudScoutConfig(
|
|
id=str(uuid.uuid4()),
|
|
user_id="00000000-0000-0000-0000-000000000003",
|
|
provider="gmail",
|
|
name="Inbox",
|
|
data_types=[],
|
|
prompt_template="",
|
|
oauth_token_encrypted="encrypted-blob",
|
|
schedule_cron="0 * * * *",
|
|
enabled=True,
|
|
auto_trash_spam=False,
|
|
device_inactivity_pause_days=14,
|
|
gmail_history_id="100",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_metadata_returns_subject_and_snippet():
|
|
scout = _make_scout()
|
|
conn = GmailConnector()
|
|
fake_message = {
|
|
"id": "msg-1",
|
|
"snippet": "preview text",
|
|
"payload": {"headers": [
|
|
{"name": "Subject", "value": "Hello"},
|
|
{"name": "From", "value": "alice@example.com"},
|
|
{"name": "Date", "value": "Wed, 14 May 2026 10:00:00 +0000"},
|
|
]},
|
|
}
|
|
with patch("app.scouts.connectors.gmail._get_gmail_service") as mock_svc:
|
|
mock_svc.return_value.users().messages().get().execute.return_value = fake_message
|
|
meta = await conn.fetch_metadata(scout, ItemRef(source_msg_ref="msg-1"))
|
|
assert meta.subject == "Hello"
|
|
assert meta.sender == "alice@example.com"
|
|
assert meta.snippet == "preview text"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_content_returns_body_text():
|
|
import base64
|
|
scout = _make_scout()
|
|
conn = GmailConnector()
|
|
body_data = base64.urlsafe_b64encode(b"hello world").decode()
|
|
fake_message = {
|
|
"id": "msg-1",
|
|
"snippet": "hello world",
|
|
"payload": {
|
|
"mimeType": "text/plain",
|
|
"headers": [
|
|
{"name": "Subject", "value": "S"},
|
|
{"name": "From", "value": "a@b"},
|
|
],
|
|
"body": {"data": body_data},
|
|
},
|
|
}
|
|
with patch("app.scouts.connectors.gmail._get_gmail_service") as mock_svc:
|
|
mock_svc.return_value.users().messages().get().execute.return_value = fake_message
|
|
content = await conn.fetch_content(scout, ItemRef(source_msg_ref="msg-1"))
|
|
assert content.body_text == "hello world"
|
|
assert content.metadata.subject == "S"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_archive_calls_trash():
|
|
scout = _make_scout()
|
|
conn = GmailConnector()
|
|
with patch("app.scouts.connectors.gmail._get_gmail_service") as mock_svc:
|
|
await conn.archive(scout, ItemRef(source_msg_ref="msg-1"))
|
|
mock_svc.return_value.users().messages().trash.assert_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_labels_returns_id_and_name():
|
|
scout = _make_scout()
|
|
conn = GmailConnector()
|
|
fake = {"labels": [
|
|
{"id": "INBOX", "name": "INBOX", "type": "system"},
|
|
{"id": "Label_1", "name": "Work", "type": "user"},
|
|
]}
|
|
with patch("app.scouts.connectors.gmail._get_gmail_service") as mock_svc:
|
|
mock_svc.return_value.users().labels().list().execute.return_value = fake
|
|
labels = await conn.list_labels(scout)
|
|
assert {"id": "INBOX", "name": "INBOX"} in labels
|
|
assert {"id": "Label_1", "name": "Work"} in labels
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_watch_calls_stop():
|
|
scout = _make_scout()
|
|
conn = GmailConnector()
|
|
with patch("app.scouts.connectors.gmail._get_gmail_service") as mock_svc:
|
|
await conn.stop_watch(scout)
|
|
mock_svc.return_value.users().stop.assert_called()
|