Spaces:
Sleeping
Sleeping
File size: 3,034 Bytes
152ccef | 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 | 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'))
|