"""Unit tests for pages.py.""" from unittest.mock import patch from flask import get_flashed_messages from granary import as2, atom, microformats2, rss from granary.tests.test_bluesky import REPLY_BSKY from granary.tests.test_as1 import ( ACTOR, COMMENT, FOLLOW_WITH_ACTOR, FOLLOW_WITH_OBJECT, LIKE, NOTE, ) from oauth_dropins.webutil import util from oauth_dropins.webutil.testutil import requests_response from flask_app import app import common from models import Object, Follower, User from . import testutil from .test_webmention import ACTOR_AS2, ACTOR_HTML, ACTOR_MF2, REPOST_AS2 def contents(activities): return [(a.get('object') or a)['content'] for a in activities] class PagesTest(testutil.TestCase): EXPECTED = contents([COMMENT, NOTE]) def setUp(self): super().setUp() self.user = self.make_user('user.com') def test_user(self): got = self.client.get('/user/user.com') self.assert_equals(200, got.status_code) def test_user_objects(self): self.add_objects() got = self.client.get('/user/user.com') self.assert_equals(200, got.status_code) def test_user_not_found(self): got = self.client.get('/user/bar.com') self.assert_equals(404, got.status_code) def test_user_use_instead(self): bar = self.make_user('bar.com') bar.use_instead = self.user.key bar.put() got = self.client.get('/user/bar.com') self.assert_equals(301, got.status_code) self.assert_equals('/user/user.com', got.headers['Location']) def test_user_object_bare_string_id(self): with self.request_context: Object(id='a', domains=['user.com'], labels=['notification'], as2=REPOST_AS2).put() got = self.client.get('/user/user.com') self.assert_equals(200, got.status_code) def test_user_object_url_object(self): with self.request_context: Object(id='a', domains=['user.com'], labels=['notification'], our_as1={ **REPOST_AS2, 'object': { 'id': 'https://mas.to/toot/id', 'url': {'value': 'http://foo', 'displayName': 'bar'}, }, }).put() got = self.client.get('/user/user.com') self.assert_equals(200, got.status_code) @patch('requests.get') def test_check_web_site(self, mock_get): redir = 'http://localhost/.well-known/webfinger?resource=acct:user.com@user.com' mock_get.side_effect = ( requests_response('', status=302, redirected_url=redir), requests_response(ACTOR_HTML, url='https://user.com/', content_type=common.CONTENT_TYPE_HTML), ) got = self.client.post('/web-site', data={'url': 'https://user.com/'}) self.assert_equals(302, got.status_code) self.assert_equals('/user/user.com', got.headers['Location']) user = User.get_by_id('user.com') self.assertTrue(user.has_hcard) self.assertEqual('Person', user.actor_as2['type']) self.assertEqual('http://localhost/user.com', user.actor_as2['id']) def test_check_web_site_bad_url(self): got = self.client.post('/web-site', data={'url': '!!!'}) self.assert_equals(200, got.status_code) self.assertEqual(['No domain found in !!!'], get_flashed_messages()) self.assertEqual(1, User.query().count()) @patch('requests.get') def test_check_web_site_fetch_fails(self, mock_get): redir = 'http://localhost/.well-known/webfinger?resource=acct:orig@orig' mock_get.side_effect = ( requests_response('', status=302, redirected_url=redir), requests_response('', status=503), ) got = self.client.post('/web-site', data={'url': 'https://orig/'}) self.assert_equals(200, got.status_code, got.headers) self.assertTrue(get_flashed_messages()[0].startswith( "Couldn't connect to https://orig/: ")) def test_followers(self): self.make_user('bar.com') Follower.get_or_create('bar.com', 'https://no/stored/follow') Follower.get_or_create('bar.com', 'https://masto/user', last_follow=FOLLOW_WITH_ACTOR) got = self.client.get('/user/bar.com/followers') self.assert_equals(200, got.status_code) body = got.get_data(as_text=True) self.assertIn('no/stored/follow', body) self.assertIn('masto/user', body) def test_followers_empty(self): self.make_user('bar.com') got = self.client.get('/user/bar.com/followers') self.assert_equals(200, got.status_code) self.assertNotIn('class="follower', got.get_data(as_text=True)) def test_followers_user_not_found(self): got = self.client.get('/user/bar.com/followers') self.assert_equals(404, got.status_code) def test_following(self): Follower.get_or_create('https://no/stored/follow', 'bar.com') Follower.get_or_create('https://masto/user', 'bar.com', last_follow=FOLLOW_WITH_OBJECT) self.make_user('bar.com') got = self.client.get('/user/bar.com/following') self.assert_equals(200, got.status_code) body = got.get_data(as_text=True) self.assertIn('no/stored/follow', body) self.assertIn('masto/user', body) def test_following_empty(self): self.make_user('bar.com') got = self.client.get('/user/bar.com/following') self.assert_equals(200, got.status_code) self.assertNotIn('class="follower', got.get_data(as_text=True)) def test_following_user_not_found(self): got = self.client.get('/user/bar.com/following') self.assert_equals(404, got.status_code) def test_following_before_empty(self): self.make_user('bar.com') got = self.client.get(f'/user/bar.com/following?before={util.now().isoformat()}') self.assert_equals(200, got.status_code) def test_following_after_empty(self): self.make_user('bar.com') got = self.client.get(f'/user/bar.com/following?after={util.now().isoformat()}') self.assert_equals(200, got.status_code) def test_feed_user_not_found(self): got = self.client.get('/user/bar.com/feed') self.assert_equals(404, got.status_code) def test_feed_html_empty(self): got = self.client.get('/user/user.com/feed') self.assert_equals(200, got.status_code) self.assert_equals([], microformats2.html_to_activities(got.text)) def test_feed_html(self): self.add_objects() got = self.client.get('/user/user.com/feed') self.assert_equals(200, got.status_code) self.assert_equals(self.EXPECTED, contents(microformats2.html_to_activities(got.text))) def test_feed_atom_empty(self): got = self.client.get('/user/user.com/feed?format=atom') self.assert_equals(200, got.status_code) self.assert_equals([], atom.atom_to_activities(got.text)) def test_feed_atom(self): self.add_objects() got = self.client.get('/user/user.com/feed?format=atom') self.assert_equals(200, got.status_code) self.assert_equals(self.EXPECTED, contents(atom.atom_to_activities(got.text))) def test_feed_rss_empty(self): got = self.client.get('/user/user.com/feed?format=rss') self.assert_equals(200, got.status_code) self.assert_equals([], rss.to_activities(got.text)) def test_feed_rss(self): self.add_objects() got = self.client.get('/user/user.com/feed?format=rss') self.assert_equals(200, got.status_code) self.assert_equals(self.EXPECTED, contents(rss.to_activities(got.text)))