73 lines
3.6 KiB
Python
73 lines
3.6 KiB
Python
"""库存数据看板旧缓存立即返回与后台刷新回归测试。"""
|
|
import importlib
|
|
import json
|
|
import unittest
|
|
from datetime import date, datetime, timedelta
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
class InventoryBoardStaleCacheTest(unittest.TestCase):
|
|
def test_expired_shared_cache_returns_immediately_and_starts_refresh(self):
|
|
from routes import dashboard
|
|
|
|
stale = {
|
|
'month': '2026-08',
|
|
'latest_date': '2026-08-31',
|
|
'all_amount': 123.0,
|
|
}
|
|
old_time = datetime.now() - timedelta(hours=2)
|
|
|
|
def fake_query(sql, params=None):
|
|
normalized = ' '.join(str(sql).split()).lower()
|
|
if normalized.startswith('select count(*) as row_count'):
|
|
return [{'row_count': 10, 'qty_sum': 20, 'updated_signature': 'sig'}]
|
|
if normalized.startswith('create table if not exists inventory_board_system_cache'):
|
|
return 0
|
|
if normalized.startswith('select payload, updated_at from inventory_board_system_cache'):
|
|
return [{'payload': json.dumps(stale), 'updated_at': old_time}]
|
|
raise AssertionError('旧缓存命中后不应继续执行冷计算SQL: ' + normalized[:120])
|
|
|
|
with mock.patch.object(dashboard, 'query', side_effect=fake_query), \
|
|
mock.patch.object(dashboard, 'cache_get', return_value=None), \
|
|
mock.patch.object(dashboard, 'cache_set') as cache_set, \
|
|
mock.patch.object(dashboard, '_start_inventory_board_system_refresh', return_value=True) as refresh:
|
|
result = dashboard._inventory_board_system('2026-08', date(2026, 8, 31))
|
|
|
|
self.assertEqual(123.0, result['all_amount'])
|
|
self.assertEqual('stale_refreshing', result['system_cache_state'])
|
|
self.assertEqual(old_time.strftime('%Y-%m-%d %H:%M:%S'), result['data_updated_at'])
|
|
refresh.assert_called_once()
|
|
cache_set.assert_called_once()
|
|
self.assertEqual(60, cache_set.call_args.kwargs['ttl'])
|
|
|
|
def test_fresh_process_cache_does_not_start_refresh(self):
|
|
from routes import dashboard
|
|
|
|
fresh = {'all_amount': 456.0, 'system_cache_state': 'fresh'}
|
|
with mock.patch.object(dashboard, 'query', return_value=[{
|
|
'row_count': 10, 'qty_sum': 20, 'updated_signature': 'sig'}]), \
|
|
mock.patch.object(dashboard, 'cache_get', return_value=fresh), \
|
|
mock.patch.object(dashboard, '_start_inventory_board_system_refresh') as refresh:
|
|
result = dashboard._inventory_board_system('2026-08', date(2026, 8, 31))
|
|
|
|
self.assertIs(fresh, result)
|
|
refresh.assert_not_called()
|
|
|
|
def test_force_refresh_bypasses_process_and_shared_cache_checks(self):
|
|
source = Path(__file__).resolve().parents[1].joinpath('routes', 'dashboard.py').read_text(encoding='utf-8')
|
|
self.assertIn('def _inventory_board_system(month, latest_date, force_refresh=False):', source)
|
|
self.assertIn('if not force_refresh:\n _cached = cache_get(_cache_key)', source)
|
|
self.assertIn('return inventory_board_stats(force_system_refresh=True)', source)
|
|
|
|
def test_frontend_shows_refresh_state_and_polls_silently(self):
|
|
app_js = Path(__file__).resolve().parents[1].joinpath('static', 'js', 'app.js').read_text(encoding='utf-8')
|
|
self.assertIn('invBoardCacheStatus', app_js)
|
|
self.assertIn('正在后台更新，当前先展示上次统计', app_js)
|
|
self.assertIn('scheduleInventoryBoardFreshPoll(d)', app_js)
|
|
self.assertIn('window._inventoryBoardRefreshAttempts > 12', app_js)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|