copilot-swe-agent[bot]
84117abd13
Agent-Logs-Url: https://github.com/christianlouis/dmarq/sessions/07b58025-2aeb-4e81-a168-7a6fccdc3569 Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
501 lines
18 KiB
Python
501 lines
18 KiB
Python
"""
|
|
Tests for MailSource model and mail-sources API endpoints.
|
|
"""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.mail_source import MailSource
|
|
|
|
|
|
class TestMailSourceModel:
|
|
"""Unit tests for the MailSource ORM model."""
|
|
|
|
def test_create_mail_source(self, db_session: Session):
|
|
source = MailSource(
|
|
name="Test IMAP",
|
|
method="IMAP",
|
|
server="imap.example.com",
|
|
port=993,
|
|
username="user@example.com",
|
|
password="secret",
|
|
use_ssl=True,
|
|
folder="INBOX",
|
|
polling_interval=60,
|
|
enabled=True,
|
|
)
|
|
db_session.add(source)
|
|
db_session.commit()
|
|
db_session.refresh(source)
|
|
|
|
assert source.id is not None
|
|
assert source.name == "Test IMAP"
|
|
assert source.method == "IMAP"
|
|
assert source.server == "imap.example.com"
|
|
assert source.port == 993
|
|
assert source.username == "user@example.com"
|
|
assert source.password == "secret"
|
|
assert source.use_ssl is True
|
|
assert source.folder == "INBOX"
|
|
assert source.polling_interval == 60
|
|
assert source.enabled is True
|
|
assert source.last_checked is None
|
|
|
|
def test_default_values(self, db_session: Session):
|
|
source = MailSource(name="Minimal", method="IMAP")
|
|
db_session.add(source)
|
|
db_session.commit()
|
|
db_session.refresh(source)
|
|
|
|
assert source.folder == "INBOX" # model default
|
|
assert source.enabled is True
|
|
assert source.last_checked is None
|
|
|
|
def test_repr(self, db_session: Session):
|
|
source = MailSource(name="Demo", method="POP3")
|
|
db_session.add(source)
|
|
db_session.commit()
|
|
db_session.refresh(source)
|
|
|
|
rep = repr(source)
|
|
assert "Demo" in rep
|
|
assert "POP3" in rep
|
|
|
|
def test_multiple_sources(self, db_session: Session):
|
|
for i in range(3):
|
|
db_session.add(MailSource(name=f"Source {i}", method="IMAP"))
|
|
db_session.commit()
|
|
|
|
all_sources = db_session.query(MailSource).all()
|
|
assert len(all_sources) == 3
|
|
|
|
|
|
class TestMailSourcesAPI:
|
|
"""Integration tests for /api/v1/mail-sources endpoints (no auth)."""
|
|
|
|
def test_list_requires_auth(self, client: TestClient):
|
|
resp = client.get("/api/v1/mail-sources")
|
|
# Without auth, expect 401 or 403
|
|
assert resp.status_code in (401, 403)
|
|
|
|
def test_create_requires_auth(self, client: TestClient):
|
|
resp = client.post("/api/v1/mail-sources", json={"name": "x", "method": "IMAP"})
|
|
assert resp.status_code in (401, 403)
|
|
|
|
def test_model_create_and_list(self, client: TestClient, db_session: Session):
|
|
"""Create a mail source directly in DB and verify it's retrievable."""
|
|
source = MailSource(
|
|
name="Direct DB Source",
|
|
method="IMAP",
|
|
server="imap.example.com",
|
|
port=993,
|
|
username="user@example.com",
|
|
password="secret",
|
|
use_ssl=True,
|
|
folder="INBOX",
|
|
polling_interval=60,
|
|
enabled=True,
|
|
)
|
|
db_session.add(source)
|
|
db_session.commit()
|
|
db_session.refresh(source)
|
|
|
|
assert source.id is not None
|
|
fetched = db_session.query(MailSource).filter_by(name="Direct DB Source").first()
|
|
assert fetched is not None
|
|
assert fetched.server == "imap.example.com"
|
|
|
|
def test_toggle_enabled(self, db_session: Session):
|
|
source = MailSource(name="Toggle Test", method="IMAP", enabled=True)
|
|
db_session.add(source)
|
|
db_session.commit()
|
|
db_session.refresh(source)
|
|
|
|
# Simulate toggle
|
|
source.enabled = not source.enabled
|
|
db_session.commit()
|
|
db_session.refresh(source)
|
|
|
|
assert source.enabled is False
|
|
|
|
source.enabled = not source.enabled
|
|
db_session.commit()
|
|
db_session.refresh(source)
|
|
|
|
assert source.enabled is True
|
|
|
|
def test_delete_source(self, db_session: Session):
|
|
source = MailSource(name="To Delete", method="IMAP")
|
|
db_session.add(source)
|
|
db_session.commit()
|
|
sid = source.id
|
|
|
|
db_session.delete(source)
|
|
db_session.commit()
|
|
|
|
fetched = db_session.query(MailSource).filter_by(id=sid).first()
|
|
assert fetched is None
|
|
|
|
def test_query_enabled_sources(self, db_session: Session):
|
|
db_session.add(MailSource(name="Enabled A", method="IMAP", enabled=True))
|
|
db_session.add(MailSource(name="Enabled B", method="IMAP", enabled=True))
|
|
db_session.add(MailSource(name="Disabled", method="IMAP", enabled=False))
|
|
db_session.commit()
|
|
|
|
enabled = db_session.query(MailSource).filter(MailSource.enabled).all()
|
|
assert len(enabled) == 2
|
|
names = {s.name for s in enabled}
|
|
assert "Enabled A" in names
|
|
assert "Enabled B" in names
|
|
assert "Disabled" not in names
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Authenticated HTTP API tests (uses authed_client fixture from conftest)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMailSourcesAPIAuthed:
|
|
"""HTTP-level tests using the authed_client fixture (auth dependency bypassed)."""
|
|
|
|
# ------------------------------------------------------------------
|
|
# List
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_list_empty(self, authed_client: TestClient):
|
|
resp = authed_client.get("/api/v1/mail-sources")
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
# ------------------------------------------------------------------
|
|
# Create
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_create_imap_source(self, authed_client: TestClient):
|
|
payload = {
|
|
"name": "My IMAP",
|
|
"method": "IMAP",
|
|
"server": "imap.example.com",
|
|
"port": 993,
|
|
"username": "user@example.com",
|
|
"password": "s3cr3t",
|
|
"use_ssl": True,
|
|
"folder": "INBOX",
|
|
"polling_interval": 60,
|
|
"enabled": True,
|
|
}
|
|
resp = authed_client.post("/api/v1/mail-sources", json=payload)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["name"] == "My IMAP"
|
|
assert data["method"] == "IMAP"
|
|
assert data["server"] == "imap.example.com"
|
|
assert data["password"] == "**redacted**"
|
|
assert data["id"] is not None
|
|
|
|
def test_create_normalizes_method_to_uppercase(self, authed_client: TestClient):
|
|
payload = {"name": "lowercase method", "method": "imap"}
|
|
resp = authed_client.post("/api/v1/mail-sources", json=payload)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["method"] == "IMAP"
|
|
|
|
# ------------------------------------------------------------------
|
|
# Get single
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_get_existing_source(self, authed_client: TestClient):
|
|
create_resp = authed_client.post(
|
|
"/api/v1/mail-sources", json={"name": "Get Test", "method": "IMAP"}
|
|
)
|
|
source_id = create_resp.json()["id"]
|
|
|
|
resp = authed_client.get(f"/api/v1/mail-sources/{source_id}")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["id"] == source_id
|
|
assert resp.json()["name"] == "Get Test"
|
|
|
|
def test_get_nonexistent_source_returns_404(self, authed_client: TestClient):
|
|
resp = authed_client.get("/api/v1/mail-sources/99999")
|
|
assert resp.status_code == 404
|
|
|
|
# ------------------------------------------------------------------
|
|
# Update
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_update_name(self, authed_client: TestClient):
|
|
create_resp = authed_client.post(
|
|
"/api/v1/mail-sources", json={"name": "Original", "method": "IMAP"}
|
|
)
|
|
source_id = create_resp.json()["id"]
|
|
|
|
update_resp = authed_client.put(
|
|
f"/api/v1/mail-sources/{source_id}", json={"name": "Updated"}
|
|
)
|
|
assert update_resp.status_code == 200
|
|
assert update_resp.json()["name"] == "Updated"
|
|
|
|
def test_update_method_normalizes_uppercase(self, authed_client: TestClient):
|
|
create_resp = authed_client.post(
|
|
"/api/v1/mail-sources", json={"name": "MethodTest", "method": "IMAP"}
|
|
)
|
|
source_id = create_resp.json()["id"]
|
|
|
|
update_resp = authed_client.put(
|
|
f"/api/v1/mail-sources/{source_id}", json={"method": "pop3"}
|
|
)
|
|
assert update_resp.status_code == 200
|
|
assert update_resp.json()["method"] == "POP3"
|
|
|
|
def test_update_nonexistent_source_returns_404(self, authed_client: TestClient):
|
|
resp = authed_client.put("/api/v1/mail-sources/99999", json={"name": "x"})
|
|
assert resp.status_code == 404
|
|
|
|
# ------------------------------------------------------------------
|
|
# Delete
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_delete_source(self, authed_client: TestClient):
|
|
create_resp = authed_client.post(
|
|
"/api/v1/mail-sources", json={"name": "Delete Me", "method": "IMAP"}
|
|
)
|
|
source_id = create_resp.json()["id"]
|
|
|
|
del_resp = authed_client.delete(f"/api/v1/mail-sources/{source_id}")
|
|
assert del_resp.status_code == 204
|
|
|
|
# Verify gone
|
|
get_resp = authed_client.get(f"/api/v1/mail-sources/{source_id}")
|
|
assert get_resp.status_code == 404
|
|
|
|
def test_delete_nonexistent_source_returns_404(self, authed_client: TestClient):
|
|
resp = authed_client.delete("/api/v1/mail-sources/99999")
|
|
assert resp.status_code == 404
|
|
|
|
# ------------------------------------------------------------------
|
|
# List after creates
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_list_multiple_sources(self, authed_client: TestClient):
|
|
for i in range(3):
|
|
authed_client.post(
|
|
"/api/v1/mail-sources", json={"name": f"Source {i}", "method": "IMAP"}
|
|
)
|
|
resp = authed_client.get("/api/v1/mail-sources")
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()) == 3
|
|
|
|
# ------------------------------------------------------------------
|
|
# Toggle
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_toggle_disables_then_enables(self, authed_client: TestClient):
|
|
create_resp = authed_client.post(
|
|
"/api/v1/mail-sources", json={"name": "Toggle", "method": "IMAP", "enabled": True}
|
|
)
|
|
source_id = create_resp.json()["id"]
|
|
|
|
toggle_resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/toggle")
|
|
assert toggle_resp.status_code == 200
|
|
assert toggle_resp.json()["enabled"] is False
|
|
|
|
toggle_resp2 = authed_client.post(f"/api/v1/mail-sources/{source_id}/toggle")
|
|
assert toggle_resp2.status_code == 200
|
|
assert toggle_resp2.json()["enabled"] is True
|
|
|
|
def test_toggle_nonexistent_returns_404(self, authed_client: TestClient):
|
|
resp = authed_client.post("/api/v1/mail-sources/99999/toggle")
|
|
assert resp.status_code == 404
|
|
|
|
# ------------------------------------------------------------------
|
|
# Test stored source
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_test_stored_imap_source_success(self, authed_client: TestClient):
|
|
create_resp = authed_client.post(
|
|
"/api/v1/mail-sources",
|
|
json={
|
|
"name": "IMAP Test",
|
|
"method": "IMAP",
|
|
"server": "imap.example.com",
|
|
"username": "u",
|
|
"password": "p",
|
|
},
|
|
)
|
|
source_id = create_resp.json()["id"]
|
|
|
|
mock_stats = {
|
|
"message_count": 10,
|
|
"unread_count": 2,
|
|
"dmarc_count": 1,
|
|
"available_mailboxes": ["INBOX"],
|
|
}
|
|
mock_client = MagicMock()
|
|
mock_client.test_connection.return_value = (True, "Connection successful", mock_stats)
|
|
|
|
with patch("app.api.api_v1.endpoints.mail_sources.IMAPClient", return_value=mock_client):
|
|
resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/test")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["success"] is True
|
|
assert data["message"] == "Connection successful"
|
|
assert data["message_count"] == 10
|
|
|
|
def test_test_stored_imap_source_failure(self, authed_client: TestClient):
|
|
create_resp = authed_client.post(
|
|
"/api/v1/mail-sources",
|
|
json={"name": "IMAP Fail", "method": "IMAP", "server": "bad.host"},
|
|
)
|
|
source_id = create_resp.json()["id"]
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.test_connection.return_value = (
|
|
False,
|
|
"Connection failed. Check server address and credentials.",
|
|
{},
|
|
)
|
|
|
|
with patch("app.api.api_v1.endpoints.mail_sources.IMAPClient", return_value=mock_client):
|
|
resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/test")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["success"] is False
|
|
|
|
def test_test_stored_non_imap_source(self, authed_client: TestClient):
|
|
create_resp = authed_client.post(
|
|
"/api/v1/mail-sources", json={"name": "POP3 Source", "method": "POP3"}
|
|
)
|
|
source_id = create_resp.json()["id"]
|
|
|
|
resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/test")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["success"] is False
|
|
assert "not yet implemented" in data["message"]
|
|
|
|
def test_test_stored_nonexistent_returns_404(self, authed_client: TestClient):
|
|
resp = authed_client.post("/api/v1/mail-sources/99999/test")
|
|
assert resp.status_code == 404
|
|
|
|
# ------------------------------------------------------------------
|
|
# Ad-hoc test connection
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_adhoc_imap_success(self, authed_client: TestClient):
|
|
payload = {
|
|
"server": "imap.example.com",
|
|
"port": 993,
|
|
"username": "user@example.com",
|
|
"password": "secret",
|
|
"ssl": True,
|
|
"method": "IMAP",
|
|
}
|
|
mock_stats = {
|
|
"message_count": 5,
|
|
"unread_count": 1,
|
|
"dmarc_count": 0,
|
|
"available_mailboxes": ["INBOX"],
|
|
}
|
|
mock_client = MagicMock()
|
|
mock_client.test_connection.return_value = (True, "Connection successful", mock_stats)
|
|
|
|
with patch("app.api.api_v1.endpoints.mail_sources.IMAPClient", return_value=mock_client):
|
|
resp = authed_client.post("/api/v1/mail-sources/test-connection", json=payload)
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["success"] is True
|
|
assert data["message_count"] == 5
|
|
|
|
def test_adhoc_non_imap_returns_not_implemented(self, authed_client: TestClient):
|
|
payload = {
|
|
"server": "pop3.example.com",
|
|
"port": 110,
|
|
"username": "u",
|
|
"password": "p",
|
|
"ssl": False,
|
|
"method": "POP3",
|
|
}
|
|
resp = authed_client.post("/api/v1/mail-sources/test-connection", json=payload)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["success"] is False
|
|
assert "not yet implemented" in data["message"]
|
|
|
|
def test_adhoc_gmail_api_returns_not_implemented(self, authed_client: TestClient):
|
|
payload = {"method": "GMAIL_API"}
|
|
resp = authed_client.post("/api/v1/mail-sources/test-connection", json=payload)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["success"] is False
|
|
assert "not yet implemented" in resp.json()["message"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _sanitize_for_log helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSanitizeForLog:
|
|
"""Unit tests for the _sanitize_for_log helper."""
|
|
|
|
def test_strips_newline(self):
|
|
from app.api.api_v1.endpoints.mail_sources import _sanitize_for_log
|
|
|
|
assert "\n" not in _sanitize_for_log("hello\nworld")
|
|
|
|
def test_strips_carriage_return(self):
|
|
from app.api.api_v1.endpoints.mail_sources import _sanitize_for_log
|
|
|
|
assert "\r" not in _sanitize_for_log("foo\rbar")
|
|
|
|
def test_integer_is_safe(self):
|
|
from app.api.api_v1.endpoints.mail_sources import _sanitize_for_log
|
|
|
|
assert _sanitize_for_log(42) == "42"
|
|
|
|
def test_normal_string_unchanged(self):
|
|
from app.api.api_v1.endpoints.mail_sources import _sanitize_for_log
|
|
|
|
assert _sanitize_for_log("example.com") == "example.com"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Source-to-response helper (password masking)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSourceToResponse:
|
|
"""Tests for the _source_to_response password-masking helper."""
|
|
|
|
def test_password_is_redacted_when_set(self, db_session: Session):
|
|
from app.api.api_v1.endpoints.mail_sources import _source_to_response
|
|
|
|
source = MailSource(name="Redact Test", method="IMAP", password="plaintext")
|
|
db_session.add(source)
|
|
db_session.commit()
|
|
db_session.refresh(source)
|
|
|
|
response = _source_to_response(source)
|
|
assert response.password == "**redacted**"
|
|
|
|
def test_password_is_none_when_not_set(self, db_session: Session):
|
|
from app.api.api_v1.endpoints.mail_sources import _source_to_response
|
|
|
|
source = MailSource(name="No Password", method="IMAP")
|
|
db_session.add(source)
|
|
db_session.commit()
|
|
db_session.refresh(source)
|
|
|
|
response = _source_to_response(source)
|
|
assert response.password is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pytest marker to avoid warnings for test methods without assertions
|
|
# ---------------------------------------------------------------------------
|
|
pytestmark = pytest.mark.usefixtures("_reset_report_store")
|