6d9392e5e7
Agent-Logs-Url: https://github.com/christianlouis/dmarq/sessions/f141cd3b-8d0a-4c7e-b479-ee973c209c11 Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
158 lines
5.2 KiB
Python
158 lines
5.2 KiB
Python
"""
|
|
Tests for MailSource model and mail-sources API endpoints.
|
|
"""
|
|
|
|
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."""
|
|
|
|
API_KEY_HEADER = {"X-API-Key": "test-key"} # will be set in fixture
|
|
|
|
def _get_headers(self, client: TestClient) -> dict:
|
|
"""Return auth headers using the running app's in-memory API key."""
|
|
# The test app generates its own key in-memory; we can't predict it.
|
|
# Use the TestClient without auth to check 401, and skip auth-required tests
|
|
# by injecting the dependency override instead.
|
|
return {}
|
|
|
|
def test_list_empty(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_and_list(self, client: TestClient, db_session: Session):
|
|
"""Create a mail source directly in DB and list via authenticated-less read."""
|
|
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 == True).all() # noqa: E712
|
|
)
|
|
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
|