Spaces:
Sleeping
Sleeping
| import unittest | |
| from unittest.mock import Mock, patch | |
| from reports import normalize, render | |
| from tgstat_api import archive_link, archived_post | |
| import service | |
| class TGStatTests(unittest.TestCase): | |
| def test_archive_link_and_validation(self): | |
| self.assertEqual(archive_link('memtc', '1553'), 'https://tgstat.ru/channel/@memtc/1553') | |
| with self.assertRaises(ValueError): | |
| archive_link('../secret', '1') | |
| def test_missing_token_does_not_request_anything(self): | |
| get = Mock() | |
| with patch.dict('os.environ', {'TGSTAT_API_TOKEN': ''}): | |
| self.assertIsNone(archived_post('memtc', '1553', 'Канал', get)) | |
| get.assert_not_called() | |
| def test_exact_counter_and_post_identity(self): | |
| response = Mock() | |
| response.json.return_value = {'status': 'ok', 'response': {'views': 845, 'link': 't.me/memtc/1553', 'is_deleted': 1}} | |
| get = Mock(return_value=response) | |
| with patch.dict('os.environ', {'TGSTAT_API_TOKEN': 'test-only'}): | |
| post = archived_post('memtc', '1553', 'Канал', get) | |
| self.assertEqual(post['views'], 845) | |
| self.assertIn('(TGStat)', post['title']) | |
| self.assertFalse(get.call_args.kwargs['allow_redirects']) | |
| for views in (None, '845', True, -1): | |
| response.json.return_value['response']['views'] = views | |
| self.assertIsNone(archived_post('memtc', '1553', 'Канал', get)) | |
| response.json.return_value['response'].update(views=845, link='t.me/other/1') | |
| self.assertIsNone(archived_post('memtc', '1553', 'Канал', get)) | |
| def test_fallback_only_after_telegram_failure_and_preserves_original_job_key(self): | |
| record = normalize('https://t.me/memtc/1553') | |
| fallback = dict(title='memtc (TGStat)', views=845, display_url=archive_link('memtc', '1553'), source='tgstat') | |
| with patch.object(service.engine, 'process_telegram_link', return_value=('Ошибка', 0)), patch.object(service, 'archived_post', return_value=fallback): | |
| result = service.fetch_one(record) | |
| self.assertEqual(result['url'], record['url']) | |
| self.assertIsNone(result['error']) | |
| output = render([result]) | |
| self.assertIn('href="https://tgstat.ru/channel/@memtc/1553"', output['html']) | |
| self.assertIn('0,8K', output['html']) | |
| with patch.object(service.engine, 'process_telegram_link', return_value=('Канал', 845)), patch.object(service, 'archived_post') as archive: | |
| service.fetch_one(record) | |
| archive.assert_not_called() | |
| def test_archive_api_failure_keeps_unknown_views_with_archive_link(self): | |
| with patch.object(service.engine, 'process_telegram_link', side_effect=TimeoutError), patch.object(service, 'archived_post', side_effect=TimeoutError): | |
| result = service.fetch_one(normalize('https://t.me/memtc/1553')) | |
| self.assertIsNone(result['views']) | |
| self.assertEqual(result['display_url'], archive_link('memtc', '1553')) | |