95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
 
|
"""Mail from the address book is never filed as spam (cross-plugin, §C.4).
|
||
|
|
|
||
|
|
Exercises the real lookup against the contacts plugin's own table, which is what
|
||
|
|
the hook consults. Requires both plugins installed.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import uuid
|
||
|
|
from collections.abc import AsyncIterator
|
||
|
|
|
||
|
|
import pytest_asyncio
|
||
|
|
from sqlalchemy import delete, text
|
||
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
|
|
from sqlalchemy.pool import NullPool
|
||
|
|
from webmail_plugin_antispam.policy import sender_in_contacts
|
||
|
|
|
||
|
|
from app.config import get_settings
|
||
|
|
from app.identity.passwords import hash_password
|
||
|
|
from app.models import User
|
||
|
|
from app.plugins.sdk import sdk
|
||
|
|
|
||
|
|
_SDK = sdk("antispam")
|
||
|
|
|
||
|
|
|
||
|
|
@pytest_asyncio.fixture
|
||
|
|
async def user_with_contact() -> AsyncIterator[uuid.UUID]:
|
||
|
|
engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
|
||
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
||
|
|
async with sm() as s:
|
||
|
|
user = User(
|
||
|
|
handle=f"ac{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw")
|
||
|
|
)
|
||
|
|
s.add(user)
|
||
|
|
await s.commit()
|
||
|
|
user_id = user.id
|
||
|
|
|
||
|
|
async with _SDK.session(user_id) as s:
|
||
|
|
book_id = (
|
||
|
|
await s.execute(
|
||
|
|
text(
|
||
|
|
"INSERT INTO plg_contacts_books (user_id, name) VALUES (:u, 'Test') "
|
||
|
|
"RETURNING id"
|
||
|
|
),
|
||
|
|
{"u": str(user_id)},
|
||
|
|
)
|
||
|
|
).scalar()
|
||
|
|
await s.execute(
|
||
|
|
text(
|
||
|
|
"INSERT INTO plg_contacts_contacts "
|
||
|
|
" (user_id, book_id, vcard_uid, name_display, default_email, emails) "
|
||
|
|
"VALUES (:u, :b, :v, 'A Friend', 'Friend@Example.COM', "
|
||
|
|
" CAST(:e AS jsonb))"
|
||
|
|
),
|
||
|
|
{
|
||
|
|
"u": str(user_id),
|
||
|
|
"b": str(book_id),
|
||
|
|
"v": str(uuid.uuid4()),
|
||
|
|
"e": '[{"value": "Other.Address@Example.com", "type": "work"}]',
|
||
|
|
},
|
||
|
|
)
|
||
|
|
await s.commit()
|
||
|
|
try:
|
||
|
|
yield user_id
|
||
|
|
finally:
|
||
|
|
async with sm() as s:
|
||
|
|
await s.execute(
|
||
|
|
text("DELETE FROM plg_contacts_books WHERE user_id = :u"), {"u": str(user_id)}
|
||
|
|
)
|
||
|
|
await s.execute(delete(User).where(User.id == user_id))
|
||
|
|
await s.commit()
|
||
|
|
await engine.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
async def test_primary_address_matches_case_insensitively(user_with_contact: uuid.UUID) -> None:
|
||
|
|
"""Contacts store what the user typed, so the comparison cannot be exact."""
|
||
|
|
for probe in ("friend@example.com", "Friend@Example.COM", " FRIEND@EXAMPLE.com "):
|
||
|
|
assert await sender_in_contacts(user_with_contact, probe) is True
|
||
|
|
|
||
|
|
|
||
|
|
async def test_secondary_addresses_in_the_emails_array_match(
|
||
|
|
user_with_contact: uuid.UUID,
|
||
|
|
) -> None:
|
||
|
|
assert await sender_in_contacts(user_with_contact, "other.address@example.com") is True
|
||
|
|
|
||
|
|
|
||
|
|
async def test_a_stranger_does_not_match(user_with_contact: uuid.UUID) -> None:
|
||
|
|
assert await sender_in_contacts(user_with_contact, "nobody@example.com") is False
|
||
|
|
assert await sender_in_contacts(user_with_contact, "") is False
|
||
|
|
|
||
|
|
|
||
|
|
async def test_another_users_contacts_do_not_count(user_with_contact: uuid.UUID) -> None:
|
||
|
|
"""RLS: one user's address book must not shield another user's spam."""
|
||
|
|
assert await sender_in_contacts(uuid.uuid4(), "friend@example.com") is False
|