506 lines
30 KiB
Python
506 lines
30 KiB
Python
import io
|
|
import re
|
|
import time
|
|
import unittest
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import datetime, timedelta, timezone
|
|
from email import policy
|
|
from email.parser import BytesParser
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
import httpx
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from PIL import Image
|
|
|
|
from backend.app import db
|
|
from backend.app.auth import get_current_user
|
|
from backend.app.routers import newsletters as router
|
|
from backend.app.services import newsletters as service, newsletter_store as store, newsletter_catalog as catalog
|
|
from backend.app.services import newsletter_email as template, recap_store, recap_email as mail, email_recaps
|
|
from backend.app.services.jellyfin_identity import link_user, source_key
|
|
from backend.tests.test_backend_quality import TempDatabaseMixin
|
|
from backend.tests.test_email_recaps import runtime
|
|
|
|
USER_ID, SERVER_ID, VIEW_ID = 'a' * 32, 'b' * 32, 'c' * 32
|
|
MOVIE_ID, SERIES_ID, EPISODE_ID, SECOND_EPISODE = '1' * 32, '2' * 32, '3' * 32, '4' * 32
|
|
START = datetime(2026, 9, 4, 9, tzinfo=timezone.utc)
|
|
END = START + timedelta(days=7)
|
|
|
|
|
|
def arrivals():
|
|
return [dict(Id=MOVIE_ID, Name='Arrival', Type='Movie', DateCreated='2026-09-10T10:00:00Z', ProductionYear=2016, ImageTags={'Primary': 'art'}, Overview='A story <with> meaning.'),
|
|
dict(Id=EPISODE_ID, Name='Episode one', Type='Episode', SeriesId=SERIES_ID, SeriesName='Severance', SeriesPrimaryImageTag='art', DateCreated='2026-09-09T10:00:00Z', ParentIndexNumber=2, IndexNumber=1),
|
|
dict(Id=SECOND_EPISODE, Name='Episode two', Type='Episode', SeriesId=SERIES_ID, SeriesName='Severance', SeriesPrimaryImageTag='art', DateCreated='2026-09-08T10:00:00Z', ParentIndexNumber=2, IndexNumber=2)]
|
|
|
|
|
|
def content():
|
|
titles = catalog.group_arrivals(arrivals(), START, END)
|
|
for title in titles:
|
|
title['selected'] = True
|
|
return dict(titles=titles, total_titles=len(titles), source=source_key('http://jellyfin'), server_id=SERVER_ID,
|
|
playback_url='https://watch.example.test', period_start=START.isoformat(), period_end=END.isoformat())
|
|
|
|
|
|
def jpeg():
|
|
image = Image.new('RGB', (100, 150), '#69508c')
|
|
target = io.BytesIO()
|
|
image.save(target, format='JPEG')
|
|
return target.getvalue()
|
|
|
|
|
|
class NewsletterFixture(TempDatabaseMixin):
|
|
def setUp(self):
|
|
super().setUp()
|
|
db.create_user('viewer', 'Example-Password123!', role='admin', email='viewer@example.test')
|
|
link_user('viewer', USER_ID, 'http://jellyfin')
|
|
self.user = db.get_user_by_username('viewer')
|
|
self.runtime = runtime()
|
|
self.runtime.jellyfin_api_key = 'PRIVATE-JF-KEY'
|
|
self.runtime.jellyfin_public_url = 'https://watch.example.test'
|
|
for target, name, result in [(service, 'get_runtime_settings', self.runtime), (email_recaps, 'get_runtime_settings', self.runtime),
|
|
(router, 'get_runtime_settings', self.runtime), (mail, 'get_runtime_settings', self.runtime), (service, 'smtp_email_config_ready', (True, 'ok'))]:
|
|
mocked = patch.object(target, name, return_value=result)
|
|
mocked.start(); self.addCleanup(mocked.stop)
|
|
env = patch.dict('os.environ', {'BACKGROUND_TASKS_ENABLED': 'true'})
|
|
env.start(); self.addCleanup(env.stop)
|
|
self.config = store.save_settings({**store.public_settings(), 'public_url': 'https://beta.example.test'}, datetime.now(timezone.utc))
|
|
|
|
def subscribe(self, when=None, table=store):
|
|
now = time.time() - 10 if when is None else when
|
|
token = table.request_confirmation(self.user, source_key('http://jellyfin'), USER_ID, now)
|
|
self.assertTrue(table.confirm(table.subscription(self.user['id']), now+1))
|
|
return table.subscription(self.user['id']), token
|
|
|
|
def draft(self):
|
|
return store.create_edition(content(), 'Weekend discoveries', '', 'viewer', time.time())
|
|
|
|
def queue(self, sub=None, edition=None, request='test-request'):
|
|
sub = sub or self.subscribe()[0]
|
|
edition = edition or self.draft()
|
|
return store.enqueue_test(sub, edition['id'], edition['revision'], request, self.config['public_url'], time.time())
|
|
|
|
def delivery(self, identity):
|
|
return store.read_one('SELECT * FROM newsletter_deliveries WHERE id=?', (identity,))
|
|
|
|
|
|
class NewsletterConsentTests(NewsletterFixture, unittest.IsolatedAsyncioTestCase):
|
|
async def test_confirmation_only_then_explicit_public_post(self):
|
|
with patch.object(mail, 'send_email') as sender:
|
|
result = await service.subscribe(self.user)
|
|
self.assertEqual(result['state'], 'pending')
|
|
rendered = sender.call_args.args[1]
|
|
self.assertNotIn('Arrival', rendered['body_html'])
|
|
url = re.search(r'https://[^\s]+', rendered['body_text']).group(0)
|
|
self.assertEqual(urlsplit(url).path, '/newsletter-subscription')
|
|
token = parse_qs(urlsplit(url).fragment)['token'][0]
|
|
self.assertEqual(service.token_action(token, 'confirm')['state'], 'ready')
|
|
self.assertEqual(store.subscription(self.user['id'])['state'], 'pending')
|
|
self.assertEqual(service.token_action(token, 'confirm', apply=True)['state'], 'enabled')
|
|
with self.assertRaises(service.NewsletterError):
|
|
service.token_action(token, 'confirm', apply=True)
|
|
|
|
async def test_monthly_consent_is_not_automatic_newsletter_consent(self):
|
|
recap, _ = self.subscribe(table=recap_store)
|
|
self.assertEqual(service.preferences(self.user)['state'], 'off')
|
|
with patch.object(mail, 'send_email') as sender:
|
|
result = await service.subscribe(self.user)
|
|
self.assertEqual(result['state'], 'enabled')
|
|
sender.assert_not_called()
|
|
sub = store.subscription(self.user['id'])
|
|
service.token_action(sub['unsubscribe_token'], 'unsubscribe', apply=True)
|
|
self.assertEqual(recap_store.subscription(self.user['id'])['state'], 'enabled')
|
|
with self.assertRaises(service.NewsletterError):
|
|
service.token_action(recap['unsubscribe_token'], 'unsubscribe', apply=True)
|
|
|
|
def test_unsubscribe_read_only_check_and_repeated_apply(self):
|
|
sub, _ = self.subscribe()
|
|
identity = self.queue(sub)
|
|
token = sub['unsubscribe_token']
|
|
self.assertEqual(service.token_action(token, 'unsubscribe')['state'], 'ready')
|
|
self.assertEqual(self.delivery(identity)['state'], 'queued')
|
|
for _ in range(2):
|
|
self.assertEqual(service.token_action(token, 'unsubscribe', apply=True)['state'], 'off')
|
|
self.assertEqual(self.delivery(identity)['state'], 'cancelled')
|
|
|
|
def test_address_and_identity_changes_revoke_consent(self):
|
|
for index, change in enumerate(['email', 'identity', 'blocked', 'source']):
|
|
with self.subTest(change=change):
|
|
self.subscribe(time.time() - 1600 + index * 400)
|
|
if change == 'email':
|
|
db.set_user_email('viewer', 'changed@example.test')
|
|
db.set_user_email('viewer', 'viewer@example.test')
|
|
elif change == 'identity':
|
|
with store.transaction() as conn:
|
|
conn.execute('UPDATE jellyfin_user_links SET jellyfin_user_id=?', ('d' * 32,))
|
|
conn.execute('UPDATE jellyfin_user_links SET jellyfin_user_id=?', (USER_ID,))
|
|
elif change == 'blocked':
|
|
with store.transaction() as conn:
|
|
conn.execute('UPDATE users SET is_blocked=1')
|
|
conn.execute('UPDATE users SET is_blocked=0')
|
|
else:
|
|
self.runtime.jellyfin_base_url = 'http://changed-jellyfin'
|
|
self.assertEqual(service.preferences(self.user)['state'], 'off')
|
|
|
|
async def test_failed_confirmation_stays_pending_and_rate_limited(self):
|
|
with patch.object(mail, 'send_email', side_effect=mail.DeliveryError('unknown', 'fixture')):
|
|
with self.assertRaises(service.NewsletterError) as raised:
|
|
await service.subscribe(self.user)
|
|
self.assertEqual(raised.exception.status, 502)
|
|
with self.assertRaises(service.NewsletterError) as raised:
|
|
await service.subscribe(self.user)
|
|
self.assertEqual(raised.exception.status, 429)
|
|
self.assertEqual(service.preferences(self.user)['state'], 'pending')
|
|
|
|
def test_expired_or_changed_confirmation_cannot_enable(self):
|
|
token = store.request_confirmation(self.user, source_key('http://jellyfin'), USER_ID, time.time()-90000)
|
|
self.assertEqual(service.preferences(self.user)['state'], 'expired')
|
|
with self.assertRaises(service.NewsletterError):
|
|
service.token_action(token, 'confirm', apply=True)
|
|
token = store.request_confirmation(self.user, source_key('http://jellyfin'), USER_ID, time.time())
|
|
previous = store.subscription(self.user['id'])
|
|
db.set_user_email('viewer', 'changed@example.test')
|
|
self.assertFalse(store.confirm(previous, time.time()))
|
|
with self.assertRaises(service.NewsletterError):
|
|
service.token_action(token, 'confirm', apply=True)
|
|
|
|
|
|
class NewsletterEditionTests(NewsletterFixture, unittest.TestCase):
|
|
def test_defaults_paused_no_users_opted_in(self):
|
|
self.assertFalse(store.settings()['enabled'])
|
|
self.assertEqual(store.overview()['subscribers'], 0)
|
|
self.assertIsNone(store.claim_weekly(datetime.now(timezone.utc)))
|
|
|
|
def test_conflicts_and_title_injection_fail(self):
|
|
row = self.draft()
|
|
selections = [{key: title[key] for key in ('id', 'selected', 'featured')} for title in row['content']['titles']]
|
|
for values in [[], [*selections, selections[0]], [{**selections[0], 'id': 'f'*32}, selections[1]], [{**selections[0], 'selected': False, 'featured': True}, selections[1]]]:
|
|
with self.assertRaises(store.Conflict):
|
|
store.update_edition(row['id'], 1, 'Updated', '', values, time.time())
|
|
result = store.update_edition(row['id'], 1, 'Updated', 'An announcement', selections, time.time())
|
|
self.assertEqual(result['revision'], 2)
|
|
with self.assertRaises(store.Conflict):
|
|
store.update_edition(row['id'], 1, 'Stale tab', '', selections, time.time())
|
|
|
|
def test_test_version_is_frozen_and_request_idempotent(self):
|
|
sub, _ = self.subscribe()
|
|
row = self.draft()
|
|
identity = self.queue(sub, row)
|
|
self.assertEqual(self.queue(sub, row), identity)
|
|
selections = [{key: title[key] for key in ('id', 'selected', 'featured')} for title in row['content']['titles']]
|
|
changed = store.update_edition(row['id'], 1, 'Changed after test queued', '', selections, time.time())
|
|
self.assertEqual(store.version(self.delivery(identity))['subject'], 'Weekend discoveries')
|
|
with self.assertRaises(store.Conflict):
|
|
self.queue(sub, changed)
|
|
with self.assertRaises(store.Conflict):
|
|
self.queue(sub, changed, 'different-request')
|
|
|
|
def test_publish_is_immutable_and_delivery_deduplicated_under_concurrency(self):
|
|
self.subscribe()
|
|
row = self.draft()
|
|
now = time.time()
|
|
store.publish(row['id'], 1, now, now)
|
|
self.assertEqual(store.publish(row['id'], 1, now+1, now+1)['send_at'], now)
|
|
with self.assertRaises(store.Conflict):
|
|
store.update_edition(row['id'], 1, 'No longer editable', '', [], now)
|
|
with ThreadPoolExecutor(max_workers=4) as pool:
|
|
list(pool.map(store.enqueue_due, [now+5]*4))
|
|
self.assertEqual(store.overview()['total'], 1)
|
|
with ThreadPoolExecutor(max_workers=4) as pool:
|
|
claims = list(pool.map(store.claim_delivery, [now+5]*4))
|
|
self.assertEqual(sum(claim is not None for claim in claims), 1)
|
|
store.cancel(row['id'], now+6)
|
|
self.assertEqual(store.overview()['deliveries'][0]['state'], 'cancelled')
|
|
self.assertFalse(store.begin_sending(next(claim for claim in claims if claim), now+6))
|
|
|
|
def test_empty_edition_needs_announcement(self):
|
|
row = store.create_edition({**content(), 'titles': []}, 'Announcement', '', 'viewer', time.time())
|
|
with self.assertRaises(store.Conflict):
|
|
store.publish(row['id'], 1, time.time(), time.time())
|
|
row = store.update_edition(row['id'], 1, 'Announcement', 'Welcome to the weekend.', [], time.time())
|
|
self.assertEqual(store.publish(row['id'], 2, time.time(), time.time())['state'], 'scheduled')
|
|
|
|
def test_schedule_boundary_utc_and_no_past_time(self):
|
|
self.assertEqual(store.next_due(END, 4, 9), END+timedelta(days=7))
|
|
self.assertEqual(store.next_due(END-timedelta(seconds=1), 4, 9), END)
|
|
row = self.draft()
|
|
for when in [datetime.now(), datetime.now(timezone.utc)-timedelta(days=1), datetime.now(timezone.utc)+timedelta(days=91)]:
|
|
with self.assertRaises(service.NewsletterError):
|
|
service.publish(row['id'], 1, when)
|
|
|
|
def test_weekly_catchup_once_not_every_missed_week_and_late_subscriber_excluded(self):
|
|
config = store.save_settings({**self.config, 'enabled': True}, START-timedelta(days=40))
|
|
self.subscribe(END.timestamp()+30)
|
|
claimed = store.claim_weekly(END+timedelta(hours=1))
|
|
self.assertEqual(claimed['due'], END)
|
|
self.assertIsNone(store.claim_weekly(END+timedelta(hours=1)))
|
|
store.complete_weekly(claimed, content(), END+timedelta(hours=1))
|
|
store.complete_weekly(claimed, content(), END+timedelta(hours=1))
|
|
store.enqueue_due((END+timedelta(hours=1)).timestamp())
|
|
self.assertEqual(len(store.overview()['editions']), 1)
|
|
self.assertEqual(store.overview()['total'], 0)
|
|
self.assertEqual(store.settings()['next_send_at'], (END+timedelta(days=7)).timestamp())
|
|
self.assertTrue(config['enabled'])
|
|
|
|
def test_empty_week_is_skipped_and_generation_failure_retries_three_times(self):
|
|
store.save_settings({**self.config, 'enabled': True}, START)
|
|
now = END
|
|
for attempt in range(3):
|
|
claimed = store.claim_weekly(now)
|
|
self.assertEqual(claimed['generation_attempts'], attempt+1)
|
|
store.complete_weekly(claimed, None, now, 'Jellyfin unavailable')
|
|
self.assertIsNone(store.claim_weekly(now+timedelta(seconds=1)))
|
|
now += timedelta(seconds=301)
|
|
self.assertEqual(store.overview()['editions'], [])
|
|
claimed = store.claim_weekly(END+timedelta(days=7))
|
|
store.complete_weekly(claimed, {**content(), 'titles': []}, END+timedelta(days=7))
|
|
self.assertEqual(store.overview()['editions'][0]['state'], 'skipped')
|
|
self.assertEqual(store.overview()['total'], 0)
|
|
|
|
def test_pause_cancels_weekly_but_preserves_manual_and_stale_generation_cannot_publish(self):
|
|
self.subscribe(START.timestamp()-10)
|
|
manual = self.draft()
|
|
store.publish(manual['id'], 1, END.timestamp()+3600, START.timestamp())
|
|
config = store.save_settings({**self.config, 'enabled': True}, START)
|
|
claimed = store.claim_weekly(END)
|
|
store.complete_weekly(claimed, content(), END)
|
|
store.enqueue_due(END.timestamp())
|
|
store.save_settings({**config, 'enabled': False}, END)
|
|
self.assertEqual(store.edition(manual['id'])['state'], 'scheduled')
|
|
self.assertEqual(store.overview()['deliveries'][0]['state'], 'cancelled')
|
|
store.complete_weekly(claimed, content(), END)
|
|
self.assertEqual(len(store.overview()['editions']), 2)
|
|
with self.assertRaises(store.Conflict):
|
|
store.save_settings(config, END)
|
|
|
|
def test_stale_smtp_claim_becomes_unknown_and_never_retried(self):
|
|
identity = self.queue()
|
|
now = time.time()
|
|
claimed = store.claim_delivery(now)
|
|
self.assertTrue(store.begin_sending(claimed, now))
|
|
self.assertIsNone(store.claim_delivery(now+1801))
|
|
self.assertEqual(self.delivery(identity)['state'], 'unknown')
|
|
self.assertIsNone(store.claim_delivery(now+7200))
|
|
|
|
|
|
class NewsletterCatalogTests(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self.runtime = runtime()
|
|
self.runtime.jellyfin_api_key = 'private'
|
|
|
|
def test_grouped_tv_boundary_deduplication_virtual_and_missing_names(self):
|
|
rows = arrivals()
|
|
rows += [rows[0], {**rows[0], 'Id': '5'*32, 'DateCreated': END.isoformat()},
|
|
{**rows[0], 'Id': '6'*32, 'DateCreated': START.isoformat()},
|
|
{**rows[0], 'Id': '7'*32, 'LocationType': 'Virtual'},
|
|
{**rows[1], 'Id': '8'*32, 'SeriesId': '9'*32, 'SeriesName': None}]
|
|
result = catalog.group_arrivals(rows, START, END)
|
|
self.assertEqual(len(result), 3)
|
|
tv = next(title for title in result if title['type'] == 'series')
|
|
self.assertEqual(tv['id'], SERIES_ID)
|
|
self.assertEqual(len(tv['items']), 2)
|
|
self.assertTrue(tv['has_artwork'])
|
|
self.assertIn('2 new episodes', template.description(tv))
|
|
|
|
async def test_collect_pages_and_fails_closed_on_incomplete_or_changed_library(self):
|
|
async def collect_pages(pages):
|
|
with patch.object(catalog, 'PAGE_SIZE', 2), patch.object(catalog, 'MAX_ITEMS', 4), patch.object(catalog, 'get_json', new=AsyncMock(side_effect=[{'Id': SERVER_ID}, *pages])):
|
|
return await catalog.collect(self.runtime, START, END, 1)
|
|
rows = arrivals()
|
|
result = await collect_pages([{'Items': rows[:2], 'TotalRecordCount': 3}, {'Items': rows[2:], 'TotalRecordCount': 3}])
|
|
self.assertEqual(result['total_titles'], 2)
|
|
self.assertEqual(sum(title['selected'] for title in result['titles']), 1)
|
|
cases = [
|
|
[{'Items': rows[:1], 'TotalRecordCount': 3}],
|
|
[{'Items': rows[:2], 'TotalRecordCount': None}],
|
|
[{'Items': rows[:2], 'TotalRecordCount': 3}, {'Items': rows[:1], 'TotalRecordCount': 3}],
|
|
[{'Items': [rows[1], rows[0]], 'TotalRecordCount': 2}],
|
|
[{'Items': [{**rows[0], 'DateCreated': None}], 'TotalRecordCount': 1}],
|
|
[{'Items': rows[:2], 'TotalRecordCount': 5}, {'Items': [rows[2], {**rows[2], 'Id': '5'*32}], 'TotalRecordCount': 5}],
|
|
]
|
|
for pages in cases:
|
|
with self.subTest(pages=pages), self.assertRaises(catalog.CatalogError):
|
|
await collect_pages(pages)
|
|
|
|
async def test_recipient_scope_requires_permitted_parent_even_with_ids(self):
|
|
async def jellyfin(client, runtime, path, params=None):
|
|
if path == '/System/Info': return {'Id': SERVER_ID}
|
|
if path.startswith('/Users/'): return {'Id': USER_ID, 'Policy': {'IsDisabled': False}}
|
|
if path == '/UserViews':
|
|
self.assertEqual(params['UserId'], USER_ID)
|
|
return {'Items': [{'Id': VIEW_ID}]}
|
|
self.assertEqual(path, '/Items')
|
|
self.assertEqual(params['ParentId'], VIEW_ID)
|
|
self.assertEqual(params['UserId'], USER_ID)
|
|
self.assertEqual(params['Recursive'], 'true')
|
|
# The movie and second episode belong to a restricted library.
|
|
return {'Items': [{'Id': SERIES_ID}, {'Id': EPISODE_ID}]}
|
|
with patch.object(catalog, 'get_json', side_effect=jellyfin):
|
|
result = await catalog.for_recipient(self.runtime, content(), USER_ID)
|
|
self.assertEqual(len(result['titles']), 1)
|
|
self.assertEqual(len(result['titles'][0]['items']), 1)
|
|
self.assertIn('1 new episode', template.description(result['titles'][0]))
|
|
|
|
async def test_no_views_does_not_fall_back_to_unrestricted_lookup(self):
|
|
responses = [{'Id': SERVER_ID}, {'Id': USER_ID, 'Policy': {}}, {'Items': []}]
|
|
with patch.object(catalog, 'get_json', new=AsyncMock(side_effect=responses)) as get:
|
|
result = await catalog.for_recipient(self.runtime, content(), USER_ID)
|
|
self.assertEqual(result['titles'], [])
|
|
self.assertEqual(get.await_count, 3)
|
|
|
|
async def test_changed_server_or_user_and_disabled_account_stop_delivery(self):
|
|
for responses in [[{'Id': 'f'*32}], [{'Id': SERVER_ID}, {'Id': 'f'*32, 'Policy': {}}], [{'Id': SERVER_ID}, {}]]:
|
|
with patch.object(catalog, 'get_json', new=AsyncMock(side_effect=responses)), self.assertRaises(catalog.CatalogError):
|
|
await catalog.for_recipient(self.runtime, content(), USER_ID)
|
|
with patch.object(catalog, 'get_json', new=AsyncMock(side_effect=[{'Id': SERVER_ID}, {'Id': USER_ID, 'Policy': {'IsDisabled': True}}])):
|
|
result = await catalog.for_recipient(self.runtime, content(), USER_ID)
|
|
self.assertTrue(result['recipient_disabled'])
|
|
|
|
async def test_posters_are_bounded_reencoded_cached_and_keys_stay_server_side(self):
|
|
original = httpx.AsyncClient
|
|
requests = []
|
|
def handler(request):
|
|
requests.append(request)
|
|
self.assertEqual(request.headers['X-Emby-Token'], 'private')
|
|
self.assertNotIn('private', str(request.url))
|
|
return httpx.Response(200, content=jpeg(), headers={'Content-Type': 'image/jpeg'})
|
|
catalog._posters.clear()
|
|
with patch.object(catalog.httpx, 'AsyncClient', side_effect=lambda **kwargs: original(transport=httpx.MockTransport(handler), **kwargs)):
|
|
result = await catalog.poster(self.runtime, MOVIE_ID)
|
|
self.assertEqual(await catalog.poster(self.runtime, MOVIE_ID), result)
|
|
self.assertEqual(len(requests), 1)
|
|
with Image.open(io.BytesIO(result)) as image:
|
|
self.assertEqual(image.format, 'JPEG')
|
|
self.assertLessEqual(image.width, 160)
|
|
for data in [b'not an image', b'x' * (512*1024+1)]:
|
|
catalog._posters.clear()
|
|
with patch.object(catalog.httpx, 'AsyncClient', side_effect=lambda **kwargs: original(transport=httpx.MockTransport(lambda request: httpx.Response(200, content=data)), **kwargs)):
|
|
self.assertIsNone(await catalog.poster(self.runtime, MOVIE_ID))
|
|
|
|
|
|
class NewsletterDeliveryTests(NewsletterFixture, unittest.IsolatedAsyncioTestCase):
|
|
async def test_weekly_worker_collects_once_and_delivers_to_confirmed_subscriber(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.subscribe((now-timedelta(days=14)).timestamp())
|
|
store.save_settings({**self.config, 'enabled': True}, now-timedelta(days=10))
|
|
def scoped(runtime, snapshot, identity):
|
|
self.assertEqual(identity, USER_ID)
|
|
return snapshot
|
|
def captured(recipient, rendered, message_id, before_data):
|
|
before_data()
|
|
with patch.object(service, 'collect', new=AsyncMock(return_value=content())) as collect, patch.object(catalog, 'for_recipient', new=AsyncMock(side_effect=scoped)), patch.object(catalog, 'posters', new=AsyncMock(return_value={})), patch.object(mail, 'send_email', side_effect=captured) as send:
|
|
await service.run_once()
|
|
await service.run_once()
|
|
self.assertEqual(collect.await_count, 1)
|
|
self.assertEqual(send.call_count, 1)
|
|
self.assertEqual(store.overview()['total'], 1)
|
|
self.assertEqual(store.overview()['deliveries'][0]['state'], 'sent')
|
|
self.assertEqual(store.overview()['editions'][0]['state'], 'complete')
|
|
self.assertGreater(store.settings()['next_send_at'], now.timestamp())
|
|
|
|
async def test_captured_smtp_contains_inline_posters_and_only_accessible_titles(self):
|
|
identity = self.queue()
|
|
scoped = {**content(), 'titles': content()['titles'][1:]}
|
|
scoped.update(subject='Weekend discoveries', intro='A <b>plain text</b> welcome')
|
|
smtp = MagicMock()
|
|
smtp.mail.return_value = smtp.rcpt.return_value = smtp.data.return_value = (250, b'ok')
|
|
with patch.object(catalog, 'for_recipient', new=AsyncMock(return_value=scoped)) as scope, patch.object(catalog, 'posters', new=AsyncMock(return_value={SERIES_ID: jpeg()})), patch.object(mail.smtplib, 'SMTP', return_value=smtp):
|
|
await service.process_delivery(store.claim_delivery(time.time()))
|
|
self.assertEqual(scope.call_args.args[2], USER_ID)
|
|
self.assertEqual(self.delivery(identity)['state'], 'sent')
|
|
message = BytesParser(policy=policy.default).parsebytes(smtp.data.call_args.args[0])
|
|
body = message.get_body(preferencelist=('html',)).get_content()
|
|
self.assertIn('Severance', body)
|
|
self.assertNotIn('Arrival</h3>', body)
|
|
self.assertIn('<b>plain text</b>', body)
|
|
self.assertNotIn('PRIVATE-', body)
|
|
self.assertIn('cid:newsletter-', body)
|
|
self.assertNotIn('data:image', body)
|
|
self.assertIn('serverId=' + SERVER_ID, body)
|
|
images = [part for part in message.walk() if part.get_content_type() == 'image/jpeg']
|
|
self.assertEqual(len(images), 1)
|
|
self.assertEqual(images[0].get_payload(decode=True), jpeg())
|
|
self.assertIn('/newsletter-subscription#action=unsubscribe', body)
|
|
self.assertEqual(str(message['To']), 'viewer@example.test')
|
|
|
|
async def test_cancel_between_preparation_and_smtp_data_stops_send(self):
|
|
identity = self.queue()
|
|
scoped = {**content(), 'subject': 'Subject', 'intro': ''}
|
|
def send(recipient, rendered, message_id, before_data):
|
|
store.disable(self.user['id'])
|
|
before_data()
|
|
self.fail('SMTP DATA must not run after unsubscribe')
|
|
with patch.object(catalog, 'for_recipient', new=AsyncMock(return_value=scoped)), patch.object(catalog, 'posters', new=AsyncMock(return_value={})), patch.object(mail, 'send_email', side_effect=send):
|
|
await service.process_delivery(store.claim_delivery(time.time()))
|
|
self.assertEqual(self.delivery(identity)['state'], 'cancelled')
|
|
|
|
async def test_access_failure_retries_but_empty_access_skips(self):
|
|
identity = self.queue()
|
|
with patch.object(catalog, 'for_recipient', new=AsyncMock(side_effect=catalog.CatalogError('fixture'))), patch.object(mail, 'send_email') as send:
|
|
await service.process_delivery(store.claim_delivery(time.time()))
|
|
self.assertEqual(self.delivery(identity)['state'], 'retry')
|
|
send.assert_not_called()
|
|
with patch.object(catalog, 'for_recipient', new=AsyncMock(return_value={**content(), 'titles': [], 'intro': ''})), patch.object(mail, 'send_email') as send:
|
|
await service.process_delivery(store.claim_delivery(time.time()+301))
|
|
self.assertEqual(self.delivery(identity)['state'], 'skipped')
|
|
send.assert_not_called()
|
|
|
|
def test_multiple_images_share_one_related_mime_container(self):
|
|
rendered = template.render({**content(), 'subject': 'Subject', 'intro': ''}, {MOVIE_ID: jpeg(), SERIES_ID: jpeg()}, self.config['public_url'], self.runtime.jellyfin_public_url, 'https://beta.example.test/profile#newsletters')
|
|
smtp = MagicMock()
|
|
smtp.mail.return_value = smtp.rcpt.return_value = smtp.data.return_value = (250, b'ok')
|
|
with patch.object(mail.smtplib, 'SMTP', return_value=smtp):
|
|
mail.send_email('viewer@example.test', rendered, '<test@example.test>')
|
|
message = BytesParser(policy=policy.default).parsebytes(smtp.data.call_args.args[0])
|
|
related = [part for part in message.walk() if part.get_content_type() == 'multipart/related']
|
|
self.assertEqual(len(related), 1)
|
|
self.assertEqual(len(related[0].get_payload()), 3)
|
|
|
|
|
|
class NewsletterApiTests(NewsletterFixture, unittest.TestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
app = FastAPI()
|
|
app.include_router(router.router)
|
|
self.actor = {'username': 'viewer', 'role': 'admin'}
|
|
app.dependency_overrides[get_current_user] = lambda: self.actor
|
|
self.client = TestClient(app)
|
|
self.addCleanup(self.client.close)
|
|
|
|
def test_admin_endpoints_and_personal_preference_do_not_accept_other_identity(self):
|
|
identity = self.draft()['id']
|
|
self.actor['role'] = 'user'
|
|
for method, path, payload in [('GET', '/admin/newsletters', None), ('PUT', '/admin/newsletters', self.config),
|
|
('POST', '/admin/newsletters/drafts', {'days': 7}), ('GET', '/admin/newsletters/editions/'+identity, None),
|
|
('POST', f'/admin/newsletters/editions/{identity}/publish', {'revision': 1}), ('GET', '/admin/newsletters/artwork/'+MOVIE_ID, None)]:
|
|
self.assertEqual(self.client.request(method, path, json=payload).status_code, 403)
|
|
response = self.client.get('/profile/newsletters')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('no-store', response.headers['cache-control'])
|
|
for extra in [{'email': 'other@example.test'}, {'user_id': 99}, {'jellyfin_id': USER_ID}]:
|
|
self.assertEqual(self.client.put('/profile/newsletters', json={'enabled': True, **extra}).status_code, 422)
|
|
|
|
def test_admin_settings_validate_origin_revision_and_subject(self):
|
|
payload = {key: self.config[key] for key in ('enabled', 'weekday', 'hour', 'limit_titles', 'public_url', 'intro', 'revision')}
|
|
for invalid in ['https://beta.example.test/path', 'javascript:alert(1)', 'https://user:pass@example.test']:
|
|
self.assertEqual(self.client.put('/admin/newsletters', json={**payload, 'public_url': invalid}).status_code, 422)
|
|
self.assertEqual(self.client.put('/admin/newsletters', json=payload).status_code, 200)
|
|
self.assertEqual(self.client.put('/admin/newsletters', json=payload).status_code, 409)
|
|
row = self.draft()
|
|
values = [{'id': entry['id'], 'selected': entry['selected'], 'featured': entry['featured']} for entry in row['content']['titles']]
|
|
for subject in [' ', 'Subject\r\nBcc: someone@example.test']:
|
|
self.assertEqual(self.client.put(f"/admin/newsletters/editions/{row['id']}", json={'revision': 1, 'subject': subject, 'intro': '', 'titles': values}).status_code, 422)
|
|
|
|
def test_confirmation_get_never_changes_consent_and_invalid_action_rejected(self):
|
|
token = store.request_confirmation(self.user, source_key('http://jellyfin'), USER_ID, time.time())
|
|
self.assertEqual(self.client.get('/newsletter-subscription/confirm').status_code, 405)
|
|
self.assertEqual(store.subscription(self.user['id'])['state'], 'pending')
|
|
self.assertEqual(self.client.post('/newsletter-subscription/check', json={'token': token, 'action': 'confirm'}).json()['state'], 'ready')
|
|
self.assertEqual(self.client.post('/newsletter-subscription/confirm', json={'token': token, 'action': 'subscribe'}).status_code, 422)
|
|
self.assertEqual(self.client.post('/newsletter-subscription/confirm', json={'token': token, 'action': 'confirm'}).json()['state'], 'enabled')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|