mrwnslz
74bb3a44df
- Source.notify; ein neuer Post wird beim Eintreffen markiert (Item.notify), nie beim ersten Abruf eines Blogs, nie älter als 2 Tage - GET /feeds/notifications/?after=<id>: ungelesene markierte Posts von Blogs mit Glocke nach der zuletzt gesehenen id; ohne after nur der Startpunkt - Glocken-Schalter in der Blog-Leiste und in den Einstellungen des Blogs; window.__dioraFeedsOpenItem öffnet einen Post (Antippen der Benachrichtigung) - sw v94 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
362 lines
20 KiB
Python
362 lines
20 KiB
Python
import json
|
||
from datetime import timedelta
|
||
from unittest import mock
|
||
|
||
import feedparser
|
||
from django.contrib.auth.models import User
|
||
from django.test import TestCase
|
||
from django.utils import timezone
|
||
|
||
from podcasts.views import FeedFetch
|
||
|
||
from . import cluster, fetch, logins, refresh
|
||
from .models import FeedSettings, Item, SiteLogin, Source, Story
|
||
|
||
|
||
def rss(items, title='Testfeed', link='https://example.org/'):
|
||
body = ''.join(
|
||
f'<item><guid>{i["guid"]}</guid><title>{i["title"]}</title><link>{i.get("link", "https://example.org/" + i["guid"])}</link>'
|
||
f'<pubDate>{i.get("date", timezone.now().strftime("%a, %d %b %Y %H:%M:%S +0000"))}</pubDate>'
|
||
f'<description><![CDATA[{i.get("text", "")}]]></description></item>'
|
||
for i in items)
|
||
return f'<?xml version="1.0"?><rss version="2.0"><channel><title>{title}</title><link>{link}</link>{body}</channel></rss>'.encode()
|
||
|
||
|
||
def fetched(content, url='https://example.org/feed'):
|
||
return FeedFetch(200, feedparser.parse(content), 'etag', '', url, url, 'hash')
|
||
|
||
|
||
LONG = '<p>' + 'Ein langer Absatz mit sehr viel Text. ' * 40 + '</p>'
|
||
|
||
|
||
class FetchTest(TestCase):
|
||
def test_relative_links_become_absolute_and_scripts_go(self):
|
||
out = fetch.clean_html('<p>Hi <a href="/a">x</a><img src="i.jpg"><script>bad()</script>'
|
||
'<img data-src="/real.jpg" src="placeholder.gif"></p>', 'https://ex.org/post/1')
|
||
self.assertIn('href="https://ex.org/a"', out)
|
||
self.assertIn('src="https://ex.org/post/i.jpg"', out)
|
||
self.assertIn('src="https://ex.org/real.jpg"', out)
|
||
self.assertNotIn('placeholder', out)
|
||
self.assertNotIn('bad()', out)
|
||
|
||
def test_parse_entry(self):
|
||
e = feedparser.parse(rss([{'guid': 'g1', 'title': 'A & B', 'text': '<p>Kurz <img src="/p.jpg"></p>'}])).entries[0]
|
||
f = fetch.parse_entry(e, 'https://example.org/feed')
|
||
self.assertEqual(f['title'], 'A & B')
|
||
self.assertEqual(f['image_url'], 'https://example.org/p.jpg')
|
||
self.assertTrue(fetch.is_teaser(f['content']))
|
||
|
||
def test_drop_repeated_headline(self):
|
||
self.assertEqual(fetch._drop_title('<img src="x"><h1>Titel</h1><p>Text</p>'), '<img src="x"><p>Text</p>')
|
||
|
||
def test_short_title(self):
|
||
self.assertEqual(fetch.short_title('tagesschau.de - die erste Adresse für Nachrichten'), 'tagesschau.de')
|
||
self.assertEqual(fetch.short_title('Aktuell - FAZ.NET'), 'FAZ.NET')
|
||
self.assertEqual(fetch.short_title('DIE ZEIT | Nachrichten, News, Hintergründe'), 'DIE ZEIT')
|
||
self.assertEqual(fetch.short_title('Golem.de'), 'Golem.de')
|
||
self.assertEqual(fetch.short_title('Kotlin-Blog'), 'Kotlin-Blog')
|
||
|
||
def test_discover(self):
|
||
feeds, icon = fetch.discover('<link rel="alternate" type="application/rss+xml" href="/feed" title="F">'
|
||
'<link rel="icon" href="/fav.png">', 'https://ex.org/')
|
||
self.assertEqual(feeds, [('https://ex.org/feed', 'F')])
|
||
self.assertEqual(icon, 'https://ex.org/fav.png')
|
||
|
||
|
||
class ClusterTest(TestCase):
|
||
def setUp(self):
|
||
self.u = User.objects.create_user('u', password='pw12345678')
|
||
self.a = Source.objects.create(user=self.u, url='https://a/', title='A')
|
||
self.b = Source.objects.create(user=self.u, url='https://b/', title='B')
|
||
self.c = Source.objects.create(user=self.u, url='https://c/', title='C')
|
||
|
||
def item(self, source, title, summary=''):
|
||
return Item.objects.create(source=source, user=self.u, kind='news', guid_hash=title[:60],
|
||
title=title, summary=summary, published=timezone.now())
|
||
|
||
def test_same_event_bundled_other_apart(self):
|
||
c = cluster.Clusterer(self.u, timezone.now())
|
||
s1 = c.add(self.item(self.a, 'Oktoberfest endet mit neuem Besucherrekord in München'))
|
||
s2 = c.add(self.item(self.b, 'Rekord-Wiesn: 7,4 Millionen Besucher auf dem Oktoberfest in München'))
|
||
s3 = c.add(self.item(self.c, 'Bahn führt Familienticket dauerhaft ein'))
|
||
self.assertEqual(s1, s2)
|
||
self.assertNotEqual(s1, s3)
|
||
s1.refresh_from_db()
|
||
self.assertEqual((s1.size, s1.sources), (2, 2))
|
||
|
||
def test_same_source_needs_to_be_very_close(self):
|
||
c = cluster.Clusterer(self.u, timezone.now())
|
||
s1 = c.add(self.item(self.a, 'Nations League: Pressekonferenz vor Irland gegen Israel abgebrochen'))
|
||
s2 = c.add(self.item(self.a, 'Nations League: England schlägt Kroatien deutlich'))
|
||
self.assertNotEqual(s1, s2)
|
||
|
||
|
||
class ApiTest(TestCase):
|
||
def setUp(self):
|
||
self.u = User.objects.create_user('tester', password='pw12345678')
|
||
self.client.force_login(self.u)
|
||
|
||
def post(self, url, data=None):
|
||
return self.client.post(url, json.dumps(data or {}), content_type='application/json')
|
||
|
||
def add(self, kind, feed, url='https://example.org/feed'):
|
||
with mock.patch('feeds.refresh._fetch_feed', return_value=fetched(feed, url)), \
|
||
mock.patch('feeds.fetch.fetch_article', return_value=(LONG, 'https://example.org/img.jpg', False)):
|
||
r = self.post('/feeds/sources/add/', {'url': url, 'kind': kind, 'category': 'Politik'})
|
||
self.assertEqual(r.status_code, 200)
|
||
return r.json()['source']
|
||
|
||
def test_requires_login(self):
|
||
self.client.logout()
|
||
self.assertEqual(self.client.get('/feeds/state/').status_code, 401)
|
||
|
||
def test_add_news_fetches_articles_and_bundles(self):
|
||
s = self.add('news', rss([{'guid': 'a', 'title': 'Merz besucht Kiew und sagt Hilfe zu', 'text': 'Kurz.'},
|
||
{'guid': 'b', 'title': 'Bahn Familienticket dauerhaft', 'text': 'Kurz.'}]))
|
||
self.assertEqual(s['title'], 'Testfeed')
|
||
self.assertEqual(s['unread'], 2)
|
||
items = Item.objects.filter(user=self.u)
|
||
self.assertTrue(all(i.content_full for i in items))
|
||
self.assertEqual(Story.objects.filter(user=self.u).count(), 2)
|
||
news = self.client.get('/feeds/news/').json()
|
||
self.assertEqual(len(news['stories']), 2)
|
||
self.assertEqual(news['stories'][0]['category'], 'Politik')
|
||
b = self.client.get('/feeds/briefing/').json()
|
||
self.assertEqual(b['total'], 2)
|
||
self.assertEqual(b['sections'][0]['category'], 'Politik')
|
||
self.post('/feeds/briefing/seen/')
|
||
self.assertFalse(Item.objects.filter(user=self.u, read_at=None).exists())
|
||
self.assertEqual(self.client.get('/feeds/briefing/').json()['total'], 0)
|
||
|
||
def test_blog_keeps_full_feed_text_and_lists(self):
|
||
self.add('blog', rss([{'guid': 'p1', 'title': 'Ein Artikel', 'text': LONG}]))
|
||
item = Item.objects.get(user=self.u)
|
||
self.assertFalse(item.content_full) # the feed had the whole text
|
||
self.assertIsNone(item.story)
|
||
blogs = self.client.get('/feeds/blogs/?unread=1').json()
|
||
self.assertEqual([i['title'] for i in blogs['items']], ['Ein Artikel'])
|
||
detail = self.client.get(f'/feeds/items/{item.id}/').json()['item']
|
||
self.assertIn('langer Absatz', detail['content'])
|
||
self.post('/feeds/items/read/', {'ids': [item.id]})
|
||
self.assertEqual(self.client.get('/feeds/blogs/?unread=1').json()['items'], [])
|
||
self.post(f'/feeds/items/{item.id}/star/', {'starred': True})
|
||
self.assertEqual(len(self.client.get('/feeds/starred/').json()['items']), 1)
|
||
|
||
def test_blog_folders(self):
|
||
with mock.patch('feeds.refresh._fetch_feed', return_value=fetched(rss([{'guid': 'x', 'title': 'Essay', 'text': LONG}]))):
|
||
self.post('/feeds/sources/add/', {'url': 'https://a.example/feed', 'kind': 'blog', 'category': 'Philosophie'})
|
||
with mock.patch('feeds.refresh._fetch_feed', return_value=fetched(rss([{'guid': 'y', 'title': 'Code', 'text': LONG}]))):
|
||
self.post('/feeds/sources/add/', {'url': 'https://b.example/feed', 'kind': 'blog', 'category': ''})
|
||
self.assertEqual(self.client.get('/feeds/state/').json()['folders'], ['Philosophie'])
|
||
self.assertEqual([i['title'] for i in self.client.get('/feeds/blogs/?folder=Philosophie').json()['items']], ['Essay'])
|
||
self.assertEqual([i['title'] for i in self.client.get('/feeds/blogs/?folder=').json()['items']], ['Code'])
|
||
self.assertIn('text="Blogs – Philosophie"', self.client.get('/feeds/export.opml').content.decode())
|
||
|
||
def test_old_news_expires_and_is_not_taken_in_again(self):
|
||
old = (timezone.now() - timedelta(days=3)).strftime('%a, %d %b %Y %H:%M:%S +0000')
|
||
feed = rss([{'guid': 'new', 'title': 'Heute passiert', 'text': LONG},
|
||
{'guid': 'old', 'title': 'Vorgestern passiert', 'text': LONG, 'date': old}])
|
||
src = self.add('news', feed)
|
||
self.assertEqual(Item.objects.filter(user=self.u).count(), 1) # too old to take in
|
||
Item.objects.filter(user=self.u).update(published=timezone.now() - timedelta(days=3))
|
||
refresh.expire(self.u)
|
||
self.assertEqual(Item.objects.filter(user=self.u, gone=False).count(), 0)
|
||
self.assertEqual(Story.objects.filter(user=self.u).count(), 0)
|
||
source = Source.objects.get(pk=src['id'])
|
||
with mock.patch('feeds.refresh._fetch_feed', return_value=fetched(feed)):
|
||
refresh.refresh([source])
|
||
self.assertEqual(Item.objects.filter(user=self.u, gone=False).count(), 0)
|
||
|
||
def test_feed_wide_timestamp_means_first_seen(self):
|
||
stamp = (timezone.now() - timedelta(hours=5)).strftime('%a, %d %b %Y %H:%M:%S +0000')
|
||
self.add('news', rss([{'guid': f'g{n}', 'title': f'Meldung {n}', 'text': LONG, 'date': stamp} for n in range(6)]))
|
||
for i in Item.objects.filter(user=self.u):
|
||
self.assertLess((timezone.now() - i.published).total_seconds(), 600) # first fetch: feed order, minutes apart
|
||
|
||
def test_new_blog_without_dates_keeps_order_and_few_unread(self):
|
||
feed = (b'<?xml version="1.0"?><rss version="2.0"><channel><title>Undatiert</title>'
|
||
+ b''.join(b'<item><guid>u%d</guid><title>Post %d</title><description>x</description></item>' % (n, n)
|
||
for n in range(6)) + b'</channel></rss>')
|
||
self.add('blog', feed)
|
||
items = list(Item.objects.filter(user=self.u).order_by('-published'))
|
||
self.assertEqual([i.title for i in items], [f'Post {n}' for n in range(6)])
|
||
self.assertEqual(sum(1 for i in items if i.read_at is None), 3)
|
||
|
||
def test_switching_kind_moves_items(self):
|
||
src = self.add('news', rss([{'guid': 'a', 'title': 'Etwas', 'text': LONG}]))
|
||
self.post(f'/feeds/sources/{src["id"]}/', {'kind': 'blog'})
|
||
item = Item.objects.get(user=self.u)
|
||
self.assertEqual((item.kind, item.story), ('blog', None))
|
||
self.assertEqual(Story.objects.filter(user=self.u).count(), 0)
|
||
|
||
def test_other_users_items_invisible(self):
|
||
self.add('blog', rss([{'guid': 'p1', 'title': 'Privat', 'text': LONG}]))
|
||
item = Item.objects.get(user=self.u)
|
||
other = User.objects.create_user('o', password='pw12345678')
|
||
self.client.force_login(other)
|
||
self.assertEqual(self.client.get(f'/feeds/items/{item.id}/').status_code, 404)
|
||
self.assertEqual(self.client.get('/feeds/blogs/').json()['items'], [])
|
||
self.post('/feeds/items/read/', {'ids': [item.id]})
|
||
item.refresh_from_db()
|
||
self.assertIsNone(item.read_at)
|
||
|
||
def test_opml_export(self):
|
||
self.add('blog', rss([]), url='https://blog.example/feed')
|
||
r = self.client.get('/feeds/export.opml')
|
||
self.assertIn(b'xmlUrl="https://blog.example/feed"', r.content)
|
||
self.assertIn('text="Blogs – Politik"', r.content.decode())
|
||
|
||
def test_settings(self):
|
||
self.post('/feeds/settings/', {'news_keep_days': 5})
|
||
self.assertEqual(FeedSettings.objects.get(user=self.u).news_keep_days, 5)
|
||
|
||
|
||
class FakeResponse:
|
||
def __init__(self, status=302, location='/', text='', url='https://www.jungewelt.de/x', ctype='text/html'):
|
||
self.status_code = status
|
||
self.headers = {'Location': location, 'Content-Type': ctype}
|
||
self.text = text
|
||
self.url = url
|
||
|
||
|
||
PAYWALL = ('<html><body><article><h1>T</h1><p>' + 'Freier Anfang des Artikels. ' * 12 + '</p></article>'
|
||
'<p>Unterstützen Sie die junge Welt mit einem Abonnement und erhalten Sie Zugang zu allen Analysen</p></body></html>')
|
||
FULL = '<html><body><article><h1>T</h1>' + ('<p>' + 'Der ganze Artikel für Abonnenten. ' * 15 + '</p>') * 4 + '</article></body></html>'
|
||
|
||
|
||
class SiteLoginTest(TestCase):
|
||
def setUp(self):
|
||
self.u = User.objects.create_user('tester', password='pw12345678')
|
||
self.client.force_login(self.u)
|
||
self.src = Source.objects.create(user=self.u, url='https://www.jungewelt.de/feeds/newsticker.rss',
|
||
title='junge Welt', kind='news')
|
||
self.item = Item.objects.create(source=self.src, user=self.u, kind='news', guid_hash='a', title='Artikel',
|
||
url='https://www.jungewelt.de/artikel/1.html', summary='Anriss',
|
||
content='<p>Anriss</p>', published=timezone.now())
|
||
logins._sessions.clear()
|
||
|
||
def post(self, url, data):
|
||
return self.client.post(url, json.dumps(data), content_type='application/json')
|
||
|
||
def test_encryption_round_trip_and_not_plain(self):
|
||
token = logins.encrypt('geheim')
|
||
self.assertNotIn('geheim', token)
|
||
self.assertEqual(logins.decrypt(token), 'geheim')
|
||
self.assertIsNone(logins.decrypt('kaputt'))
|
||
|
||
def test_paywalled_page_keeps_free_part_and_waits(self):
|
||
with mock.patch('requests.get', return_value=FakeResponse(200, text=PAYWALL)):
|
||
refresh.load_article(self.item)
|
||
self.item.refresh_from_db()
|
||
self.assertTrue(self.item.paywalled)
|
||
self.assertFalse(self.item.content_full)
|
||
self.assertIn('Freier Anfang', self.item.content)
|
||
self.assertFalse(refresh.needs_article(self.item, self.src)) # no login: don't try again
|
||
|
||
def test_wrong_login_is_refused_and_not_stored(self):
|
||
with mock.patch('requests.Session.post', return_value=FakeResponse(302, 'https://www.jungewelt.de/loginFailed.php?x')):
|
||
r = self.post('/feeds/logins/', {'source': self.src.id, 'username': 'u', 'password': 'falsch'})
|
||
self.assertEqual(r.status_code, 400)
|
||
self.assertIn('falsch', r.json()['error'])
|
||
self.assertFalse(SiteLogin.objects.exists())
|
||
|
||
def test_login_stores_encrypted_and_refetches_full_articles(self):
|
||
Item.objects.filter(pk=self.item.pk).update(paywalled=True)
|
||
|
||
def login_post(session, url, **kw):
|
||
session.cookies.set('jW', 'sitzung')
|
||
return FakeResponse(302, 'https://www.jungewelt.de/')
|
||
|
||
with mock.patch('requests.Session.post', autospec=True, side_effect=login_post), \
|
||
mock.patch('requests.Session.get', return_value=FakeResponse(200, text=FULL)):
|
||
r = self.post('/feeds/logins/', {'source': self.src.id, 'username': 'leser', 'password': 'abo-passwort'})
|
||
self.assertEqual(r.status_code, 200)
|
||
self.assertEqual(r.json()['refetched'], 1)
|
||
login = SiteLogin.objects.get(user=self.u)
|
||
self.assertNotIn('abo-passwort', login.secret)
|
||
self.item.refresh_from_db()
|
||
self.assertTrue(self.item.content_full)
|
||
self.assertFalse(self.item.paywalled)
|
||
self.assertIn('ganze Artikel', self.item.content)
|
||
state = self.client.get('/feeds/state/').json()
|
||
info = state['sources'][0]['login']
|
||
self.assertEqual((info['username'], info['ok'], info['site']), ('leser', True, 'junge Welt'))
|
||
self.assertNotIn('abo-passwort', json.dumps(state))
|
||
self.post('/feeds/logins/delete/', {'source': self.src.id})
|
||
self.assertFalse(SiteLogin.objects.exists())
|
||
|
||
def test_sources_of_other_sites_have_no_login(self):
|
||
other = Source.objects.create(user=self.u, url='https://blog.example/feed', title='B', kind='blog')
|
||
state = self.client.get('/feeds/state/').json()
|
||
by_id = {s['id']: s for s in state['sources']}
|
||
self.assertIsNone(by_id[other.id]['login'])
|
||
self.assertIsNotNone(by_id[self.src.id]['login'])
|
||
|
||
|
||
class BlogFeedTest(TestCase):
|
||
def setUp(self):
|
||
self.u = User.objects.create_user('tester', password='pw12345678')
|
||
self.client.force_login(self.u)
|
||
now = timezone.now()
|
||
self.busy = Source.objects.create(user=self.u, url='https://busy/', title='Busy', kind='blog', category='Politik')
|
||
self.rare = Source.objects.create(user=self.u, url='https://rare/', title='Rare', kind='blog', category='Politik')
|
||
for n in range(10): # a post every two hours
|
||
Item.objects.create(source=self.busy, user=self.u, kind='blog', guid_hash=f'b{n}', title=f'Busy {n}',
|
||
content='<p>' + 'wort ' * 460 + '</p>', words=460, published=now - timedelta(hours=2 * n))
|
||
Item.objects.create(source=self.rare, user=self.u, kind='blog', guid_hash='r', title='Rare essay',
|
||
published=now - timedelta(hours=5))
|
||
|
||
def post(self, url, data):
|
||
return self.client.post(url, json.dumps(data), content_type='application/json')
|
||
|
||
def by_id(self):
|
||
return {s['id']: s for s in self.client.get('/feeds/state/').json()['sources']}
|
||
|
||
def test_interval_new_and_words(self):
|
||
st = self.by_id()
|
||
self.assertLess(st[self.busy.id]['interval_days'], 1)
|
||
self.assertEqual(st[self.rare.id]['interval_days'], 30.0) # one post: counts as rare
|
||
self.assertEqual(st[self.busy.id]['new'], 10)
|
||
items = self.client.get('/feeds/blogs/?unread=1').json()['items']
|
||
self.assertEqual(next(i for i in items if i['title'] == 'Busy 0')['words'], 460)
|
||
|
||
def test_hidden_blog_only_in_its_folder_and_not_counted(self):
|
||
self.post(f'/feeds/sources/{self.busy.id}/', {'hidden': True})
|
||
state = self.client.get('/feeds/state/').json()
|
||
self.assertEqual((state['blogs_unread'], state['blogs_new']), (1, 1))
|
||
main = [i['title'] for i in self.client.get('/feeds/blogs/?unread=1').json()['items']]
|
||
self.assertEqual(main, ['Rare essay'])
|
||
folder = self.client.get('/feeds/blogs/?unread=1&folder=Politik').json()['items']
|
||
self.assertEqual(len(folder), 11)
|
||
|
||
def test_favorites_filter(self):
|
||
self.post(f'/feeds/sources/{self.rare.id}/', {'fav': True})
|
||
self.assertTrue(self.by_id()[self.rare.id]['fav'])
|
||
favs = [i['title'] for i in self.client.get('/feeds/blogs/?fav=1').json()['items']]
|
||
self.assertEqual(favs, ['Rare essay'])
|
||
|
||
|
||
class NotifyTest(TestCase):
|
||
def setUp(self):
|
||
self.u = User.objects.create_user('tester', password='pw12345678')
|
||
self.client.force_login(self.u)
|
||
self.src = Source.objects.create(user=self.u, url='https://example.org/feed', title='Blog', kind='blog', notify=True)
|
||
|
||
def fetch(self, items):
|
||
with mock.patch('feeds.refresh._fetch_feed', return_value=fetched(rss(items))):
|
||
refresh.refresh([self.src])
|
||
|
||
def test_rules(self):
|
||
old = (timezone.now() - timedelta(days=5)).strftime('%a, %d %b %Y %H:%M:%S +0000')
|
||
self.fetch([{'guid': 'a', 'title': 'Schon da', 'text': LONG}]) # first fetch: never
|
||
start = self.client.get('/feeds/notifications/').json()
|
||
self.assertEqual(start['items'], [])
|
||
self.fetch([{'guid': 'a', 'title': 'Schon da', 'text': LONG},
|
||
{'guid': 'b', 'title': 'Neuer Post', 'text': LONG},
|
||
{'guid': 'c', 'title': 'Alter Post', 'text': LONG, 'date': old}])
|
||
got = self.client.get(f"/feeds/notifications/?after={start['last']}").json()
|
||
self.assertEqual([i['title'] for i in got['items']], ['Neuer Post'])
|
||
# Seen up to there: nothing again; read or bell off: nothing either.
|
||
self.assertEqual(self.client.get(f"/feeds/notifications/?after={got['last']}").json()['items'], [])
|
||
Source.objects.filter(pk=self.src.pk).update(notify=False)
|
||
self.assertEqual(self.client.get(f"/feeds/notifications/?after={start['last']}").json()['items'], [])
|