565 lines
26 KiB
Python
565 lines
26 KiB
Python
"""Regression tests for the QotNews API server.
|
|
|
|
Run unit tests from this directory with ``python tests.py`` (or
|
|
``python -m unittest``). They mock network calls and use an in-memory database.
|
|
|
|
Live integrations are opt-in: run ``QOTNEWS_RUN_INTEGRATION=1 python tests.py``
|
|
to call configured services and public feed APIs. The Meilisearch integration
|
|
test creates and removes a temporary ``qotnews_test`` index.
|
|
"""
|
|
|
|
import contextlib
|
|
import datetime
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
# The application modules use top-level imports (``import settings``), so make
|
|
# this file runnable both from apiserver/ and from the repository root.
|
|
APISERVER_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
if APISERVER_DIR not in sys.path:
|
|
sys.path.insert(0, APISERVER_DIR)
|
|
|
|
import database
|
|
import feed
|
|
import search
|
|
import utils
|
|
from feeds import hackernews, lobsters, manual, reddit, tildes
|
|
|
|
|
|
class UtilityTests(unittest.TestCase):
|
|
def test_generated_ids_have_expected_length_and_alphabet(self):
|
|
value = utils.gen_rand_id()
|
|
self.assertEqual(len(value), utils.NUM_ID_CHARS)
|
|
self.assertTrue(value.isalpha())
|
|
self.assertTrue(value.isupper())
|
|
|
|
def test_render_markdown_handles_empty_and_formatted_input(self):
|
|
self.assertEqual(utils.render_md(''), '')
|
|
self.assertIn('<strong>bold</strong>', utils.render_md('**bold**'))
|
|
|
|
def test_clean_removes_disallowed_markup(self):
|
|
cleaned = utils.clean('<p>safe</p><script>alert(1)</script>')
|
|
self.assertIn('<p>safe</p>', cleaned)
|
|
self.assertNotIn('<script>', cleaned)
|
|
|
|
|
|
class DatabaseTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
self.old_engine = database.engine
|
|
self.old_session = database.Session
|
|
self.engine = create_engine('sqlite:///:memory:')
|
|
database.engine = self.engine
|
|
database.Session = sessionmaker(bind=self.engine)
|
|
database.init()
|
|
|
|
def tearDown(self):
|
|
database.Session = self.old_session
|
|
database.engine = self.old_engine
|
|
self.engine.dispose()
|
|
|
|
def test_put_and_get_story_preserves_full_and_metadata_json(self):
|
|
story = {
|
|
'id': 'ABCD', 'ref': 'ref-1', 'title': 'Test story',
|
|
'source': 'manual', 'text': '<p>article</p>', 'comments': [{'text': 'comment'}],
|
|
}
|
|
database.put_story(story)
|
|
|
|
stored = database.get_story('ABCD')
|
|
self.assertEqual(stored.ref, 'ref-1')
|
|
self.assertEqual(json.loads(stored.full_json)['text'], '<p>article</p>')
|
|
metadata = json.loads(stored.meta_json)
|
|
self.assertEqual(metadata['title'], 'Test story')
|
|
self.assertNotIn('text', metadata)
|
|
self.assertNotIn('comments', metadata)
|
|
|
|
def test_put_story_updates_existing_record(self):
|
|
database.put_story({'id': 'ABCD', 'ref': 'ref-1', 'title': 'Old title'})
|
|
database.put_story({'id': 'ABCD', 'ref': 'ref-1', 'title': 'New title'})
|
|
|
|
self.assertEqual(database.get_story('ABCD').title, 'New title')
|
|
self.assertEqual(database.count_stories(), 1)
|
|
|
|
def test_reflist_orders_newest_first_and_stories_exclude_untitled(self):
|
|
database.put_story({'id': 'AAAA', 'ref': 'one', 'title': 'First'})
|
|
database.put_story({'id': 'BBBB', 'ref': 'two'})
|
|
database.put_story({'id': 'CCCC', 'ref': 'three', 'title': 'Third'})
|
|
database.put_ref('one', 'AAAA', 'hackernews')
|
|
database.put_ref('two', 'BBBB', 'reddit')
|
|
database.put_ref('three', 'CCCC', 'lobsters')
|
|
|
|
self.assertEqual([x['ref'] for x in database.get_reflist(10)], ['three', 'two', 'one'])
|
|
stories = [json.loads(item) for item in database.get_stories(10)]
|
|
self.assertEqual([item['title'] for item in stories], ['Third', 'First'])
|
|
|
|
def test_delete_ref_and_lookup_by_ref(self):
|
|
database.put_story({'id': 'ABCD', 'ref': 'ref-1', 'title': 'Test'})
|
|
database.put_ref('ref-1', 'ABCD', 'manual')
|
|
|
|
self.assertEqual(database.get_story_by_ref('ref-1').sid, 'ABCD')
|
|
database.del_ref('ref-1')
|
|
self.assertEqual(database.get_reflist(10), [])
|
|
# Deleting a feed reference does not delete the story itself.
|
|
self.assertIsNotNone(database.get_story('ABCD'))
|
|
|
|
|
|
class FeedTests(unittest.TestCase):
|
|
def test_list_combines_enabled_sources_in_configured_order(self):
|
|
with mock.patch.multiple(
|
|
feed.settings,
|
|
NUM_HACKERNEWS=2, NUM_LOBSTERS=1, NUM_REDDIT=1, NUM_TILDES=1,
|
|
), mock.patch.object(feed.hackernews, 'feed', return_value=['h1', 'h2']), \
|
|
mock.patch.object(feed.lobsters, 'feed', return_value=['l1']), \
|
|
mock.patch.object(feed.reddit, 'feed', return_value=['r1']), \
|
|
mock.patch.object(feed.tildes, 'feed', return_value=['t1']):
|
|
self.assertEqual(feed.list(), [
|
|
('h1', 'hackernews'), ('h2', 'hackernews'), ('l1', 'lobsters'),
|
|
('r1', 'reddit'), ('t1', 'tildes'),
|
|
])
|
|
|
|
def test_update_story_dispatches_and_merges_source_data(self):
|
|
story = {'id': 'ABCD', 'ref': '123', 'source': 'manual'}
|
|
source_data = {'title': 'A title', 'date': int(time.time()), 'url': '', 'text': 'body'}
|
|
with mock.patch.object(feed.manual, 'story', return_value=source_data):
|
|
self.assertTrue(feed.update_story(story))
|
|
self.assertEqual(story['title'], 'A title')
|
|
self.assertEqual(story['text'], 'body')
|
|
|
|
def test_update_story_rejects_old_non_manual_articles(self):
|
|
story = {'ref': '123', 'source': 'manual'}
|
|
source_data = {'title': 'Old', 'date': int(time.time()) - feed.TWO_DAYS - 1, 'url': ''}
|
|
with mock.patch.object(feed.manual, 'story', return_value=source_data):
|
|
self.assertFalse(feed.update_story(story))
|
|
|
|
def test_update_story_allows_old_manual_submissions(self):
|
|
story = {'ref': 'https://example.com', 'source': 'manual'}
|
|
source_data = {'title': 'Old', 'date': int(time.time()) - feed.TWO_DAYS - 1, 'url': ''}
|
|
with mock.patch.object(feed.manual, 'story', return_value=source_data):
|
|
self.assertTrue(feed.update_story(story, is_manual=True))
|
|
|
|
def test_update_story_rejects_non_html_content(self):
|
|
story = {'ref': '123', 'source': 'manual'}
|
|
source_data = {
|
|
'title': 'Document', 'date': int(time.time()),
|
|
'url': 'https://example.com/file.pdf',
|
|
}
|
|
with mock.patch.object(feed.manual, 'story', return_value=source_data), \
|
|
mock.patch.object(feed, 'get_content_type', return_value='application/pdf'):
|
|
self.assertFalse(feed.update_story(story))
|
|
|
|
def test_update_story_rejects_blocked_domains(self):
|
|
story = {'ref': '123', 'source': 'manual'}
|
|
source_data = {
|
|
'title': 'Article', 'date': int(time.time()),
|
|
'url': 'https://www.bloomberg.com/article',
|
|
}
|
|
with mock.patch.object(feed.manual, 'story', return_value=source_data), \
|
|
mock.patch.object(feed, 'get_content_type', return_value='text/html'):
|
|
self.assertFalse(feed.update_story(story))
|
|
|
|
def test_update_story_rejects_filtered_titles(self):
|
|
story = {'ref': '123', 'source': 'manual'}
|
|
source_data = {
|
|
'title': 'A Musk announcement', 'date': int(time.time()),
|
|
'url': 'https://example.com/article',
|
|
}
|
|
with mock.patch.object(feed.manual, 'story', return_value=source_data), \
|
|
mock.patch.object(feed, 'get_content_type', return_value='text/html'):
|
|
self.assertFalse(feed.update_story(story))
|
|
|
|
def test_update_story_returns_false_on_source_exception(self):
|
|
story = {'ref': '123', 'source': 'manual'}
|
|
with mock.patch.object(feed.manual, 'story', side_effect=ValueError('bad response')), \
|
|
mock.patch.object(feed.utils, 'alert_tanner'):
|
|
self.assertFalse(feed.update_story(story))
|
|
|
|
def test_get_article_posts_url_and_returns_reader_body_unchanged(self):
|
|
article_url = 'https://t0.vc/7/'
|
|
reader_body = (
|
|
'<div id="readability-page-1" class="page">'
|
|
'<p>Fake dog barking for home security while on vacation.</p>'
|
|
'<p>I set up a fake dog that barks if my surveillance cameras are triggered.</p>'
|
|
'</div>'
|
|
)
|
|
response = mock.Mock(status_code=200, text=reader_body)
|
|
with mock.patch.object(feed.requests, 'post', return_value=response) as post:
|
|
self.assertEqual(feed.get_article(article_url), reader_body)
|
|
post.assert_called_once_with(
|
|
feed.settings.READER_URL,
|
|
data={'url': article_url},
|
|
timeout=20,
|
|
)
|
|
|
|
|
|
class ParserTests(unittest.TestCase):
|
|
def test_hackernews_alg_comment_parses_and_counts_nested_comments(self):
|
|
item = {
|
|
'author': 'parent', 'points': 4, 'created_at_i': 100,
|
|
'text': '<p>hello</p>',
|
|
'children': [{'author': 'child', 'points': 2, 'created_at_i': 101,
|
|
'text': 'reply', 'children': []}],
|
|
}
|
|
parsed = hackernews.alg_comment(item)
|
|
self.assertEqual(parsed['author'], 'parent')
|
|
self.assertEqual(len(parsed['comments']), 1)
|
|
self.assertEqual(hackernews.alg_comment_count({'author': '', 'comments': [parsed]}), 2)
|
|
|
|
def test_hackernews_story_parses_algolia_fields(self):
|
|
response = {
|
|
'type': 'story', 'author': 'alice', 'points': 30,
|
|
'created_at_i': 123, 'title': 'HN story', 'url': 'https://example.com',
|
|
'children': [],
|
|
}
|
|
with mock.patch.object(hackernews, 'api', return_value=response):
|
|
parsed = hackernews.alg_story('42')
|
|
self.assertEqual(parsed['title'], 'HN story')
|
|
self.assertEqual(parsed['link'], 'https://news.ycombinator.com/item?id=42')
|
|
self.assertEqual(parsed['num_comments'], 0)
|
|
|
|
def test_hackernews_item_32250487_parses_real_story_and_comment_data(self):
|
|
response = {
|
|
'id': 32250487,
|
|
'type': 'story',
|
|
'author': 'tannercollin',
|
|
'points': 318,
|
|
'created_at_i': 1658930484,
|
|
'title': 'Fake Dog for Home Security',
|
|
'url': 'https://t0.vc/7/',
|
|
'children': [{
|
|
'author': 'CoffeeOnWrite',
|
|
'points': 0,
|
|
'created_at_i': 1658931140,
|
|
'text': 'A fake dog can scare off opportunistic pirates.',
|
|
'children': [],
|
|
}],
|
|
}
|
|
with mock.patch.object(hackernews, 'api', return_value=response) as api_call:
|
|
parsed = hackernews.story('32250487')
|
|
|
|
api_call.assert_called_once_with(hackernews.ALG_API_ITEM, '32250487')
|
|
self.assertEqual(parsed['title'], 'Fake Dog for Home Security')
|
|
self.assertEqual(parsed['author'], 'tannercollin')
|
|
self.assertEqual(parsed['score'], 318)
|
|
self.assertEqual(parsed['date'], 1658930484)
|
|
self.assertEqual(parsed['url'], 'https://t0.vc/7/')
|
|
self.assertEqual(parsed['num_comments'], 1)
|
|
self.assertEqual(parsed['comments'][0]['author'], 'CoffeeOnWrite')
|
|
|
|
def test_hackernews_story_rejects_non_story_and_low_engagement(self):
|
|
with mock.patch.object(hackernews, 'api', return_value={'type': 'comment'}):
|
|
self.assertFalse(hackernews.alg_story('42'))
|
|
low_engagement = {
|
|
'type': 'story', 'author': 'alice', 'points': 1, 'created_at_i': 123,
|
|
'title': 'Quiet story', 'url': '', 'children': [],
|
|
}
|
|
with mock.patch.object(hackernews, 'api', return_value=low_engagement):
|
|
self.assertFalse(hackernews.story('42'))
|
|
|
|
def test_lobsters_unix_parses_timezone_aware_timestamp(self):
|
|
timestamp = lobsters.unix('2020-01-02T03:04:05.000-0500')
|
|
expected = int(datetime.datetime(2020, 1, 2, 8, 4, 5, tzinfo=datetime.timezone.utc).timestamp())
|
|
self.assertEqual(timestamp, expected)
|
|
|
|
def test_lobsters_comments_are_nested_by_depth(self):
|
|
comments = [
|
|
{'depth': 0, 'comment': 'root', 'created_at': '2020-01-02T03:04:05.000+0000'},
|
|
{'depth': 1, 'comment': 'reply', 'created_at': '2020-01-02T03:04:06.000+0000'},
|
|
{'depth': 0, 'comment': 'another root', 'created_at': '2020-01-02T03:04:07.000+0000'},
|
|
]
|
|
parsed = lobsters.iter_comments(comments)
|
|
self.assertEqual(len(parsed), 2)
|
|
self.assertEqual(parsed[0]['comments'][0]['text'], 'reply')
|
|
self.assertEqual(parsed[1]['comments'], [])
|
|
|
|
def test_manual_story_extracts_title_from_html(self):
|
|
with mock.patch.object(manual, 'api', return_value='<html><title>Page title</title></html>'):
|
|
parsed = manual.story('https://example.com')
|
|
self.assertEqual(parsed['title'], 'Page title')
|
|
self.assertEqual(parsed['url'], 'https://example.com')
|
|
self.assertEqual(parsed['comments'], [])
|
|
|
|
|
|
class SearchTests(unittest.TestCase):
|
|
def test_meili_api_sends_bearer_token_and_decodes_json(self):
|
|
response = mock.Mock(status_code=200)
|
|
response.json.return_value = {'uid': 'qotnews'}
|
|
with mock.patch.object(search.settings, 'MEILI_URL', 'http://search/'), \
|
|
mock.patch.object(search.settings, 'MEILI_API_KEY', 'secret'), \
|
|
mock.patch.object(search.requests, 'post', return_value=response) as post:
|
|
result = search.meili_api(search.requests.post, 'indexes', json={'uid': 'qotnews'})
|
|
|
|
self.assertEqual(result, {'uid': 'qotnews'})
|
|
self.assertEqual(post.call_args[0][0], 'http://search/indexes')
|
|
self.assertEqual(post.call_args.kwargs['headers']['Authorization'], 'Bearer secret')
|
|
self.assertEqual(post.call_args.kwargs['timeout'], 4)
|
|
|
|
def test_meili_api_returns_false_for_http_error(self):
|
|
response = mock.Mock(status_code=500)
|
|
with mock.patch.object(search.requests, 'get', return_value=response):
|
|
self.assertFalse(search.meili_api(search.requests.get, 'health'))
|
|
|
|
def test_update_settings_uses_patch_and_supports_a_test_index(self):
|
|
with mock.patch.object(search, 'meili_api', return_value={'taskUid': 1}) as api_call:
|
|
result = search.update_settings('qotnews_test')
|
|
|
|
self.assertEqual(result, {'taskUid': 1})
|
|
self.assertEqual(api_call.call_args.args[0], search.requests.patch)
|
|
self.assertEqual(api_call.call_args.args[1], 'indexes/qotnews_test/settings')
|
|
settings_payload = api_call.call_args.kwargs['json']
|
|
self.assertEqual(settings_payload['searchableAttributes'], ['title', 'url', 'author'])
|
|
self.assertIn('id', settings_payload['displayedAttributes'])
|
|
self.assertEqual(settings_payload['rankingRules'][3], 'date:desc')
|
|
|
|
def test_search_forwards_query_and_returns_raw_response(self):
|
|
with mock.patch.object(search, 'meili_api', return_value='{"hits":[]}') as api_call:
|
|
result = search.search('python')
|
|
self.assertEqual(result, '{"hits":[]}')
|
|
self.assertEqual(api_call.call_args.args[1], 'indexes/qotnews/search')
|
|
self.assertEqual(api_call.call_args.kwargs['json']['q'], 'python')
|
|
self.assertFalse(api_call.call_args.kwargs['parse_json'])
|
|
|
|
|
|
class ServerRouteTests(unittest.TestCase):
|
|
"""Import the legacy server module with its runtime side effects disabled."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
import gevent.monkey
|
|
import gevent.pywsgi
|
|
|
|
cls.bootstrap_patches = [
|
|
mock.patch.object(gevent.monkey, 'patch_all'),
|
|
mock.patch.object(database, 'init'),
|
|
mock.patch.object(search, 'init'),
|
|
mock.patch('urllib.request.urlopen', side_effect=OSError('offline test')),
|
|
mock.patch('gevent.spawn'),
|
|
mock.patch.object(gevent.pywsgi, 'WSGIServer'),
|
|
]
|
|
for patcher in cls.bootstrap_patches:
|
|
patcher.start()
|
|
try:
|
|
import server
|
|
cls.server = server
|
|
except Exception:
|
|
for patcher in reversed(cls.bootstrap_patches):
|
|
patcher.stop()
|
|
raise
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
for patcher in reversed(cls.bootstrap_patches):
|
|
patcher.stop()
|
|
|
|
def setUp(self):
|
|
self.client = self.server.flask_app.test_client()
|
|
|
|
def test_api_returns_stories_as_json(self):
|
|
rows = [json.dumps({'id': 'ABCD', 'title': 'First'})]
|
|
with mock.patch.object(database, 'get_stories', return_value=rows) as get_stories:
|
|
response = self.client.get('/api')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.get_json(), {'stories': [{'id': 'ABCD', 'title': 'First'}]})
|
|
get_stories.assert_called_once_with(self.server.settings.FEED_LENGTH, 0)
|
|
|
|
def test_api_source_filter_pages_until_it_finds_matching_stories(self):
|
|
rows = [
|
|
json.dumps({'id': 'AAAA', 'source': 'reddit'}),
|
|
json.dumps({'id': 'BBBB', 'source': 'hackernews'}),
|
|
]
|
|
with mock.patch.object(database, 'get_stories', side_effect=[rows, []]) as get_stories:
|
|
response = self.client.get('/api?source=hackernews&limit=1')
|
|
self.assertEqual(response.get_json(), {'stories': [{'id': 'BBBB', 'source': 'hackernews'}]})
|
|
self.assertEqual(get_stories.call_count, 1)
|
|
|
|
def test_stats_route_reports_counts_and_id_space(self):
|
|
with mock.patch.object(database, 'count_stories', return_value=12):
|
|
response = self.client.get('/api/stats')
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.get_json()
|
|
self.assertEqual(data['total_stories'], 12)
|
|
self.assertEqual(data['id_space'], 26 ** utils.NUM_ID_CHARS)
|
|
self.assertEqual(data['len_ref_list'], 0)
|
|
|
|
def test_search_route_skips_short_queries(self):
|
|
with mock.patch.object(search, 'search', return_value='[{"id":"ABCD"}]') as do_search:
|
|
short = self.client.get('/api/search?q=ab')
|
|
long = self.client.get('/api/search?q=python')
|
|
self.assertEqual(short.get_json(), [])
|
|
self.assertEqual(long.get_json(), [{'id': 'ABCD'}])
|
|
do_search.assert_called_once_with('python')
|
|
|
|
def test_story_route_returns_full_story_and_404_for_missing_story(self):
|
|
full_json = json.dumps({'id': 'ABCD', 'title': 'A story'})
|
|
with mock.patch.object(database, 'get_story', return_value=SimpleNamespace(full_json=full_json)):
|
|
response = self.client.get('/api/ABCD')
|
|
self.assertEqual(response.get_json(), {'story': {'id': 'ABCD', 'title': 'A story'}})
|
|
with mock.patch.object(database, 'get_story', return_value=None):
|
|
missing = self.client.get('/api/ZZZZ')
|
|
self.assertEqual(missing.status_code, 404)
|
|
|
|
def test_submit_normalizes_url_and_returns_new_story_id(self):
|
|
def update(story, is_manual=False):
|
|
story.update({'title': 'Submitted', 'date': int(time.time()), 'url': '', 'text': 'body'})
|
|
return True
|
|
|
|
with mock.patch.object(self.server, 'new_id', return_value='ABCD'), \
|
|
mock.patch.object(database, 'get_story_by_ref', return_value=None), \
|
|
mock.patch.object(feed, 'update_story', side_effect=update) as update_story, \
|
|
mock.patch.object(database, 'put_story') as put_story, \
|
|
mock.patch.object(search, 'put_story') as index_story:
|
|
response = self.client.post('/api/submit', data={'url': 'example.com/article'})
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.get_json(), {'nid': 'ABCD'})
|
|
self.assertEqual(update_story.call_args.args[0]['ref'], 'http://example.com/article')
|
|
put_story.assert_called_once()
|
|
index_story.assert_called_once()
|
|
|
|
def test_submit_reuses_existing_story(self):
|
|
with mock.patch.object(self.server, 'new_id', return_value='NEW1'), \
|
|
mock.patch.object(database, 'get_story_by_ref', return_value=SimpleNamespace(sid='OLD1')):
|
|
response = self.client.post('/api/submit', data={'url': 'https://example.com/article'})
|
|
self.assertEqual(response.get_json(), {'nid': 'OLD1'})
|
|
|
|
def test_submit_returns_bad_request_for_invalid_submission(self):
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
response = self.client.post('/api/submit', data={})
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn('error', response.get_json())
|
|
|
|
|
|
RUN_INTEGRATION_TESTS = os.environ.get('QOTNEWS_RUN_INTEGRATION') == '1'
|
|
|
|
|
|
@unittest.skipUnless(
|
|
RUN_INTEGRATION_TESTS,
|
|
'Set QOTNEWS_RUN_INTEGRATION=1 to run tests against live services and public APIs.',
|
|
)
|
|
class LiveIntegrationTests(unittest.TestCase):
|
|
"""Read-only checks against configured services and upstream feed APIs."""
|
|
|
|
def test_reader_server_reads_t0_article(self):
|
|
body = feed.get_article('https://t0.vc/7/')
|
|
self.assertTrue(body, 'Reader server returned an empty body')
|
|
self.assertIn('readability-page-1', body)
|
|
self.assertIn('Fake dog barking for home security while on vacation.', body)
|
|
self.assertIn('I set up a fake dog that barks', body)
|
|
|
|
def test_hackernews_firebase_topstories(self):
|
|
story_ids = hackernews.feed()
|
|
self.assertTrue(story_ids)
|
|
self.assertTrue(all(story_id.isdigit() for story_id in story_ids[:10]))
|
|
|
|
def test_algolia_parses_hackernews_item_32250487(self):
|
|
story = hackernews.story('32250487')
|
|
self.assertIsInstance(story, dict)
|
|
self.assertEqual(story['title'], 'Fake Dog for Home Security')
|
|
self.assertEqual(story['author'], 'tannercollin')
|
|
self.assertEqual(story['url'].rstrip('/'), 'https://t0.vc/7')
|
|
self.assertGreater(story['num_comments'], 0)
|
|
|
|
def test_reddit_hot_feed_and_story(self):
|
|
story_ids = reddit.feed()
|
|
self.assertTrue(story_ids)
|
|
story = reddit.story(story_ids[0])
|
|
self.assertIsInstance(story, dict)
|
|
self.assertTrue(story['title'])
|
|
self.assertIn('reddit.com', story['link'])
|
|
|
|
def test_lobsters_hottest_feed_and_story(self):
|
|
story_ids = lobsters.feed()
|
|
self.assertTrue(story_ids)
|
|
story = lobsters.story(story_ids[0])
|
|
self.assertIsInstance(story, dict)
|
|
self.assertTrue(story['title'])
|
|
self.assertIn('lobste.rs/s/', story['link'])
|
|
|
|
def test_tildes_topic_listing(self):
|
|
topic_ids = tildes.feed()
|
|
self.assertTrue(topic_ids)
|
|
self.assertTrue(all(topic_id for topic_id in topic_ids[:10]))
|
|
|
|
def test_manual_parser_reads_t0_article(self):
|
|
story = manual.story('https://t0.vc/7/')
|
|
self.assertIsInstance(story, dict)
|
|
self.assertIn('Fake Dog for Home Security', story['title'])
|
|
self.assertEqual(story['url'], 'https://t0.vc/7/')
|
|
|
|
def test_meilisearch_temporary_index_search_and_cleanup(self):
|
|
index_uid = 'qotnews_test'
|
|
index_url = search.settings.MEILI_URL.rstrip('/') + '/indexes/' + index_uid
|
|
headers = {'Authorization': 'Bearer ' + search.settings.MEILI_API_KEY}
|
|
|
|
existing = search.requests.get(index_url, headers=headers, timeout=5)
|
|
if existing.status_code == 200:
|
|
self.fail('Refusing to overwrite or delete the existing qotnews_test index')
|
|
self.assertEqual(existing.status_code, 404, existing.text)
|
|
|
|
create_task_uid = None
|
|
try:
|
|
created = search.create_index(index_uid)
|
|
self.assertIsInstance(created, dict)
|
|
create_task_uid = created.get('taskUid')
|
|
self.assertIsNotNone(create_task_uid, created)
|
|
self._wait_for_meili_task(create_task_uid)
|
|
|
|
settings_task = search.update_settings(index_uid)
|
|
self.assertIsInstance(settings_task, dict)
|
|
self._wait_for_meili_task(settings_task['taskUid'])
|
|
|
|
document = {
|
|
'id': 'qotnews_test_doc',
|
|
'title': 'qotnews integration unique marker',
|
|
'url': 'https://example.com/integration-test',
|
|
'author': 'integration-test',
|
|
'text': 'this field should not be searchable or displayed',
|
|
}
|
|
add_task = search.put_story(document, index_uid)
|
|
self.assertIsInstance(add_task, dict)
|
|
self._wait_for_meili_task(add_task['taskUid'])
|
|
|
|
raw_results = search.search('qotnews integration unique marker', index_uid)
|
|
self.assertIsInstance(raw_results, str)
|
|
results = json.loads(raw_results)
|
|
self.assertEqual(results['hits'][0]['id'], document['id'])
|
|
self.assertEqual(results['hits'][0]['title'], document['title'])
|
|
self.assertNotIn('text', results['hits'][0])
|
|
finally:
|
|
# Even failed assertions should not leave test documents or indexes behind.
|
|
current = search.requests.get(index_url, headers=headers, timeout=5)
|
|
if current.status_code == 200:
|
|
deleted = search.meili_api(search.requests.delete, 'indexes/' + index_uid)
|
|
self.assertIsInstance(deleted, dict)
|
|
self.assertIn('taskUid', deleted)
|
|
self._wait_for_meili_task(deleted['taskUid'])
|
|
elif current.status_code != 404:
|
|
self.fail('Could not verify test index before cleanup: {}'.format(current.text))
|
|
|
|
after_cleanup = search.requests.get(index_url, headers=headers, timeout=5)
|
|
self.assertEqual(after_cleanup.status_code, 404, after_cleanup.text)
|
|
|
|
def _wait_for_meili_task(self, task_uid, timeout=30):
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
task = search.meili_api(search.requests.get, 'tasks/{}'.format(task_uid))
|
|
self.assertIsInstance(task, dict, 'Could not read Meilisearch task {}'.format(task_uid))
|
|
status = task.get('status')
|
|
if status in ('succeeded', 'completed'):
|
|
return task
|
|
if status in ('failed', 'canceled'):
|
|
self.fail('Meilisearch task {} {}: {}'.format(task_uid, status, task))
|
|
time.sleep(0.25)
|
|
self.fail('Timed out waiting for Meilisearch task {}'.format(task_uid))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|