218 lines
12 KiB
Python
218 lines
12 KiB
Python
"""Sales cache freshness, code provenance and pagination regression checks."""
|
|
import tempfile
|
|
import unittest
|
|
from datetime import date
|
|
from unittest.mock import Mock, MagicMock, patch
|
|
from flask import session
|
|
|
|
from app import app
|
|
from routes import spot
|
|
from services import simple_cache, spot_sales_cache
|
|
|
|
|
|
class SalesSnapshotTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.directory = patch.object(simple_cache, 'CACHE_DIR', self.temp.name)
|
|
self.directory.start()
|
|
self.addCleanup(self.directory.stop)
|
|
self.version = patch.object(spot_sales_cache, '_data_version_signature', return_value='v1').start()
|
|
self.addCleanup(patch.stopall)
|
|
self.products = [{'货品编号': 'SKU', '商家编码': '', '平台': '京东'}]
|
|
self.build = Mock(return_value=([{'商家编码': 'M1'}], date(2026, 10, 14)))
|
|
|
|
def test_reuses_snapshot_across_worker_memory(self):
|
|
a = spot_sales_cache.sales_snapshot(self.products, self.build)
|
|
simple_cache._cache.clear()
|
|
simple_cache._file_signatures.clear()
|
|
self.assertEqual(spot_sales_cache.sales_snapshot(self.products, self.build), a)
|
|
self.build.assert_called_once()
|
|
|
|
def test_import_version_invalidates_snapshot(self):
|
|
spot_sales_cache.sales_snapshot(self.products, self.build)
|
|
self.version.return_value = 'v2'
|
|
spot_sales_cache.sales_snapshot(self.products, self.build)
|
|
self.assertEqual(self.build.call_count, 2)
|
|
|
|
def test_product_edit_or_classification_change_invalidates_snapshot(self):
|
|
spot_sales_cache.sales_snapshot(self.products, self.build)
|
|
self.products[0]['商家编码'] = 'CORRECTED'
|
|
spot_sales_cache.sales_snapshot(self.products, self.build)
|
|
spot_sales_cache.sales_snapshot([], self.build)
|
|
self.assertEqual(self.build.call_count, 3)
|
|
|
|
def test_expiry_rebuilds_snapshot(self):
|
|
spot_sales_cache.sales_snapshot(self.products, self.build)
|
|
with patch.object(simple_cache.time, 'time', return_value=simple_cache.time.time() + 61):
|
|
spot_sales_cache.sales_snapshot(self.products, self.build)
|
|
self.assertEqual(self.build.call_count, 2)
|
|
|
|
def test_import_during_build_does_not_publish_snapshot(self):
|
|
self.version.side_effect = ['v1', 'v2']
|
|
spot_sales_cache.sales_snapshot(self.products, self.build)
|
|
self.assertIsNone(simple_cache.get('spot-sales-snapshot-v1'))
|
|
|
|
|
|
class SpotSalesRowsTest(unittest.TestCase):
|
|
def test_manual_inbound_overrides_auto_and_clear_restores_auto(self):
|
|
automatic = {'SKU': date(2026, 10, 1)}
|
|
products = [{'product_code': 'SKU', 'spot_first_inbound_override': date(2026, 9, 1)}]
|
|
self.assertEqual(spot._spot_effective_inbound_map(products, automatic)['SKU'], date(2026, 9, 1))
|
|
products[0]['spot_first_inbound_override'] = None
|
|
self.assertEqual(spot._spot_effective_inbound_map(products, automatic), automatic)
|
|
|
|
def test_search_actual_brand(self):
|
|
with app.test_request_context('/api/spot/sales/list?search=泊紫汀兰'), \
|
|
patch.object(spot, 'role_required', return_value=True), \
|
|
patch.object(spot, 'query', return_value=[]), \
|
|
patch.object(spot, 'sales_snapshot', return_value=([
|
|
{'货品编号': 'SKU', '品牌': '泊紫汀兰'}, {'货品编号': 'OTHER', '品牌': '洁维塔'}], None)):
|
|
result = spot.sales_list.__wrapped__().get_json()
|
|
self.assertEqual(result['total'], 1)
|
|
self.assertEqual(result['data'][0]['货品编号'], 'SKU')
|
|
|
|
def test_first_inbound_date_uses_earliest_completed_domestic_order(self):
|
|
completed_date = date(2026, 10, 15)
|
|
with patch.object(spot, 'query', return_value=[
|
|
{'product_code': 'SKU1', 'completion_date': completed_date},
|
|
]) as query:
|
|
result = spot._spot_domestic_completion_map()
|
|
self.assertEqual(result, {'SKU1': completed_date})
|
|
self.assertIn('MIN(`完结时间`)', query.call_args.args[0])
|
|
self.assertIn("`状态`, '')) = '已完结'", query.call_args.args[0])
|
|
|
|
def test_only_unique_inventory_codes_fill_blanks_and_keep_existing_values(self):
|
|
products = [{'货品编号': 'SKU1', '商家编码': ' KNOWN ', 'planning_new_old': '是'},
|
|
{'货品编号': 'SKU2', '商家编码': None, 'planning_new_old': '否'},
|
|
{'货品编号': 'SKU3', '商家编码': ''}]
|
|
with patch.object(spot, 'query', return_value=[{'product_code': 'SKU2', 'goods_code': '000123'}]) as query, \
|
|
patch.object(spot, '_spot_sales_map', return_value=({}, date(2026, 10, 14))), \
|
|
patch.object(spot, '_spot_first_inventory_map', return_value={}), \
|
|
patch.object(spot, '_spot_domestic_completion_map', return_value={}), \
|
|
patch.object(spot, 'compute_product_inventory_cached', return_value={}), \
|
|
patch.object(spot, 'compute_product_in_transit_cached', return_value={}):
|
|
rows, _ = spot._build_spot_sales_rows(products)
|
|
result = {r['货品编号']: r for r in rows}
|
|
self.assertEqual(result['SKU1']['新旧品'], '新品')
|
|
self.assertEqual(result['SKU2']['新旧品'], '旧品')
|
|
self.assertEqual(result['SKU3']['新旧品'], '旧品')
|
|
self.assertEqual(result['SKU1']['商家编码'], 'KNOWN')
|
|
self.assertEqual(result['SKU2']['商家编码'], '000123')
|
|
self.assertEqual(result['SKU2']['商家编码来源'], '库存数据')
|
|
self.assertEqual(result['SKU3']['商家编码'], '')
|
|
self.assertEqual(result['SKU3']['商家编码来源'], '未找到唯一对应编码')
|
|
self.assertIn('HAVING COUNT(DISTINCT TRIM(goods_code)) = 1', query.call_args.args[0])
|
|
self.assertEqual(query.call_args.args[1], ['SKU2', 'SKU3'])
|
|
self.assertIsNone(products[1]['商家编码'])
|
|
|
|
def test_search_resolved_code_and_paginate_without_mutating_cache(self):
|
|
cached = [{'商家编码': 'FOUND', '货品编号': 'SKU1', 'category1': '类目'},
|
|
{'商家编码': 'FOUND2', '货品编号': 'SKU2', 'category1': '类目'}]
|
|
with app.test_request_context('/api/spot/sales/list?search=found&page=2&size=1'), \
|
|
patch.object(spot, 'role_required', return_value=True), \
|
|
patch.object(spot, 'query', return_value=[]), \
|
|
patch.object(spot, 'sales_snapshot', return_value=(cached, date(2026, 10, 14))):
|
|
result = spot.sales_list.__wrapped__().get_json()
|
|
self.assertEqual(result['total'], 2)
|
|
self.assertEqual(result['data'][0]['商家编码'], 'FOUND2')
|
|
self.assertNotIn('category1', result['data'][0])
|
|
self.assertEqual(cached[1]['category1'], '类目')
|
|
|
|
def test_permission_check_before_data_or_cache_access(self):
|
|
with app.test_request_context('/api/spot/sales/list'), \
|
|
patch.object(spot, 'role_required', return_value=False), \
|
|
patch.object(spot, 'query') as query, patch.object(spot, 'sales_snapshot') as cache:
|
|
_, status = spot.sales_list.__wrapped__()
|
|
self.assertEqual(status, 403)
|
|
query.assert_not_called()
|
|
cache.assert_not_called()
|
|
|
|
|
|
class SpotSalesInboundPermissionTest(unittest.TestCase):
|
|
def allowed(self, role_cfg=None, position_cfg=None, user_cfg=None,
|
|
page_view=True, page_edit=False, role='viewer'):
|
|
with app.test_request_context('/api/spot/sales/list'), \
|
|
patch.object(spot, 'role_required', return_value=page_view), \
|
|
patch.object(spot, 'operation_required', return_value=page_edit), \
|
|
patch('services.field_perms._load_field_config', return_value=(
|
|
role_cfg or {}, position_cfg or {}, user_cfg or {}, set())):
|
|
session.update(user_id=23, role=role)
|
|
return spot._can_edit_spot_sales_inbound()
|
|
|
|
def test_field_grant_works_without_page_edit(self):
|
|
grant = {'首批入库时间': {'edit': 1, 'view': 1}}
|
|
self.assertTrue(self.allowed(role_cfg=grant))
|
|
self.assertTrue(self.allowed(position_cfg=grant))
|
|
|
|
def test_override_priority_and_denials(self):
|
|
grant = {'首批入库时间': {'edit': 1}}
|
|
deny = {'首批入库时间': {'edit': 0}}
|
|
self.assertFalse(self.allowed(grant, deny, page_edit=True))
|
|
self.assertFalse(self.allowed(position_cfg=grant, user_cfg=deny))
|
|
self.assertTrue(self.allowed(position_cfg=deny, user_cfg=grant))
|
|
self.assertFalse(self.allowed(position_cfg=grant, page_view=False))
|
|
self.assertFalse(self.allowed(position_cfg={'首批入库时间': {'edit': 1, 'view': 0}}))
|
|
|
|
def test_defaults_and_admin(self):
|
|
self.assertFalse(self.allowed())
|
|
self.assertTrue(self.allowed(page_edit=True))
|
|
self.assertTrue(self.allowed(role='admin'))
|
|
|
|
def test_list_permission_is_not_shared_with_cached_rows(self):
|
|
cached = [{'货品编号': 'SKU'}]
|
|
with app.test_request_context('/api/spot/sales/list'), \
|
|
patch.object(spot, 'role_required', return_value=True), \
|
|
patch.object(spot, 'query', return_value=[]), \
|
|
patch.object(spot, 'sales_snapshot', return_value=(cached, None)), \
|
|
patch.object(spot, '_can_edit_spot_sales_inbound', side_effect=[True, False]):
|
|
self.assertTrue(spot.sales_list.__wrapped__().get_json()['can_edit_inbound_date'])
|
|
self.assertFalse(spot.sales_list.__wrapped__().get_json()['can_edit_inbound_date'])
|
|
self.assertEqual(cached, [{'货品编号': 'SKU'}])
|
|
|
|
|
|
class SpotSalesInboundSaveTest(unittest.TestCase):
|
|
def request_save(self, payload, allowed=True, exists=True):
|
|
conn = MagicMock()
|
|
cur = conn.cursor.return_value.__enter__.return_value
|
|
cur.fetchone.return_value = {'id': 1} if exists else None
|
|
with app.test_request_context('/api/spot/sales/inbound-date', method='POST', json=payload), \
|
|
patch.object(spot, 'role_required', return_value=True), \
|
|
patch.object(spot, '_can_edit_spot_sales_inbound', return_value=allowed), \
|
|
patch.object(spot, 'get_db', return_value=conn) as db, \
|
|
patch.object(spot, 'bump_data_version') as bump, \
|
|
patch.object(spot, 'add_operation_log'):
|
|
result = spot.sales_inbound_date_save.__wrapped__()
|
|
return result, conn, cur, db, bump
|
|
|
|
def test_save_and_clear_are_persisted_and_invalidate_cache(self):
|
|
for value, expected in [('2026-10-01', date(2026, 10, 1)), ('', None)]:
|
|
result, conn, cur, _, bump = self.request_save({'product_code': 'SKU', 'date': value})
|
|
self.assertEqual(result.get_json()['code'], 0)
|
|
self.assertEqual(cur.execute.call_args.args[1], (expected, 'SKU', '%现货%'))
|
|
conn.commit.assert_called_once()
|
|
bump.assert_called_once()
|
|
|
|
def test_invalid_or_missing_date_does_not_access_db(self):
|
|
for payload in [{'product_code': 'SKU'}, *[
|
|
{'product_code': 'SKU', 'date': value} for value in ['2026-02-30', '20261001', 12]]]:
|
|
result, _, _, db, _ = self.request_save(payload)
|
|
self.assertEqual(result[1], 400)
|
|
db.assert_not_called()
|
|
|
|
def test_edit_permission_is_required(self):
|
|
result, _, _, db, _ = self.request_save({'product_code': 'SKU', 'date': ''}, allowed=False)
|
|
self.assertEqual(result[1], 403)
|
|
db.assert_not_called()
|
|
|
|
def test_missing_or_non_spot_product_is_not_changed(self):
|
|
result, conn, cur, _, bump = self.request_save({'product_code': 'SKU', 'date': ''}, exists=False)
|
|
self.assertEqual(result[1], 404)
|
|
self.assertEqual(cur.execute.call_count, 1)
|
|
conn.commit.assert_not_called()
|
|
bump.assert_not_called()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|