117 lines
4.8 KiB
Python
117 lines
4.8 KiB
Python
import importlib
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
from pathlib import Path
|
|
|
|
import services.simple_cache as simple_cache
|
|
|
|
|
|
class SimpleCacheTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
simple_cache.CACHE_DIR = self.temp_dir.name
|
|
simple_cache._cache.clear()
|
|
simple_cache._file_signatures.clear()
|
|
|
|
def tearDown(self):
|
|
self.temp_dir.cleanup()
|
|
|
|
def test_value_can_be_loaded_after_local_memory_is_cleared(self):
|
|
simple_cache.set('dashboard_stats', {'total': 12}, ttl=60)
|
|
simple_cache._cache.clear()
|
|
simple_cache._file_signatures.clear()
|
|
|
|
self.assertEqual({'total': 12}, simple_cache.get('dashboard_stats'))
|
|
|
|
def test_external_delete_invalidates_local_memory(self):
|
|
simple_cache.set('dashboard_stats', {'total': 12}, ttl=60)
|
|
os.remove(simple_cache._path('dashboard_stats'))
|
|
|
|
self.assertIsNone(simple_cache.get('dashboard_stats'))
|
|
|
|
def test_expired_file_is_not_returned(self):
|
|
simple_cache.set('dashboard_stats', {'total': 12}, ttl=-1)
|
|
|
|
self.assertIsNone(simple_cache.get('dashboard_stats'))
|
|
self.assertFalse(os.path.exists(simple_cache._path('dashboard_stats')))
|
|
|
|
|
|
class DashboardStatsCacheTest(unittest.TestCase):
|
|
def test_cache_hit_skips_database_queries(self):
|
|
from routes import dashboard
|
|
|
|
app = importlib.import_module('app').app
|
|
expected = {'total_records': 123}
|
|
with app.test_request_context('/api/dashboard/stats'):
|
|
with mock.patch.object(dashboard, 'cache_get', return_value=expected), \
|
|
mock.patch.object(dashboard, 'query') as query_mock:
|
|
response = dashboard.dashboard_stats.__wrapped__()
|
|
|
|
self.assertEqual(expected, response.get_json()['data'])
|
|
query_mock.assert_not_called()
|
|
|
|
|
|
class DashboardStatsPerformanceTest(unittest.TestCase):
|
|
def test_stale_cache_returns_immediately_and_starts_background_refresh(self):
|
|
from routes import dashboard
|
|
|
|
app = importlib.import_module('app').app
|
|
stale = {'total_products': 321}
|
|
with app.test_request_context('/api/dashboard/stats'):
|
|
with mock.patch.object(dashboard, 'cache_get', side_effect=[None, stale]), \
|
|
mock.patch.object(dashboard, '_start_dashboard_stats_refresh', return_value=True) as refresh, \
|
|
mock.patch.object(dashboard, '_compute_dashboard_stats') as compute:
|
|
response = dashboard.dashboard_stats.__wrapped__()
|
|
|
|
payload = response.get_json()
|
|
self.assertEqual(stale, payload['data'])
|
|
self.assertEqual('stale_refreshing', payload['meta']['cache'])
|
|
refresh.assert_called_once_with()
|
|
compute.assert_not_called()
|
|
|
|
def test_field_permission_filter_does_not_mutate_shared_cache_result(self):
|
|
from routes import dashboard
|
|
|
|
cached = {
|
|
'total_products': 12,
|
|
'total_cost': 345.0,
|
|
'finished_cost': 200.0,
|
|
'daily_stats': [{'date_str': '2026-10-27', '总成本': 100.0}],
|
|
}
|
|
original = __import__('copy').deepcopy(cached)
|
|
app = importlib.import_module('app').app
|
|
with app.test_request_context('/api/dashboard/stats'):
|
|
with mock.patch.object(dashboard, 'role_required', return_value=True), \
|
|
mock.patch.object(dashboard, 'operation_required', return_value=True), \
|
|
mock.patch('services.field_perms.field_perms_for_page', return_value={'总成本': {'view': False}}), \
|
|
mock.patch.object(dashboard, '_filter_field_rows'):
|
|
filtered = dashboard._filter_dashboard_cost_fields(cached)
|
|
|
|
self.assertEqual(original, cached)
|
|
self.assertNotIn('total_cost', filtered)
|
|
self.assertNotIn('finished_cost', filtered)
|
|
self.assertEqual(12, filtered['total_products'])
|
|
|
|
def test_publish_keeps_fresh_and_stale_cache_layers(self):
|
|
from routes import dashboard
|
|
|
|
result = {'total_products': 12}
|
|
with mock.patch.object(dashboard, 'cache_set') as cache_set_mock:
|
|
dashboard._publish_dashboard_stats(result)
|
|
|
|
self.assertEqual(2, cache_set_mock.call_count)
|
|
cache_set_mock.assert_any_call('dashboard_stats', result, ttl=3600)
|
|
cache_set_mock.assert_any_call('dashboard_stats_stale', result, ttl=86400)
|
|
|
|
def test_product_ranking_aggregates_before_product_join(self):
|
|
source = Path(__file__).resolve().parents[1].joinpath('routes', 'dashboard.py').read_text(encoding='utf-8')
|
|
self.assertIn('FROM (\n SELECT d.货品编号', source)
|
|
self.assertIn('LEFT JOIN products p ON a.货品编号 = p.product_code', source)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|