diora-web/feeds/tests.py

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, refresh
from .models import FeedSettings, Item, 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 &amp; 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')):
            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_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(), 60)

    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(b'text="Blogs"', r.content)

    def test_settings(self):
        self.post('/feeds/settings/', {'news_keep_days': 5})
        self.assertEqual(FeedSettings.objects.get(user=self.u).news_keep_days, 5)