92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
## Copyright © 2026 Olaf Kolkman
|
|
## SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
from email.message import EmailMessage
|
|
|
|
from backend.app.database import get_connection
|
|
from backend.app.services.inbound_email_service import ingest_message, parse_incoming_email
|
|
|
|
|
|
def _raw_message(sender: str, subject: str, body: str, message_id: str) -> bytes:
|
|
message = EmailMessage()
|
|
message['From'] = sender
|
|
message['To'] = 'capture@linklog.example'
|
|
message['Subject'] = subject
|
|
message['Message-ID'] = message_id
|
|
message.set_content(body)
|
|
return message.as_bytes()
|
|
|
|
|
|
def test_parse_incoming_email_extracts_title_comment_tags_and_link():
|
|
parsed = parse_incoming_email(_raw_message(
|
|
'Alice <alice@example.com>',
|
|
'A useful article #AI',
|
|
'Interesting summary about the protocol.\nhttps://example.com/posts/1\n#Security',
|
|
'<msg-1@example.com>',
|
|
))
|
|
|
|
assert parsed['from_email'] == 'alice@example.com'
|
|
assert parsed['title'] == 'A useful article'
|
|
assert parsed['url'] == 'https://example.com/posts/1'
|
|
assert parsed['comment'] == 'Interesting summary about the protocol.'
|
|
assert parsed['tags'] == ['#AI', '#Security']
|
|
|
|
|
|
def test_ingest_message_stores_link_for_primary_email_sender():
|
|
result = ingest_message(
|
|
'INBOX',
|
|
'101',
|
|
_raw_message(
|
|
'Alice <alice@example.com>',
|
|
'Email sourced link #Fediverse',
|
|
'A comment from the inbox.\nhttps://example.com/inbox-link',
|
|
'<msg-2@example.com>',
|
|
),
|
|
)
|
|
|
|
assert result['status'] == 'stored'
|
|
with get_connection() as conn:
|
|
link = conn.execute(
|
|
'SELECT title, url, comment, user_id FROM links WHERE id = ?',
|
|
(result['link_id'],),
|
|
).fetchone()
|
|
tags = conn.execute(
|
|
'''SELECT tags.name
|
|
FROM tags
|
|
JOIN link_tags ON link_tags.tag_id = tags.id
|
|
WHERE link_tags.link_id = ?
|
|
ORDER BY tags.name''',
|
|
(result['link_id'],),
|
|
).fetchall()
|
|
audit = conn.execute(
|
|
'SELECT status FROM inbound_email_messages WHERE mailbox = ? AND uid = ?',
|
|
('INBOX', '101'),
|
|
).fetchone()
|
|
|
|
assert link['user_id'] == 'user-1'
|
|
assert link['title'] == 'Email sourced link'
|
|
assert link['url'] == 'https://example.com/inbox-link'
|
|
assert link['comment'] == 'A comment from the inbox.'
|
|
assert [row['name'] for row in tags] == ['#Fediverse']
|
|
assert audit['status'] == 'stored'
|
|
|
|
|
|
def test_ingest_message_accepts_verified_additional_address_and_ignores_duplicates():
|
|
with get_connection() as conn:
|
|
conn.execute(
|
|
'INSERT INTO user_email_addresses (id, user_id, email, verified) VALUES (?, ?, ?, 1)',
|
|
('alt-email-test', 'user-1', 'alice-alt@example.com'),
|
|
)
|
|
conn.commit()
|
|
|
|
raw_message = _raw_message(
|
|
'Alice Alt <alice-alt@example.com>',
|
|
'Secondary sender #Links',
|
|
'https://example.com/secondary',
|
|
'<msg-3@example.com>',
|
|
)
|
|
first = ingest_message('INBOX', '102', raw_message)
|
|
second = ingest_message('INBOX', '102', raw_message)
|
|
|
|
assert first['status'] == 'stored'
|
|
assert second['status'] == 'duplicate' |