mrwnslz
54fbd58293
- PodcastSettings (Migration 0014): skip_back / skip_forward (5–60 s),
smart_played, speed_presets (0,5–3,5×), keep_skipped, delete_dequeue,
mobile_downloads, mobile_streaming (allow/ask/never), times_by_speed;
die ersten vier sind anfangs leer (null) und werden mit POST
settings/ {"initial": {...}} nur dort gesetzt, wo noch nichts steht
- GET/POST /podcasts/settings/ und state.settings mit allen Feldern,
geprüft, nichts ändert sich bei einem ungültigen Wert
- PodcastFeed: episode_sort, episode_filter, skip_silence in
feeds/<id>/settings/ und den Feeds des States
- Tempo pro Podcast bis 3,5× (Server und AntennaPod-Import)
- AntennaPod-Import: Sortierung, Filter und Stille pro Podcast, soweit
diora sie zeigen kann; nicht darstellbare werden gezählt
- Browser-Host: Sprungweiten, "gehört" und keep_skipped aus dem Konto,
Tempo bis 3,5×, neue Einstellungen werden gespeichert
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
764 lines
44 KiB
Python
764 lines
44 KiB
Python
import os
|
|
import sqlite3
|
|
import tempfile
|
|
from datetime import datetime, timedelta, timezone
|
|
from io import StringIO
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.core.management import call_command
|
|
from django.core.management.base import CommandError
|
|
from django.test import TestCase
|
|
|
|
from .antennapod import filter_words, normalize_url, parse_filter_terms
|
|
from .models import EpisodeProgress, PodcastEpisode, PodcastFeed, PodcastQueue, PodcastSettings
|
|
|
|
BACKUP_DATE = datetime(2026, 9, 1, 12, 0, tzinfo=timezone.utc)
|
|
T0 = datetime(2026, 8, 1, tzinfo=timezone.utc)
|
|
|
|
|
|
def ms(dt):
|
|
return int(dt.timestamp() * 1000)
|
|
|
|
|
|
class Backup:
|
|
"""A small AntennaPod database: the tables and columns the import reads."""
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE Feeds (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, custom_title TEXT,
|
|
download_url TEXT, link TEXT, author TEXT, image_url TEXT, keep_updated INTEGER DEFAULT 1,
|
|
include_filter TEXT DEFAULT '', exclude_filter TEXT DEFAULT '',
|
|
minimal_duration_filter INTEGER DEFAULT -1, new_episodes_action INTEGER DEFAULT 0,
|
|
feed_playback_speed REAL DEFAULT -1, feed_skip_intro INTEGER DEFAULT 0,
|
|
feed_skip_ending INTEGER DEFAULT 0, feed_volume_adaption INTEGER DEFAULT 0,
|
|
auto_download INTEGER DEFAULT 0, episode_notification INTEGER DEFAULT 0, state INTEGER DEFAULT 0,
|
|
tags TEXT, sort_order TEXT, hide TEXT, feed_skip_silence INTEGER DEFAULT 1);
|
|
CREATE TABLE FeedItems (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, pubDate INTEGER,
|
|
read INTEGER, link TEXT, description TEXT, media INTEGER, feed INTEGER,
|
|
item_identifier TEXT, image_url TEXT);
|
|
CREATE TABLE FeedMedia (id INTEGER PRIMARY KEY AUTOINCREMENT, duration INTEGER, download_url TEXT,
|
|
position INTEGER, playback_completion_date INTEGER, feeditem INTEGER, played_duration INTEGER,
|
|
last_played_time INTEGER, downloaded INTEGER DEFAULT 0, filesize INTEGER DEFAULT 0);
|
|
CREATE TABLE Queue (id INTEGER PRIMARY KEY AUTOINCREMENT, feeditem INTEGER, feed INTEGER);
|
|
CREATE TABLE Favorites (id INTEGER PRIMARY KEY AUTOINCREMENT, feeditem INTEGER, feed INTEGER);
|
|
"""
|
|
|
|
def __init__(self, path):
|
|
self.path = path
|
|
self.db = sqlite3.connect(path)
|
|
self.db.executescript(self.SCHEMA)
|
|
|
|
def feed(self, url, title='Feed', **cols):
|
|
cols = {'download_url': url, 'title': title, **cols}
|
|
cur = self.db.execute(f'INSERT INTO Feeds ({",".join(cols)}) VALUES ({",".join("?" * len(cols))})',
|
|
list(cols.values()))
|
|
self.db.commit()
|
|
return cur.lastrowid
|
|
|
|
def item(self, feed, guid, read=0, audio=None, pub=T0, duration=1800, position=0, completed=None,
|
|
last_played=None, title=None, description='', image_url=None, media=True, downloaded=0, size=0):
|
|
cur = self.db.execute(
|
|
'INSERT INTO FeedItems (title, pubDate, read, description, feed, item_identifier, image_url) '
|
|
'VALUES (?,?,?,?,?,?,?)',
|
|
(title or f'Episode {guid}', ms(pub) if pub else 0, read, description, feed, guid, image_url))
|
|
item_id = cur.lastrowid
|
|
if media:
|
|
self.db.execute(
|
|
'INSERT INTO FeedMedia (duration, download_url, position, playback_completion_date, feeditem, '
|
|
'played_duration, last_played_time, downloaded, filesize) VALUES (?,?,?,?,?,?,?,?,?)',
|
|
(duration * 1000, audio or f'https://cdn.example.org/{guid}.mp3', position * 1000,
|
|
ms(completed) if completed else 0, item_id, 0, ms(last_played) if last_played else 0,
|
|
downloaded, size))
|
|
self.db.commit()
|
|
return item_id
|
|
|
|
def queue(self, *items):
|
|
for i in items:
|
|
self.db.execute('INSERT INTO Queue (feeditem, feed) SELECT id, feed FROM FeedItems WHERE id=?', (i,))
|
|
self.db.commit()
|
|
|
|
def favorite(self, *items):
|
|
for i in items:
|
|
self.db.execute('INSERT INTO Favorites (feeditem, feed) SELECT id, feed FROM FeedItems WHERE id=?', (i,))
|
|
self.db.commit()
|
|
|
|
def close(self):
|
|
self.db.close()
|
|
|
|
|
|
class ImportTestCase(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username='alice', password='pw12345678')
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.ap = Backup(os.path.join(self.tmp.name, 'AntennaPodBackup.db'))
|
|
|
|
def tearDown(self):
|
|
self.ap.close()
|
|
self.tmp.cleanup()
|
|
|
|
def run_import(self, *args, user='alice'):
|
|
out = StringIO()
|
|
call_command('import_antennapod', self.ap.path, '--user', user,
|
|
'--backup-date', BACKUP_DATE.isoformat(), *args, stdout=out)
|
|
return out.getvalue()
|
|
|
|
def diora_feed(self, url='https://example.org/feed', **kw):
|
|
return PodcastFeed.objects.create(user=self.user, rss_url=url, title=kw.pop('title', 'Diora'), **kw)
|
|
|
|
def episode(self, feed, guid, audio=None, **kw):
|
|
return PodcastEpisode.objects.create(feed=feed, guid=guid, title=guid,
|
|
audio_url=audio or f'https://cdn.example.org/{guid}.mp3', **kw)
|
|
|
|
def progress(self, episode):
|
|
return EpisodeProgress.objects.filter(user=self.user, episode=episode).first()
|
|
|
|
def snapshot(self):
|
|
return (
|
|
list(PodcastFeed.objects.order_by('id').values()),
|
|
list(PodcastEpisode.objects.order_by('id').values()),
|
|
list(EpisodeProgress.objects.order_by('id').values()),
|
|
list(PodcastQueue.objects.order_by('id').values('id', 'user_id', 'episode_id', 'position')),
|
|
)
|
|
|
|
def queue(self):
|
|
return list(PodcastQueue.objects.filter(user=self.user).order_by('position')
|
|
.values_list('episode_id', flat=True))
|
|
|
|
|
|
class HelperTests(TestCase):
|
|
def test_normalize_url(self):
|
|
same = ['https://www.Example.org/feed/', 'http://example.org/feed', 'HTTPS://EXAMPLE.ORG/feed',
|
|
'https://example.org:443/feed#x', 'http://example.org:80/feed//']
|
|
self.assertEqual({normalize_url(u) for u in same}, {'example.org/feed'})
|
|
self.assertNotEqual(normalize_url('https://example.org/Feed'), normalize_url('https://example.org/feed'))
|
|
self.assertEqual(normalize_url('https://example.org:8080/f?a=1'), 'example.org:8080/f?a=1')
|
|
for bad in ('antennapod_local:content://x', '', None, 'ftp://example.org/f', 'https://'):
|
|
self.assertIsNone(normalize_url(bad))
|
|
|
|
def test_filter_terms(self):
|
|
self.assertEqual(parse_filter_terms('news "long talk" Teaser'), ['news', 'long talk', 'Teaser'])
|
|
self.assertEqual(parse_filter_terms('"news headlines" '), ['news headlines'])
|
|
self.assertEqual(parse_filter_terms(' a b '), ['a', 'b'])
|
|
self.assertEqual(parse_filter_terms('"open quote'), ['open', 'quote'])
|
|
self.assertEqual(parse_filter_terms('a,b A'), ['a b', 'A'])
|
|
self.assertEqual(parse_filter_terms('x X'), ['x'])
|
|
self.assertEqual(parse_filter_terms(''), [])
|
|
self.assertEqual(filter_words('news "long talk"'), 'news, long talk')
|
|
self.assertEqual(len(filter_words(' '.join(['word%03d' % i for i in range(100)]))), 493)
|
|
|
|
|
|
class FeedImportTests(ImportTestCase):
|
|
def test_feeds_match_by_normalised_url_and_missing_ones_are_created(self):
|
|
mine = self.diora_feed('https://www.Example.org/feed/')
|
|
only_diora = self.diora_feed('https://example.org/only-in-diora')
|
|
self.ap.feed('http://example.org/feed', title='Same')
|
|
self.ap.feed('https://new.example.org/rss', title='Original', custom_title='Renamed', author='Ann',
|
|
link='https://new.example.org/', image_url='https://new.example.org/art.jpg')
|
|
self.ap.feed('antennapod_local:content://folder', title='Local')
|
|
self.ap.feed('https://gone.example.org/rss', title='Not subscribed', state=1)
|
|
|
|
out = self.run_import()
|
|
|
|
self.assertEqual(PodcastFeed.objects.filter(user=self.user).count(), 3)
|
|
mine.refresh_from_db()
|
|
self.assertEqual(mine.title, 'Diora') # matched, not renamed
|
|
self.assertTrue(PodcastFeed.objects.filter(pk=only_diora.pk).exists())
|
|
new = PodcastFeed.objects.get(user=self.user, rss_url='https://new.example.org/rss')
|
|
self.assertEqual((new.title, new.author, new.link, new.artwork_url, new.last_refreshed_at),
|
|
('Renamed', 'Ann', 'https://new.example.org/', 'https://new.example.org/art.jpg', None))
|
|
self.assertIn('Feeds: 4 in backup, 3 subscribed (1 not subscribed ignored); 1 matched, 1 created, '
|
|
'1 skipped (not http/https), 0 duplicates', out)
|
|
self.assertIn('Created feeds (1):\n - Renamed\n', out)
|
|
self.assertIn('Skipped feeds (1):\n - Local (antennapod_local:content://folder)\n', out)
|
|
|
|
def test_two_backup_feeds_for_one_url(self):
|
|
self.ap.feed('https://example.org/feed')
|
|
self.ap.feed('http://www.example.org/feed/')
|
|
out = self.run_import()
|
|
self.assertEqual(PodcastFeed.objects.filter(user=self.user).count(), 1)
|
|
self.assertIn('0 matched, 1 created, 0 skipped (not http/https), 1 duplicates', out)
|
|
|
|
def test_settings(self):
|
|
keep = self.diora_feed('https://example.org/keep', playback_speed=1.5, skip_intro=10, volume=2.0,
|
|
new_episodes='queue', include_words='x', min_duration=5, auto_download=True)
|
|
changed = self.diora_feed('https://example.org/changed', playback_speed=1.5)
|
|
self.ap.feed('https://example.org/keep', auto_download=1) # all AntennaPod defaults
|
|
self.ap.feed('https://example.org/changed', feed_playback_speed=1.2000000476837158, feed_skip_intro=30,
|
|
feed_skip_ending=700, feed_volume_adaption=2, include_filter='news "long talk" ',
|
|
exclude_filter='"Trailer"', minimal_duration_filter=1500, new_episodes_action=2,
|
|
auto_download=2)
|
|
self.ap.feed('https://example.org/fast', feed_playback_speed=3.5, feed_volume_adaption=5,
|
|
minimal_duration_filter=59, new_episodes_action=3)
|
|
self.ap.feed('https://example.org/slow', feed_playback_speed=0.25, feed_volume_adaption=3,
|
|
new_episodes_action=1)
|
|
self.ap.feed('https://example.org/loud', feed_volume_adaption=4)
|
|
self.ap.feed('https://example.org/quiet', feed_volume_adaption=1)
|
|
|
|
out = self.run_import()
|
|
|
|
keep.refresh_from_db()
|
|
self.assertEqual(
|
|
(keep.playback_speed, keep.skip_intro, keep.volume, keep.new_episodes, keep.auto_queue,
|
|
keep.include_words, keep.min_duration, keep.auto_download),
|
|
(1.5, 10, 2.0, 'queue', True, 'x', 5, False), # auto_download "global", --global-auto-download off
|
|
)
|
|
changed.refresh_from_db()
|
|
self.assertEqual(
|
|
(changed.playback_speed, changed.skip_intro, changed.skip_outro, changed.volume, changed.include_words,
|
|
changed.exclude_words, changed.min_duration, changed.new_episodes, changed.auto_download),
|
|
(1.2, 30, 600, 0.2, 'news, long talk', 'Trailer', 25, 'ignore', True),
|
|
)
|
|
feeds = {f.rss_url.rsplit('/', 1)[1]: f for f in PodcastFeed.objects.filter(user=self.user)}
|
|
fast, slow = feeds['fast'], feeds['slow']
|
|
self.assertEqual((fast.playback_speed, fast.volume, fast.min_duration, fast.new_episodes, fast.auto_queue),
|
|
(3.5, 2.5, 0, 'queue', True))
|
|
self.assertEqual((slow.playback_speed, slow.volume, slow.new_episodes, slow.auto_queue),
|
|
(0.5, 1.5, 'inbox', False))
|
|
self.assertEqual((feeds['loud'].volume, feeds['quiet'].volume), (2.0, 0.5))
|
|
self.assertIn('Settings: changed: speed 3, skip intro 1, skip outro 1, volume 5, include words 1, '
|
|
'exclude words 1, min duration 1, new episodes 2, auto-download 2', out)
|
|
|
|
def test_not_kept_updated_feeds_bring_nothing_new(self):
|
|
self.ap.feed('https://example.org/a', keep_updated=0)
|
|
self.ap.feed('https://example.org/b', keep_updated=0, new_episodes_action=1)
|
|
self.ap.feed('https://example.org/c', keep_updated=0, new_episodes_action=3)
|
|
self.ap.feed('https://example.org/d', keep_updated=1)
|
|
out = self.run_import()
|
|
new = dict(PodcastFeed.objects.filter(user=self.user).values_list('rss_url', 'new_episodes'))
|
|
self.assertEqual(new, {'https://example.org/a': 'ignore', 'https://example.org/b': 'ignore',
|
|
'https://example.org/c': 'queue', 'https://example.org/d': 'inbox'})
|
|
self.assertIn('2 feeds not kept updated in AntennaPod -> new episodes "ignore"', out)
|
|
|
|
def test_auto_download(self):
|
|
self.diora_feed('https://example.org/off', auto_download=True)
|
|
for name, code in (('off', 0), ('global', 1), ('on', 2), ('global2', 1)):
|
|
self.ap.feed(f'https://example.org/{name}', auto_download=code)
|
|
|
|
def values():
|
|
return dict(PodcastFeed.objects.filter(user=self.user).values_list('rss_url', 'auto_download'))
|
|
|
|
out = self.run_import()
|
|
self.assertEqual(values(), {'https://example.org/off': False, 'https://example.org/global': False,
|
|
'https://example.org/on': True, 'https://example.org/global2': False})
|
|
self.assertIn("Auto-download: on for 1 feeds (1 on in AntennaPod, 2 use AntennaPod's global setting -> off "
|
|
'by --global-auto-download); default for new subscriptions: off\n', out)
|
|
self.assertFalse(PodcastSettings.for_user(self.user).auto_download_default)
|
|
|
|
out = self.run_import('--global-auto-download', 'on')
|
|
self.assertEqual(values(), {'https://example.org/off': False, 'https://example.org/global': True,
|
|
'https://example.org/on': True, 'https://example.org/global2': True})
|
|
self.assertIn("Auto-download: on for 3 feeds (1 on in AntennaPod, 2 use AntennaPod's global setting -> on "
|
|
'by --global-auto-download); default for new subscriptions: on (set by this import)\n', out)
|
|
self.assertTrue(PodcastSettings.for_user(self.user).auto_download_default)
|
|
# Once more: nothing to set any more; "off" leaves the default alone.
|
|
self.assertIn('default for new subscriptions: on\n', self.run_import('--global-auto-download', 'on'))
|
|
self.assertIn('default for new subscriptions: on\n', self.run_import())
|
|
|
|
def test_episode_notifications(self):
|
|
self.diora_feed('https://example.org/diora-on', notify=True)
|
|
self.diora_feed('https://example.org/on-both', notify=True)
|
|
self.ap.feed('https://example.org/diora-on', episode_notification=0)
|
|
self.ap.feed('https://example.org/on-both', episode_notification=1)
|
|
self.ap.feed('https://example.org/on', episode_notification=1)
|
|
self.ap.feed('https://example.org/off')
|
|
out = self.run_import()
|
|
values = dict(PodcastFeed.objects.filter(user=self.user).values_list('rss_url', 'notify'))
|
|
# AntennaPod's "off" is its default: diora's own setting stays.
|
|
self.assertEqual(values, {'https://example.org/diora-on': True, 'https://example.org/on-both': True,
|
|
'https://example.org/on': True, 'https://example.org/off': False})
|
|
self.assertIn('notifications 1', out)
|
|
|
|
def test_episode_lists_and_skip_silence(self):
|
|
"""Per podcast: its episode order and filter (where diora has them) and skipping silence."""
|
|
self.diora_feed('https://example.org/mine', episode_sort='long', episode_filter=['favorites'],
|
|
skip_silence=True)
|
|
self.ap.feed('https://example.org/mine') # nothing set in AntennaPod: diora's own stay
|
|
self.ap.feed('https://example.org/series', sort_order='1', hide='unplayed', feed_skip_silence=0)
|
|
self.ap.feed('https://example.org/new', sort_order='2', hide='paused,is_favorite,downloaded,queued')
|
|
self.ap.feed('https://example.org/title', sort_order='3', hide='played', feed_skip_silence=2)
|
|
self.ap.feed('https://example.org/short', sort_order='5')
|
|
self.ap.feed('https://example.org/long', sort_order='6', feed_skip_silence=1)
|
|
self.ap.feed('https://example.org/global', sort_order='11', hide='')
|
|
self.ap.feed('https://example.org/z-a', sort_order='4', hide='unplayed,not_downloaded')
|
|
self.ap.feed('https://example.org/size', sort_order='10', hide='new')
|
|
|
|
out = self.run_import()
|
|
|
|
got = {f.rss_url.rsplit('/', 1)[1]: (f.episode_sort, f.episode_filter, f.skip_silence)
|
|
for f in PodcastFeed.objects.filter(user=self.user)}
|
|
self.assertEqual(got, {
|
|
'mine': ('long', ['favorites'], True),
|
|
'series': ('old', ['unplayed'], False),
|
|
'new': ('new', ['in_progress', 'favorites', 'downloaded', 'queued'], None),
|
|
'title': ('title', ['played'], True),
|
|
'short': ('short', [], None),
|
|
'long': ('long', [], None),
|
|
'global': ('', [], None),
|
|
'z-a': ('', [], None), # title Z-A, "not downloaded": nothing diora can show
|
|
'size': ('', [], None),
|
|
})
|
|
self.assertIn('episode order 5, episode filter 3, skip silence 2', out)
|
|
self.assertIn("episode lists diora can't show that way (left as they are): 2 orders, 2 filters", out)
|
|
# Through the API as the apps see it.
|
|
self.client.force_login(self.user)
|
|
feeds = {f['rss_url'].rsplit('/', 1)[1]: f for f in self.client.get('/podcasts/state/').json()['feeds']}
|
|
self.assertEqual((feeds['series']['episode_sort'], feeds['series']['episode_filter'],
|
|
feeds['series']['skip_silence']), ('old', ['unplayed'], False))
|
|
|
|
def test_backup_without_list_and_silence_columns(self):
|
|
for column in ('sort_order', 'hide', 'feed_skip_silence'):
|
|
self.ap.db.execute(f'ALTER TABLE Feeds DROP COLUMN {column}')
|
|
self.ap.feed('https://example.org/old')
|
|
self.run_import()
|
|
f = PodcastFeed.objects.get(user=self.user, rss_url='https://example.org/old')
|
|
self.assertEqual((f.episode_sort, f.episode_filter, f.skip_silence), ('', [], None))
|
|
|
|
def test_backup_without_episode_notification_column(self):
|
|
self.ap.db.execute('ALTER TABLE Feeds DROP COLUMN episode_notification')
|
|
self.ap.feed('https://example.org/old')
|
|
self.run_import()
|
|
self.assertFalse(PodcastFeed.objects.get(user=self.user, rss_url='https://example.org/old').notify)
|
|
|
|
def test_created_feeds_start_with_the_users_default(self):
|
|
PodcastSettings.objects.create(user=self.user, auto_download_default=True)
|
|
self.ap.feed('https://example.org/explicit-off', auto_download=0)
|
|
self.ap.feed('https://example.org/on', auto_download=2)
|
|
self.run_import()
|
|
values = dict(PodcastFeed.objects.filter(user=self.user).values_list('rss_url', 'auto_download'))
|
|
self.assertEqual(values, {'https://example.org/explicit-off': False, 'https://example.org/on': True})
|
|
|
|
|
|
class EpisodeImportTests(ImportTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.feed = self.diora_feed()
|
|
self.ap_feed = self.ap.feed('https://example.org/feed')
|
|
|
|
def test_matching_and_creating_episodes(self):
|
|
by_guid = self.episode(self.feed, 'g1')
|
|
by_url = self.episode(self.feed, 'other-guid', audio='http://www.cdn.example.org/g2.mp3')
|
|
stateless = self.episode(self.feed, 'g4')
|
|
untouched = self.episode(self.feed, 'g6')
|
|
self.ap.item(self.ap_feed, 'g1', read=1)
|
|
self.ap.item(self.ap_feed, 'g2', read=1, audio='https://cdn.example.org/g2.mp3')
|
|
self.ap.item(self.ap_feed, 'g3', read=1, title='Created', description='<p>Notes</p>', duration=3600,
|
|
pub=T0 + timedelta(days=3), image_url='https://example.org/g3.jpg')
|
|
self.ap.item(self.ap_feed, 'g4', read=0)
|
|
self.ap.item(self.ap_feed, 'g5', read=0) # no state, not in diora: not imported
|
|
self.ap.item(self.ap_feed, None, read=-1, audio='https://cdn.example.org/noguid.mp3')
|
|
self.ap.item(self.ap_feed, 'g7', read=1, media=False)
|
|
|
|
out = self.run_import()
|
|
|
|
self.assertTrue(self.progress(by_guid).played)
|
|
self.assertTrue(self.progress(by_url).played)
|
|
created = PodcastEpisode.objects.get(feed=self.feed, guid='g3')
|
|
self.assertEqual(
|
|
(created.title, created.description, created.audio_url, created.duration_seconds, created.pub_date,
|
|
created.artwork_url),
|
|
('Created', '<p>Notes</p>', 'https://cdn.example.org/g3.mp3', 3600, T0 + timedelta(days=3),
|
|
'https://example.org/g3.jpg'),
|
|
)
|
|
self.assertTrue(self.progress(created).played)
|
|
# Without a guid, the audio URL is the guid (as when diora reads the feed).
|
|
noguid = PodcastEpisode.objects.get(feed=self.feed, guid='https://cdn.example.org/noguid.mp3')
|
|
self.assertFalse(self.progress(noguid).dismissed)
|
|
self.assertFalse(PodcastEpisode.objects.filter(guid__in=['g5', 'g7']).exists())
|
|
# Known to AntennaPod and not new there: out of the inbox, not played.
|
|
p = self.progress(stateless)
|
|
self.assertEqual((p.dismissed, p.played), (True, False))
|
|
self.assertIsNone(self.progress(untouched))
|
|
self.assertIn('Episodes: 5 with a state: 1 matched by guid, 1 by URL, 2 created, '
|
|
'1 skipped (no audio URL), 0 in skipped feeds', out)
|
|
self.assertIn('1 without a state matched to diora episodes', out)
|
|
self.assertIn('Inbox: 4 before, 2 after', out) # g6 and noguid
|
|
|
|
def test_states(self):
|
|
played_at = T0 + timedelta(days=10)
|
|
self.ap.item(self.ap_feed, 'played', read=1, position=0, completed=played_at, last_played=played_at)
|
|
self.ap.item(self.ap_feed, 'marked', read=1, pub=T0 + timedelta(days=2))
|
|
self.ap.item(self.ap_feed, 'started', read=0, position=600, last_played=played_at - timedelta(hours=1))
|
|
self.ap.item(self.ap_feed, 'new', read=-1, pub=T0 + timedelta(days=5))
|
|
self.ap.item(self.ap_feed, 'nodate', read=1, pub=None)
|
|
self.ap.item(self.ap_feed, 'future', read=1, pub=BACKUP_DATE + timedelta(days=30))
|
|
self.run_import()
|
|
|
|
def state(guid):
|
|
p = self.progress(PodcastEpisode.objects.get(guid=guid))
|
|
return p.played, p.dismissed, p.position_seconds, p.updated_at
|
|
|
|
self.assertEqual(state('played'), (True, True, 0, played_at))
|
|
self.assertEqual(state('marked'), (True, True, 0, T0 + timedelta(days=2))) # no activity: pub date
|
|
self.assertEqual(state('started'), (False, True, 600, played_at - timedelta(hours=1)))
|
|
self.assertEqual(state('new'), (False, False, 0, T0 + timedelta(days=5)))
|
|
self.assertEqual(state('nodate'), (True, True, 0, BACKUP_DATE))
|
|
self.assertEqual(state('future'), (True, True, 0, BACKUP_DATE))
|
|
|
|
def test_favourites_are_a_union(self):
|
|
diora_star = self.episode(self.feed, 'a')
|
|
p = EpisodeProgress.objects.create(user=self.user, episode=diora_star, favorite=True)
|
|
EpisodeProgress.objects.filter(pk=p.pk).update(updated_at=BACKUP_DATE - timedelta(days=5))
|
|
both = self.episode(self.feed, 'b')
|
|
self.ap.item(self.ap_feed, 'a', read=1) # AntennaPod wins, but diora's star stays
|
|
b = self.ap.item(self.ap_feed, 'b', read=0)
|
|
c = self.ap.item(self.ap_feed, 'c', read=0)
|
|
self.ap.favorite(b, c)
|
|
out = self.run_import()
|
|
p = self.progress(diora_star)
|
|
self.assertEqual((p.favorite, p.played), (True, True))
|
|
self.assertTrue(self.progress(both).favorite)
|
|
self.assertTrue(self.progress(PodcastEpisode.objects.get(guid='c')).favorite)
|
|
self.assertIn('Favourites: 2 in backup; 1 before, 3 after', out)
|
|
|
|
def test_merge_rule(self):
|
|
newer = self.episode(self.feed, 'newer')
|
|
older = self.episode(self.feed, 'older')
|
|
dismissed = self.episode(self.feed, 'dismissed')
|
|
after_backup = BACKUP_DATE + timedelta(days=1)
|
|
for ep, values in ((newer, dict(position_seconds=100)), (older, dict(position_seconds=100)),
|
|
(dismissed, dict(dismissed=True))):
|
|
p = EpisodeProgress.objects.create(user=self.user, episode=ep, **values)
|
|
EpisodeProgress.objects.filter(pk=p.pk).update(
|
|
updated_at=after_backup if ep is not older else BACKUP_DATE - timedelta(days=1))
|
|
self.ap.item(self.ap_feed, 'newer', read=1, completed=BACKUP_DATE - timedelta(days=2))
|
|
self.ap.item(self.ap_feed, 'older', read=1, completed=BACKUP_DATE - timedelta(days=2))
|
|
n = self.ap.item(self.ap_feed, 'dismissed', read=-1)
|
|
self.ap.favorite(n)
|
|
out = self.run_import()
|
|
p = self.progress(newer)
|
|
self.assertEqual((p.played, p.position_seconds, p.updated_at), (False, 100, after_backup))
|
|
p = self.progress(older)
|
|
self.assertEqual((p.played, p.position_seconds, p.updated_at), (True, 0, BACKUP_DATE - timedelta(days=2)))
|
|
# diora's newer "not new" stays; the star comes over anyway, without
|
|
# touching updated_at.
|
|
p = self.progress(dismissed)
|
|
self.assertEqual((p.dismissed, p.favorite, p.updated_at), (True, True, after_backup))
|
|
self.assertIn('Progress: 0 new, 1 updated, 0 unchanged, 2 kept (diora newer)', out)
|
|
|
|
def test_activity_after_diora_change_wins(self):
|
|
ep = self.episode(self.feed, 'e')
|
|
p = EpisodeProgress.objects.create(user=self.user, episode=ep, position_seconds=50)
|
|
EpisodeProgress.objects.filter(pk=p.pk).update(updated_at=BACKUP_DATE + timedelta(hours=1))
|
|
self.ap.item(self.ap_feed, 'e', read=0, position=300, last_played=BACKUP_DATE + timedelta(hours=2))
|
|
self.run_import()
|
|
self.assertEqual(self.progress(ep).position_seconds, 300)
|
|
|
|
def test_queue(self):
|
|
x = self.episode(self.feed, 'x')
|
|
y = self.episode(self.feed, 'y')
|
|
PodcastQueue.objects.create(user=self.user, episode=x, position=0)
|
|
PodcastQueue.objects.create(user=self.user, episode=y, position=1)
|
|
a = self.ap.item(self.ap_feed, 'a', read=-1)
|
|
b = self.ap.item(self.ap_feed, 'b', read=0)
|
|
y_ap = self.ap.item(self.ap_feed, 'y', read=0)
|
|
local = self.ap.feed('antennapod_local:x')
|
|
lost = self.ap.item(local, 'l', read=0)
|
|
self.ap.queue(b, y_ap, lost, a)
|
|
out = self.run_import()
|
|
ids = {e.guid: e.id for e in PodcastEpisode.objects.filter(feed=self.feed)}
|
|
self.assertEqual(self.queue(), [ids['b'], y.id, ids['a'], x.id])
|
|
self.assertEqual(list(PodcastQueue.objects.filter(user=self.user).order_by('position')
|
|
.values_list('position', flat=True)), [0, 1, 2, 3])
|
|
# Queued is not new, even if AntennaPod still had it as new.
|
|
self.assertTrue(self.progress(PodcastEpisode.objects.get(guid='a')).dismissed)
|
|
self.assertIn('Queue: 4 in backup (1 not found); 2 before, 4 after (1 diora entries after', out)
|
|
|
|
def test_queue_kept_sorted(self):
|
|
PodcastSettings.objects.create(user=self.user, keep_sorted='old')
|
|
a = self.ap.item(self.ap_feed, 'a', read=0, pub=T0 + timedelta(days=2))
|
|
b = self.ap.item(self.ap_feed, 'b', read=0, pub=T0 + timedelta(days=1))
|
|
self.ap.queue(a, b)
|
|
out = self.run_import()
|
|
self.assertEqual([PodcastEpisode.objects.get(pk=i).guid for i in self.queue()], ['b', 'a'])
|
|
self.assertIn('kept sorted by "old"', out)
|
|
|
|
def test_dry_run_writes_nothing(self):
|
|
self.episode(self.feed, 'g1')
|
|
PodcastQueue.objects.create(user=self.user, episode=self.episode(self.feed, 'q'), position=0)
|
|
g = self.ap.item(self.ap_feed, 'g1', read=1)
|
|
self.ap.item(self.ap_feed, 'g2', read=-1)
|
|
self.ap.feed('https://new.example.org/rss', feed_playback_speed=2.0)
|
|
self.ap.queue(g)
|
|
before = self.snapshot()
|
|
out = self.run_import('--dry-run')
|
|
self.assertEqual(self.snapshot(), before)
|
|
self.assertIn('DRY RUN', out)
|
|
self.assertIn('1 created', out)
|
|
self.assertIn('Queue: 1 in backup (0 not found); 1 before, 2 after', out)
|
|
self.assertNotIn('DRY RUN', self.run_import())
|
|
self.assertNotEqual(self.snapshot(), before)
|
|
|
|
def test_running_twice_changes_nothing_more(self):
|
|
self.episode(self.feed, 'g1')
|
|
self.episode(self.feed, 'stateless')
|
|
PodcastQueue.objects.create(user=self.user, episode=self.episode(self.feed, 'q'), position=0)
|
|
g1 = self.ap.item(self.ap_feed, 'g1', read=0, position=60, last_played=T0)
|
|
self.ap.item(self.ap_feed, 'stateless', read=0)
|
|
g2 = self.ap.item(self.ap_feed, 'g2', read=-1)
|
|
self.ap.item(self.ap_feed, 'g3', read=1)
|
|
self.ap.item(self.ap_feed, 'g4', read=1, pub=None)
|
|
self.ap.feed('https://new.example.org/rss', feed_playback_speed=2.0, keep_updated=0)
|
|
self.ap.queue(g2, g1)
|
|
self.ap.favorite(g1)
|
|
self.run_import()
|
|
first = self.snapshot()
|
|
out = self.run_import()
|
|
self.assertEqual(self.snapshot(), first)
|
|
self.assertIn('2 matched, 0 created', out)
|
|
self.assertIn('Settings: changed: speed 0, skip intro 0, skip outro 0, volume 0, include words 0, '
|
|
'exclude words 0, min duration 0, new episodes 0, auto-download 0', out)
|
|
self.assertIn('0 created, 0 skipped', out)
|
|
self.assertIn('Progress: 0 new, 0 updated, 4 unchanged, 0 kept (diora newer); 0 taken out', out)
|
|
|
|
|
|
class DownloadImportTests(ImportTestCase):
|
|
"""Episodes on the phone in AntennaPod: offered to the app for download."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.feed = self.diora_feed()
|
|
self.ap_feed = self.ap.feed('https://example.org/feed')
|
|
|
|
def offered(self):
|
|
return PodcastSettings.for_user(self.user).import_downloads
|
|
|
|
def test_downloads_are_offered(self):
|
|
played = self.episode(self.feed, 'played')
|
|
known = self.episode(self.feed, 'known')
|
|
self.ap.item(self.ap_feed, 'played', read=1, pub=T0 + timedelta(days=1), downloaded=1, size=30_000_000)
|
|
self.ap.item(self.ap_feed, 'known', read=0, pub=T0 + timedelta(days=3), downloaded=ms(T0), size=20_400_000)
|
|
self.ap.item(self.ap_feed, 'on-phone', read=0, pub=T0 + timedelta(days=2), downloaded=1,
|
|
size=27_634_898_567_168) # garbage: unknown
|
|
self.ap.item(self.ap_feed, 'not-downloaded', read=0)
|
|
self.ap.item(self.ap_feed, 'local', read=0, audio='antennapod_local:x', downloaded=1, size=5)
|
|
self.ap.item(self.ap.feed('antennapod_local:folder'), 'file', read=0, downloaded=1, size=7)
|
|
|
|
before = self.snapshot()
|
|
out = self.run_import('--dry-run')
|
|
self.assertEqual(self.snapshot(), before)
|
|
self.assertFalse(PodcastSettings.objects.filter(user=self.user).exists())
|
|
self.assertIn('Downloads: 3 on the phone in AntennaPod (50 MB) -> offered to the app for download '
|
|
'(1 episodes created for them); 2 not found (feeds not imported, no audio URL)\n', out)
|
|
|
|
out = self.run_import()
|
|
created = PodcastEpisode.objects.get(feed=self.feed, guid='on-phone')
|
|
self.assertEqual(self.offered(), [ # newest first
|
|
{'episode_id': known.id, 'size': 20_400_000}, {'episode_id': created.id, 'size': 0},
|
|
{'episode_id': played.id, 'size': 30_000_000}])
|
|
# Created only to be downloaded; not new in AntennaPod, so not in the inbox.
|
|
p = self.progress(created)
|
|
self.assertEqual((p.played, p.dismissed), (False, True))
|
|
self.assertFalse(PodcastEpisode.objects.filter(guid__in=['not-downloaded', 'local', 'file']).exists())
|
|
self.assertIn('Episodes: 1 with a state: 1 matched by guid, 0 by URL, 0 created, '
|
|
'0 skipped (no audio URL), 0 in skipped feeds', out)
|
|
|
|
# Another backup later: its downloads replace the list.
|
|
self.ap.db.execute('UPDATE FeedMedia SET downloaded = 0 WHERE filesize != 30000000')
|
|
self.ap.db.commit()
|
|
out = self.run_import()
|
|
self.assertEqual(self.offered(), [{'episode_id': played.id, 'size': 30_000_000}])
|
|
self.assertIn('Downloads: 1 on the phone in AntennaPod (30 MB) -> offered to the app for download\n', out)
|
|
self.ap.db.execute('UPDATE FeedMedia SET downloaded = 0')
|
|
self.ap.db.commit()
|
|
out = self.run_import()
|
|
self.assertEqual(self.offered(), [])
|
|
self.assertIn('Downloads: 0 on the phone in AntennaPod (0 MB)', out)
|
|
|
|
def test_no_downloads_offers_nothing(self):
|
|
self.episode(self.feed, 'known')
|
|
self.ap.item(self.ap_feed, 'known', read=0, downloaded=1, size=20_400_000)
|
|
self.run_import()
|
|
self.assertEqual(len(self.offered()), 1)
|
|
out = self.run_import('--no-downloads')
|
|
self.assertEqual(self.offered(), [])
|
|
self.assertIn('Downloads: 1 on the phone in AntennaPod (20 MB) -> not offered (--no-downloads)\n', out)
|
|
|
|
def test_one_entry_per_episode(self):
|
|
ep = self.episode(self.feed, 'g')
|
|
self.ap.item(self.ap_feed, 'g', read=1, downloaded=1, size=10)
|
|
self.ap.item(self.ap.feed('http://www.example.org/feed/'), 'g', read=1, downloaded=1, size=10)
|
|
self.run_import()
|
|
self.assertEqual(self.offered(), [{'episode_id': ep.id, 'size': 10}])
|
|
|
|
def test_backup_without_download_columns(self):
|
|
self.ap.db.executescript('ALTER TABLE FeedMedia DROP COLUMN downloaded; '
|
|
'ALTER TABLE FeedMedia DROP COLUMN filesize;')
|
|
self.ap.db.execute('INSERT INTO FeedItems (title, read, feed, item_identifier) VALUES (?,?,?,?)',
|
|
('G', 1, self.ap_feed, 'g'))
|
|
self.ap.db.commit()
|
|
PodcastSettings.objects.create(user=self.user, import_downloads=[{'episode_id': 1, 'size': 1}])
|
|
out = self.run_import()
|
|
self.assertEqual(self.offered(), [])
|
|
self.assertIn('Downloads: 0 on the phone', out)
|
|
|
|
|
|
class ImportDownloadsViewTests(TestCase):
|
|
"""GET import-downloads/, its clear/, and the count in state/."""
|
|
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username='alice', password='pw12345678')
|
|
self.client.force_login(self.user)
|
|
feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/a', title='A')
|
|
self.ep = PodcastEpisode.objects.create(feed=feed, guid='e', title='E', audio_url='https://example.org/e.mp3',
|
|
duration_seconds=1800, pub_date=T0)
|
|
EpisodeProgress.objects.create(user=self.user, episode=self.ep, position_seconds=120)
|
|
no_audio = PodcastEpisode.objects.create(feed=feed, guid='n', title='N', audio_url='')
|
|
bob = User.objects.create_user(username='bob', password='pw12345678')
|
|
bobs_feed = PodcastFeed.objects.create(user=bob, rss_url='https://example.org/a', title='A')
|
|
bobs = PodcastEpisode.objects.create(feed=bobs_feed, guid='b', title='B', audio_url='https://example.org/b.mp3')
|
|
gone_feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/gone', title='Gone')
|
|
gone = PodcastEpisode.objects.create(feed=gone_feed, guid='g', title='G', audio_url='https://example.org/g.mp3')
|
|
gone_feed.delete() # unsubscribed since the import
|
|
PodcastSettings.objects.create(user=self.user, import_downloads=[
|
|
{'episode_id': i, 'size': 1000 + n} for n, i in enumerate((bobs.id, self.ep.id, no_audio.id, gone.id))])
|
|
|
|
def test_offered_episodes(self):
|
|
data = self.client.get('/podcasts/import-downloads/').json()
|
|
self.assertEqual(data['bytes'], 1001)
|
|
[e] = data['episodes']
|
|
self.assertEqual((e['id'], e['title'], e['feed__title'], e['duration_seconds'], e['pub_date'],
|
|
e['position_seconds'], e['in_queue'], e['size']),
|
|
(self.ep.id, 'E', 'A', 1800, T0.isoformat(), 120, False, 1001))
|
|
# The same keys as an inbox entry of state/, plus the size.
|
|
state = self.client.get('/podcasts/state/').json()
|
|
self.assertEqual(set(e), set(state['inbox'][0]) | {'size'})
|
|
self.assertEqual(state['import_downloads'], {'count': 1, 'bytes': 1001})
|
|
|
|
def test_clear(self):
|
|
self.assertEqual(self.client.get('/podcasts/import-downloads/clear/').status_code, 405)
|
|
r = self.client.post('/podcasts/import-downloads/clear/', '{}', content_type='application/json')
|
|
self.assertEqual(r.json(), {'ok': True})
|
|
self.assertEqual(PodcastSettings.objects.get(user=self.user).import_downloads, [])
|
|
self.assertEqual(self.client.get('/podcasts/import-downloads/').json(), {'episodes': [], 'bytes': 0})
|
|
self.assertEqual(self.client.get('/podcasts/state/').json()['import_downloads'], {'count': 0, 'bytes': 0})
|
|
|
|
def test_other_user_and_login(self):
|
|
self.client.force_login(User.objects.get(username='bob'))
|
|
self.assertEqual(self.client.get('/podcasts/import-downloads/').json(), {'episodes': [], 'bytes': 0})
|
|
self.assertEqual(self.client.get('/podcasts/state/').json()['import_downloads'], {'count': 0, 'bytes': 0})
|
|
self.client.logout()
|
|
self.assertEqual(self.client.get('/podcasts/import-downloads/').status_code, 401)
|
|
self.assertEqual(self.client.post('/podcasts/import-downloads/clear/').status_code, 401)
|
|
self.assertEqual(len(PodcastSettings.objects.get(user=self.user).import_downloads), 4)
|
|
|
|
|
|
class QuietInboxTests(ImportTestCase):
|
|
"""What AntennaPod never showed as new leaves diora's inbox too."""
|
|
|
|
def inbox(self):
|
|
hidden = EpisodeProgress.objects.filter(user=self.user, dismissed=True).values('episode_id')
|
|
return sorted(PodcastEpisode.objects.filter(feed__user=self.user).exclude(id__in=hidden)
|
|
.values_list('guid', flat=True))
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
archived = self.diora_feed('https://example.org/archived')
|
|
self.ap_feed = self.ap.feed('https://example.org/archived', keep_updated=0)
|
|
self.ap.item(self.ap_feed, 'ap-new', read=-1)
|
|
self.episode(archived, 'ap-new', pub_date=T0)
|
|
self.episode(archived, 'unknown-old', pub_date=T0 + timedelta(days=3)) # AntennaPod never fetched it
|
|
self.episode(archived, 'unknown-late', pub_date=BACKUP_DATE + timedelta(days=1))
|
|
self.episode(archived, 'undated')
|
|
self.ap.feed('https://example.org/kept')
|
|
kept = self.diora_feed('https://example.org/kept')
|
|
self.episode(kept, 'kept-unknown', pub_date=T0) # not archived: stays
|
|
only = self.diora_feed('https://example.org/only-diora')
|
|
self.episode(only, 'only-old', pub_date=T0)
|
|
self.episode(only, 'only-late', pub_date=BACKUP_DATE + timedelta(days=1))
|
|
|
|
def test_archived_podcasts(self):
|
|
out = self.run_import()
|
|
self.assertEqual(self.inbox(), ['ap-new', 'kept-unknown', 'only-late', 'only-old', 'unknown-late'])
|
|
self.assertIn('out of it: 2 of podcasts whose new episodes are ignored (AntennaPod never showed them)\n', out)
|
|
p = self.progress(PodcastEpisode.objects.get(guid='unknown-old'))
|
|
self.assertEqual((p.played, p.dismissed, p.updated_at), (False, True, BACKUP_DATE))
|
|
|
|
def test_quiet_diora_only(self):
|
|
out = self.run_import('--quiet-diora-only')
|
|
self.assertEqual(self.inbox(), ['ap-new', 'kept-unknown', 'only-late', 'unknown-late'])
|
|
self.assertIn(', 1 of podcasts only diora has', out)
|
|
|
|
def test_replace_is_a_clean_slate(self):
|
|
archived = PodcastFeed.objects.get(rss_url='https://example.org/archived')
|
|
ap_new = PodcastEpisode.objects.get(guid='ap-new')
|
|
old = PodcastEpisode.objects.get(guid='unknown-old')
|
|
late = PodcastEpisode.objects.get(guid='unknown-late')
|
|
PodcastQueue.objects.create(user=self.user, episode=late, position=0)
|
|
other = PodcastEpisode.objects.get(guid='kept-unknown')
|
|
PodcastQueue.objects.create(user=self.user, episode=other, position=1)
|
|
EpisodeProgress.objects.create(user=self.user, episode=old, position_seconds=30) # changed in diora now
|
|
queued = self.ap.item(self.ap_feed, 'unknown-late', read=0, pub=None)
|
|
self.ap.queue(queued)
|
|
out = self.run_import('--replace')
|
|
self.assertFalse(PodcastFeed.objects.filter(rss_url='https://example.org/only-diora').exists())
|
|
self.assertFalse(PodcastEpisode.objects.filter(guid__startswith='only-').exists())
|
|
self.assertEqual(self.queue(), [late.id])
|
|
self.assertTrue(self.progress(old).dismissed) # AntennaPod wins
|
|
self.assertIn('ap-new', self.inbox())
|
|
self.assertTrue(archived.episodes.filter(id=ap_new.id).exists())
|
|
self.assertIn(', 1 only in diora unsubscribed (--replace)', out)
|
|
self.assertIn(', 1 removed (--replace))', out)
|
|
self.assertIn('Unsubscribed feeds (1):\n - Diora\n', out)
|
|
|
|
def test_diora_newer_stays(self):
|
|
ep = PodcastEpisode.objects.get(guid='unknown-old')
|
|
EpisodeProgress.objects.create(user=self.user, episode=ep, position_seconds=30) # updated now
|
|
self.run_import()
|
|
self.assertIn('unknown-old', self.inbox())
|
|
|
|
|
|
class AfterFirstRefreshTests(ImportTestCase):
|
|
"""The way to move: import, let the new feeds be fetched, import again."""
|
|
|
|
def test_second_import_after_the_first_fetch(self):
|
|
import feedparser
|
|
|
|
from .views import FeedFetch, _refresh_feed
|
|
|
|
ap_feed = self.ap.feed('https://new.example.org/rss')
|
|
days = {'old1': 1, 'old2': 2, 'new': 3, 'newest': 4}
|
|
for guid, read in (('old1', 0), ('old2', 1), ('new', -1), ('newest', 0)):
|
|
self.ap.item(ap_feed, guid, read=read, pub=T0 + timedelta(days=days[guid]))
|
|
self.run_import()
|
|
feed = PodcastFeed.objects.get(user=self.user)
|
|
self.assertEqual(sorted(feed.episodes.values_list('guid', flat=True)), ['new', 'old2'])
|
|
|
|
def date(d):
|
|
return (T0 + timedelta(days=d)).strftime('%a, %d %b %Y %H:%M:%S GMT')
|
|
|
|
items = ''.join(
|
|
f'<item><guid>{g}</guid><title>{g}</title><pubDate>{date(d)}</pubDate><enclosure url="https://cdn.example.org/{g}.mp3" type="audio/mpeg" length="1"/></item>'
|
|
for g, d in days.items())
|
|
rss = f'<?xml version="1.0"?><rss version="2.0"><channel><title>New</title>{items}</channel></rss>'
|
|
with patch('podcasts.views._fetch_feed', lambda url, **kw: FeedFetch(200, feedparser.parse(rss))):
|
|
_refresh_feed(feed)
|
|
self.assertEqual(feed.episodes.count(), 4) # nothing twice
|
|
|
|
def inbox():
|
|
hidden = EpisodeProgress.objects.filter(user=self.user, dismissed=True).values('episode_id')
|
|
return sorted(feed.episodes.exclude(id__in=hidden).values_list('guid', flat=True))
|
|
|
|
self.assertEqual(inbox(), ['new', 'newest']) # the first-fetch rule leaves the newest
|
|
out = self.run_import()
|
|
self.assertEqual(inbox(), ['new'])
|
|
self.assertIn('2 without a state matched to diora episodes', out)
|
|
self.assertIn('1 taken out of the inbox', out)
|
|
|
|
|
|
class CommandErrorTests(ImportTestCase):
|
|
def test_unknown_user_and_bad_files(self):
|
|
with self.assertRaisesMessage(CommandError, 'No such user'):
|
|
self.run_import(user='nobody')
|
|
with self.assertRaisesMessage(CommandError, 'No such file'):
|
|
call_command('import_antennapod', os.path.join(self.tmp.name, 'missing.db'), '--user', 'alice')
|
|
other = os.path.join(self.tmp.name, 'other.db')
|
|
sqlite3.connect(other).close()
|
|
with self.assertRaisesMessage(CommandError, 'not an AntennaPod backup'):
|
|
call_command('import_antennapod', other, '--user', 'alice')
|
|
text = os.path.join(self.tmp.name, 'text.db')
|
|
with open(text, 'w') as f:
|
|
f.write('not a database at all, just text' * 10)
|
|
with self.assertRaisesMessage(CommandError, 'not'):
|
|
call_command('import_antennapod', text, '--user', 'alice')
|
|
with self.assertRaisesMessage(CommandError, 'Invalid --backup-date'):
|
|
call_command('import_antennapod', self.ap.path, '--user', 'alice', '--backup-date', 'gestern')
|