119 lines
3.9 KiB
Python
119 lines
3.9 KiB
Python
 
|
"""`move_to_special` must describe the move it just made.
|
||
|
|
|
||
|
|
The reading panel is keyed by thread and the passive-learning hook reads the
|
||
|
|
folder pair, so an event that names neither leaves a moved message on screen and
|
||
|
|
teaches nothing.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import datetime as dt
|
||
|
|
import uuid
|
||
|
|
from collections.abc import AsyncIterator
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
import pytest_asyncio
|
||
|
|
from sqlalchemy import delete
|
||
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
|
|
from sqlalchemy.pool import NullPool
|
||
|
|
|
||
|
|
from app.config import get_settings
|
||
|
|
from app.identity.passwords import hash_password
|
||
|
|
from app.models import Account, Mailbox, Message, Thread, User
|
||
|
|
from app.plugins.actions import PluginActions
|
||
|
|
|
||
|
|
|
||
|
|
class _Ctx:
|
||
|
|
user_id: uuid.UUID
|
||
|
|
message_id: uuid.UUID
|
||
|
|
thread_id: uuid.UUID
|
||
|
|
|
||
|
|
|
||
|
|
@pytest_asyncio.fixture
|
||
|
|
async def ctx() -> AsyncIterator[_Ctx]:
|
||
|
|
engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
|
||
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
||
|
|
c = _Ctx()
|
||
|
|
async with sm() as s:
|
||
|
|
user = User(handle=f"mv{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("p"))
|
||
|
|
s.add(user)
|
||
|
|
await s.flush()
|
||
|
|
account = Account(
|
||
|
|
user_id=user.id,
|
||
|
|
email="a@b.test",
|
||
|
|
imap_config={},
|
||
|
|
smtp_config={},
|
||
|
|
secret_ct=b"x",
|
||
|
|
discovery_src="manual",
|
||
|
|
)
|
||
|
|
s.add(account)
|
||
|
|
await s.flush()
|
||
|
|
inbox = Mailbox(account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox")
|
||
|
|
junk = Mailbox(account_id=account.id, user_id=user.id, path="SPAM", special_use="junk")
|
||
|
|
thread = Thread(
|
||
|
|
user_id=user.id,
|
||
|
|
subject_norm="t",
|
||
|
|
last_message_at=dt.datetime.now(dt.UTC),
|
||
|
|
)
|
||
|
|
s.add_all([inbox, junk, thread])
|
||
|
|
await s.flush()
|
||
|
|
msg = Message(
|
||
|
|
user_id=user.id,
|
||
|
|
account_id=account.id,
|
||
|
|
mailbox_id=inbox.id,
|
||
|
|
thread_id=thread.id,
|
||
|
|
uid=1,
|
||
|
|
message_id_hdr=f"<{uuid.uuid4()}@t>",
|
||
|
|
subject="t",
|
||
|
|
from_addr=[{"email": "s@x.test"}],
|
||
|
|
to_addr=[],
|
||
|
|
flags=[],
|
||
|
|
internal_date=dt.datetime.now(dt.UTC),
|
||
|
|
blob_sha256="0" * 64,
|
||
|
|
snippet="t",
|
||
|
|
)
|
||
|
|
s.add(msg)
|
||
|
|
await s.commit()
|
||
|
|
c.user_id, c.message_id, c.thread_id = user.id, msg.id, thread.id
|
||
|
|
try:
|
||
|
|
yield c
|
||
|
|
finally:
|
||
|
|
async with sm() as s:
|
||
|
|
await s.execute(delete(User).where(User.id == c.user_id))
|
||
|
|
await s.commit()
|
||
|
|
await engine.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
async def test_move_event_names_the_thread_and_both_folders(
|
||
|
|
ctx: _Ctx, monkeypatch: pytest_asyncio.plugin.pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
published: list[tuple[str, dict[str, Any]]] = []
|
||
|
|
enqueued: list[tuple[Any, ...]] = []
|
||
|
|
|
||
|
|
class _Bus:
|
||
|
|
async def publish(self, event: str, payload: dict[str, Any]) -> None:
|
||
|
|
published.append((event, payload))
|
||
|
|
|
||
|
|
async def _enqueue(*args: Any, **kw: Any) -> None:
|
||
|
|
enqueued.append(args)
|
||
|
|
|
||
|
|
monkeypatch.setattr("app.plugins.actions.get_event_bus", lambda: _Bus())
|
||
|
|
monkeypatch.setattr("app.queue.enqueue_job", _enqueue)
|
||
|
|
|
||
|
|
moved = await PluginActions().move_to_special(ctx.user_id, ctx.message_id, "junk")
|
||
|
|
assert moved is True
|
||
|
|
|
||
|
|
assert enqueued, "the IMAP move must still be queued"
|
||
|
|
event, payload = published[0]
|
||
|
|
assert event == "message.changed"
|
||
|
|
# The panel is keyed by thread: without this it never refetches.
|
||
|
|
assert payload["thread_id"] == str(ctx.thread_id)
|
||
|
|
# The pair is the transition passive learning reads.
|
||
|
|
assert payload["previous_special_use"] == "inbox"
|
||
|
|
assert payload["special_use"] == "junk"
|
||
|
|
|
||
|
|
|
||
|
|
async def test_no_event_when_there_is_nowhere_to_move(ctx: _Ctx) -> None:
|
||
|
|
"""An account with no such folder: no move, and nothing announced."""
|
||
|
|
assert await PluginActions().move_to_special(ctx.user_id, ctx.message_id, "archive") is False
|