Spaces:
Sleeping
Sleeping
| 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() | |