Spaces:
Sleeping
Sleeping
File size: 5,439 Bytes
ff3c435 7e31451 ff3c435 a48e17c ff3c435 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | import time
import unittest
from unittest.mock import Mock, patch
from fastapi.testclient import TestClient
import service
from reports import normalize
class ServiceTests(unittest.TestCase):
def setUp(self):
self.client = TestClient(service.app)
self.key_patch = patch.object(service, 'API_KEY', 'test-bridge-key')
self.key_patch.start()
self.addCleanup(self.key_patch.stop)
self.queue_patch = patch.object(service, 'queue', Mock())
self.queue = self.queue_patch.start()
self.addCleanup(self.queue_patch.stop)
service.jobs.clear()
self.headers = {'X-Bridge-Key': 'test-bridge-key'}
self.payload = {'request_id': '12345678-abcd-efgh', 'links': ['http://vk.ru/wall-123_456']}
def test_authentication_and_production_route_absent(self):
self.assertEqual(self.client.post('/jobs', json=self.payload).status_code, 401)
self.assertEqual(self.client.post('/jobs', json=self.payload, headers={'X-Bridge-Key': 'wrong-key'}).status_code, 401)
self.assertEqual(self.client.get('/health').json()['service'], 'ReportBridge')
self.assertEqual(self.client.post('/parse', json={'links': []}).status_code, 404)
def test_idempotency_and_canonical_dedup(self):
self.payload['links'].append('https://vk.com/wall-123_456')
a = self.client.post('/jobs', json=self.payload, headers=self.headers)
b = self.client.post('/jobs', json=self.payload, headers=self.headers)
self.assertEqual(a.status_code, 200)
self.assertEqual(a.json()['id'], b.json()['id'])
self.queue.submit.assert_called_once()
job = self.client.get('/jobs/' + a.json()['id'], headers=self.headers).json()
self.assertEqual(job['total'], 1)
self.assertNotIn('links', job)
self.payload['links'] = ['https://t.me/channel/1']
self.assertEqual(self.client.post('/jobs', json=self.payload, headers=self.headers).status_code, 409)
def test_private_and_arbitrary_urls_never_enqueue(self):
for url in ['https://t.me/c/123/456', 'http://127.0.0.1/secret', 'https://vk.com.evil.com/wall-123_456']:
self.payload['links'] = [url]
self.assertEqual(self.client.post('/jobs', json=self.payload, headers=self.headers).status_code, 422)
self.queue.submit.assert_not_called()
def test_max_case_duplicates_do_not_enqueue_or_sum_the_same_post_twice(self):
links = ['https://max.ru/denisdorokhov1987/AaFtQ2-jWec',
'https://max.ru/DenisDorokhov1987/AaFtQ2-jWec',
'https://max.ru/vkontakte/AaE-viFyB3E']
payload = dict(self.payload, links=links)
response = self.client.post('/jobs', json=payload, headers=self.headers)
self.assertEqual(response.status_code, 200)
job_id = response.json()['id']
records = self.queue.submit.call_args.args[2]
self.assertEqual(len(records), 2)
self.assertEqual(records[0]['url'], links[0])
self.assertEqual(service.jobs[job_id]['total'], 2)
with patch.object(service, 'fetch_one', side_effect=lambda r: dict(r, title=r['group'], views=1000)):
service.run_job(job_id, records)
result = service.jobs[job_id]['result']
self.assertEqual(sum(r['views'] for r in result['records']), 2000)
self.assertNotIn('ещё</a>', result['html'])
# A change to the case-sensitive post ID is not treated as a duplicate.
self.assertNotEqual(normalize(links[0])['url'], normalize(links[0].replace('/AaFt', '/aaFt'))['url'])
def test_queue_limits(self):
for i in range(5):
self.payload['request_id'] = f'12345678-abcd-{i:06d}'
self.assertEqual(self.client.post('/jobs', json=self.payload, headers=self.headers).status_code, 200)
self.payload['request_id'] = '12345678-abcd-999999'
self.assertEqual(self.client.post('/jobs', json=self.payload, headers=self.headers).status_code, 429)
def test_real_progress_and_rendering(self):
records = [normalize('https://t.me/channel/1'), normalize('https://t.me/channel/2')]
service.jobs['test'] = dict(status='queued', completed=0, total=2, created=time.time())
with patch.object(service, 'fetch_one', side_effect=lambda r: dict(r, title='Имя', views=2000)):
service.run_job('test', records)
job = service.jobs['test']
self.assertEqual(job['status'], 'complete')
self.assertEqual(job['completed'], 2)
self.assertIn('<b>4K</b>', job['result']['html'])
self.assertIn('ещё</a>', job['result']['html'])
def test_vk_missing_counter_is_reported_separately(self):
record = normalize('https://vk.com/wall-123_456')
service.jobs['test'] = dict(status='queued', completed=0, total=1, created=time.time())
with patch.object(service.engine, 'VK_ACCESS_TOKEN', 'test-only'), patch.object(service.engine, 'process_vk_batch', return_value={record['url']: ('Сообщество', None)}):
service.run_job('test', [record])
output = service.jobs['test']['result']
self.assertIsNone(output['records'][0]['views'])
self.assertEqual(len(output['errors']), 1)
self.assertIn('не передал счётчик', output['errors'][0])
self.assertIn('Неполный итог', output['html'])
if __name__ == '__main__':
unittest.main()
|