mrwnslz/diora-web
1
0
Fork 0
Code Issues 4 Pull requests Projects Releases Packages Wiki Activity Actions
diora-web/podcasts/test_antennapod.py
mrwnslz 5a2841e7a6 AntennaPod-Import und automatischer Download pro Podcast 
- Neuer Befehl `manage.py import_antennapod <backup.db> --user <name> [--dry-run]
  [--global-auto-download on|off] [--backup-date ...]`: übernimmt aus einer
  AntennaPod-Datenbanksicherung die Abos (Abgleich über normalisierte URL,
  fehlende werden angelegt, lokale übersprungen), die Podcast-Einstellungen
  (Tempo, Intro/Outro, Lautstärke, Filter mit Phrasen in Anführungszeichen,
  Mindestlänge, neue Folgen, „nicht aktualisieren“ -> ignorieren, automatischer
  Download), den Zustand jeder Folge mit Zustand (gehört, neu, Position,
  Favorit, Zeitpunkt), die Warteschlange (AntennaPods Reihenfolge zuerst) und
  nimmt bekannte, dort nicht neue Folgen aus „Neu“. Neuere Änderungen in diora
  bleiben, Favoriten werden vereinigt. Alles in einer Transaktion; ein zweiter
  Lauf ändert nichts mehr.
- PodcastFeed.auto_download (Einstellung pro Podcast, in feeds/, state/ und
  feeds/<id>/settings/) und PodcastSettings.auto_download_default (in settings/
  und state/), mit dem neue Abos (Hinzufügen, OPML, gpodder, Import) starten.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-02 11:41:04 +02:00

501 lines
28 KiB
Python
Raw Blame History

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, state INTEGER DEFAULT 0, tags TEXT);
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);
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):
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) 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))
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.0, 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_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 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')
Reference in a new issue View git blame Copy permalink