119 lines
3.9 KiB
Python
119 lines
3.9 KiB
Python
"""Tests for email_sync and nextcloud_sync service logic (with mocks)."""
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from email.mime.text import MIMEText
|
|
import email as email_lib
|
|
|
|
from app.services.email_sync import (
|
|
_decode_header_value,
|
|
_get_body_snippet,
|
|
LABEL_TODO,
|
|
LABEL_DONE,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# email_sync unit tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_decode_header_value_plain():
|
|
assert _decode_header_value("Hello World") == "Hello World"
|
|
|
|
|
|
def test_decode_header_value_encoded():
|
|
encoded = "=?utf-8?b?SGVsbG8gV29ybGQ=?=" # "Hello World" base64
|
|
assert _decode_header_value(encoded) == "Hello World"
|
|
|
|
|
|
def test_decode_header_value_none():
|
|
assert _decode_header_value(None) == ""
|
|
|
|
|
|
def test_get_body_snippet_plain():
|
|
msg = MIMEText("This is the email body content.")
|
|
result = _get_body_snippet(email_lib.message_from_string(msg.as_string()))
|
|
assert "This is the email body content." in result
|
|
|
|
|
|
def test_get_body_snippet_truncated():
|
|
long_body = "A" * 500
|
|
msg = MIMEText(long_body)
|
|
result = _get_body_snippet(email_lib.message_from_string(msg.as_string()), max_chars=100)
|
|
assert len(result) <= 100
|
|
|
|
|
|
def test_sync_emails_creates_task():
|
|
"""sync_emails() should create a Task for each untracked inbox email."""
|
|
from app.db.session import SessionLocal
|
|
from app.models import Task
|
|
|
|
# Build a fake raw email bytes
|
|
raw_msg = MIMEText("Task body snippet")
|
|
raw_msg["Subject"] = "Do the important thing"
|
|
raw_msg["From"] = "sender@example.com"
|
|
raw_msg["Message-ID"] = "<test-email-001@example.com>"
|
|
raw_bytes = raw_msg.as_bytes()
|
|
|
|
mock_conn = MagicMock()
|
|
mock_conn.uid.side_effect = [
|
|
# SEARCH call
|
|
("OK", [b"1"]),
|
|
# FETCH call
|
|
("OK", [(b"1 (RFC822 {123})", raw_bytes)]),
|
|
# STORE (apply label) call
|
|
("OK", []),
|
|
]
|
|
|
|
with patch("app.services.email_sync._get_imap_connection", return_value=mock_conn):
|
|
from app.services.email_sync import sync_emails
|
|
result = sync_emails()
|
|
|
|
assert result is True
|
|
|
|
db = SessionLocal()
|
|
task = db.query(Task).filter(Task.source_id == "<test-email-001@example.com>").first()
|
|
assert task is not None
|
|
assert task.title == "Do the important thing"
|
|
assert task.origin == "dhive_email"
|
|
db.delete(task)
|
|
db.commit()
|
|
db.close()
|
|
|
|
|
|
def test_mark_email_done_and_archive():
|
|
"""mark_email_done_and_archive() should swap labels and copy to Archive."""
|
|
mock_conn = MagicMock()
|
|
# SEARCH returns one UID
|
|
mock_conn.uid.side_effect = [
|
|
("OK", [b"42"]), # SEARCH
|
|
("OK", []), # STORE remove todo
|
|
("OK", []), # STORE add done
|
|
("OK", []), # COPY to archive
|
|
("OK", []), # STORE add \Deleted
|
|
]
|
|
mock_conn.expunge.return_value = ("OK", [])
|
|
|
|
with patch("app.services.email_sync._get_imap_connection", return_value=mock_conn):
|
|
from app.services.email_sync import mark_email_done_and_archive
|
|
result = mark_email_done_and_archive("<test-email-001@example.com>")
|
|
|
|
assert result is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# nextcloud_sync unit tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_update_remote_todo_not_found():
|
|
"""update_remote_todo() should return False when UID not found in any calendar."""
|
|
mock_principal = MagicMock()
|
|
mock_calendar = MagicMock()
|
|
mock_principal.calendars.return_value = [mock_calendar]
|
|
mock_calendar.todos.return_value = [] # no todos
|
|
|
|
with patch("app.services.nextcloud_sync._get_caldav_principal", return_value=mock_principal):
|
|
from app.services.nextcloud_sync import update_remote_todo
|
|
result = update_remote_todo("nonexistent-uid", {"status": "completed"})
|
|
|
|
assert result is False
|