184 lines
6 KiB
Python
184 lines
6 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from unittest.mock import MagicMock, patch
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
import respx
|
||
|
|
import httpx
|
||
|
|
|
||
|
|
from howl.graph.client import GraphClient, _parse_message, _extract_skip_token, _strip_html
|
||
|
|
|
||
|
|
|
||
|
|
GRAPH_BASE = "https://graph.microsoft.com/v1.0"
|
||
|
|
|
||
|
|
|
||
|
|
def make_graph_client() -> GraphClient:
|
||
|
|
return GraphClient(token_provider=lambda: "test-bearer-token")
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Unit tests (no HTTP)
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
def test_parse_message_extracts_fields(sample_emails):
|
||
|
|
raw = sample_emails[0] # customer inquiry
|
||
|
|
msg = _parse_message(raw)
|
||
|
|
assert msg.id == "msg-001-customer-inquiry"
|
||
|
|
assert msg.sender_address == "alice@acme.com"
|
||
|
|
assert msg.sender_name == "Alice Johnson"
|
||
|
|
assert msg.subject == "Question about my order #45231"
|
||
|
|
assert msg.has_attachments is False
|
||
|
|
assert isinstance(msg.received_at, datetime)
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_message_with_no_sender():
|
||
|
|
raw = {"id": "x", "conversationId": None}
|
||
|
|
msg = _parse_message(raw)
|
||
|
|
assert msg.sender_address == ""
|
||
|
|
assert msg.received_at is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_extract_skip_token_from_next_link():
|
||
|
|
link = "https://graph.microsoft.com/v1.0/users/mb/messages?$skipToken=abc123&$top=50"
|
||
|
|
assert _extract_skip_token(link) == "abc123"
|
||
|
|
|
||
|
|
|
||
|
|
def test_extract_skip_token_returns_none_when_absent():
|
||
|
|
assert _extract_skip_token("https://graph.microsoft.com/v1.0/users/mb/messages") is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_strip_html_removes_tags():
|
||
|
|
html = "<html><body><p>Hello <b>World</b></p></body></html>"
|
||
|
|
text = _strip_html(html)
|
||
|
|
assert "<" not in text
|
||
|
|
assert "Hello" in text
|
||
|
|
assert "World" in text
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# HTTP-level tests with respx
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
@respx.mock
|
||
|
|
def test_list_unread_messages_single_page():
|
||
|
|
client = make_graph_client()
|
||
|
|
mailbox = "test@example.com"
|
||
|
|
url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders/Inbox/messages"
|
||
|
|
|
||
|
|
respx.get(url).mock(return_value=httpx.Response(200, json={
|
||
|
|
"value": [
|
||
|
|
{
|
||
|
|
"id": "msg-001",
|
||
|
|
"conversationId": "conv-001",
|
||
|
|
"subject": "Test",
|
||
|
|
"sender": {"emailAddress": {"address": "a@b.com", "name": "A B"}},
|
||
|
|
"receivedDateTime": "2026-04-01T10:00:00Z",
|
||
|
|
"hasAttachments": False,
|
||
|
|
"importance": "normal",
|
||
|
|
}
|
||
|
|
]
|
||
|
|
}))
|
||
|
|
|
||
|
|
messages, next_skip = client.list_unread_messages(mailbox, top=50)
|
||
|
|
assert len(messages) == 1
|
||
|
|
assert messages[0].id == "msg-001"
|
||
|
|
assert next_skip is None
|
||
|
|
|
||
|
|
|
||
|
|
@respx.mock
|
||
|
|
def test_list_unread_messages_pagination():
|
||
|
|
client = make_graph_client()
|
||
|
|
mailbox = "test@example.com"
|
||
|
|
url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders/Inbox/messages"
|
||
|
|
|
||
|
|
respx.get(url).mock(return_value=httpx.Response(200, json={
|
||
|
|
"value": [{"id": "msg-001", "conversationId": None, "subject": "S",
|
||
|
|
"sender": {"emailAddress": {"address": "a@b.com"}},
|
||
|
|
"receivedDateTime": "2026-04-01T10:00:00Z",
|
||
|
|
"hasAttachments": False, "importance": "normal"}],
|
||
|
|
"@odata.nextLink": f"{url}?$skipToken=NEXT_TOKEN_ABC"
|
||
|
|
}))
|
||
|
|
|
||
|
|
messages, next_skip = client.list_unread_messages(mailbox, top=1)
|
||
|
|
assert len(messages) == 1
|
||
|
|
assert next_skip == "NEXT_TOKEN_ABC"
|
||
|
|
|
||
|
|
|
||
|
|
@respx.mock
|
||
|
|
def test_get_message_body_plain_text():
|
||
|
|
client = make_graph_client()
|
||
|
|
mailbox = "test@example.com"
|
||
|
|
message_id = "msg-001"
|
||
|
|
url = f"{GRAPH_BASE}/users/{mailbox}/messages/{message_id}"
|
||
|
|
|
||
|
|
respx.get(url).mock(return_value=httpx.Response(200, json={
|
||
|
|
"body": {"contentType": "text", "content": "Hello, this is the body."}
|
||
|
|
}))
|
||
|
|
|
||
|
|
body = client.get_message_body(mailbox, message_id)
|
||
|
|
assert body == "Hello, this is the body."
|
||
|
|
|
||
|
|
|
||
|
|
@respx.mock
|
||
|
|
def test_get_message_body_html_stripped():
|
||
|
|
client = make_graph_client()
|
||
|
|
mailbox = "test@example.com"
|
||
|
|
message_id = "msg-002"
|
||
|
|
url = f"{GRAPH_BASE}/users/{mailbox}/messages/{message_id}"
|
||
|
|
|
||
|
|
respx.get(url).mock(return_value=httpx.Response(200, json={
|
||
|
|
"body": {"contentType": "html", "content": "<html><body><p>Hello</p></body></html>"}
|
||
|
|
}))
|
||
|
|
|
||
|
|
body = client.get_message_body(mailbox, message_id)
|
||
|
|
assert "<" not in body
|
||
|
|
assert "Hello" in body
|
||
|
|
|
||
|
|
|
||
|
|
@respx.mock
|
||
|
|
def test_get_or_create_folder_creates_when_not_found():
|
||
|
|
client = make_graph_client()
|
||
|
|
mailbox = "test@example.com"
|
||
|
|
list_url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders"
|
||
|
|
create_url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders"
|
||
|
|
|
||
|
|
# First call: list returns empty
|
||
|
|
respx.get(list_url).mock(return_value=httpx.Response(200, json={"value": []}))
|
||
|
|
# Second call: create returns new folder
|
||
|
|
respx.post(create_url).mock(return_value=httpx.Response(201, json={
|
||
|
|
"id": "folder-id-123",
|
||
|
|
"displayName": "Customers"
|
||
|
|
}))
|
||
|
|
|
||
|
|
folder_id = client.get_or_create_folder(mailbox, "Customers")
|
||
|
|
assert folder_id == "folder-id-123"
|
||
|
|
|
||
|
|
# Second call should use cache, no additional HTTP requests
|
||
|
|
folder_id_cached = client.get_or_create_folder(mailbox, "Customers")
|
||
|
|
assert folder_id_cached == folder_id
|
||
|
|
|
||
|
|
|
||
|
|
@respx.mock
|
||
|
|
def test_move_message():
|
||
|
|
client = make_graph_client()
|
||
|
|
mailbox = "test@example.com"
|
||
|
|
message_id = "msg-001"
|
||
|
|
folder_id = "folder-abc"
|
||
|
|
url = f"{GRAPH_BASE}/users/{mailbox}/messages/{message_id}/move"
|
||
|
|
|
||
|
|
respx.post(url).mock(return_value=httpx.Response(201, json={"id": "msg-001-moved"}))
|
||
|
|
client.move_message(mailbox, message_id, folder_id) # Should not raise
|
||
|
|
|
||
|
|
|
||
|
|
@respx.mock
|
||
|
|
def test_raises_on_401():
|
||
|
|
client = make_graph_client()
|
||
|
|
mailbox = "test@example.com"
|
||
|
|
url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders/Inbox/messages"
|
||
|
|
|
||
|
|
respx.get(url).mock(return_value=httpx.Response(401, json={"error": {"code": "Unauthorized"}}))
|
||
|
|
|
||
|
|
with pytest.raises(httpx.HTTPStatusError):
|
||
|
|
client.list_unread_messages(mailbox, top=10)
|