"""Exercise the manual flag across save, pending and overseas confirmation.""" import unittest from unittest import mock from flask import Flask import routes.planning as planning import routes.purchase as purchase import services.field_perms as field_perms import services.planning_order_flag as flags class PlanningNewOldTest(unittest.TestCase): def setUp(self): self.app = Flask(__name__) self.app.secret_key = 'unit-test' def test_save_rejects_invalid_choice_and_missing_permission(self): for value, allowed, expected in [('无效值', True, 400), ('是', False, 403)]: with self.subTest(value=value, allowed=allowed), self.app.test_request_context( json={'product_code': 'P1', 'value': value} ), mock.patch.object(planning, 'operation_required', return_value=allowed), \ mock.patch.object(field_perms, 'field_perms_for_page', return_value={'新旧品': {'edit': allowed}}), \ mock.patch.object(planning, 'query') as sql: response, status = planning.planning_new_old_save.__wrapped__() self.assertEqual(status, expected) sql.assert_not_called() def test_save_both_choices(self): for value in ('新品', '旧品'): with self.subTest(value=value), self.app.test_request_context( json={'product_code': 'P1', 'value': value} ), mock.patch.object(planning, 'operation_required', return_value=True), \ mock.patch.object(field_perms, 'field_perms_for_page', return_value={'新旧品': {'edit': True}}), \ mock.patch.object(planning, 'query', return_value=[{'planning_new_old': '旧品'}]) as sql, \ mock.patch.object(planning, 'add_operation_log'): response = planning.planning_new_old_save.__wrapped__() self.assertEqual(response.get_json()['code'], 0) self.assertEqual(sql.call_args.args[1], (value, 'P1')) def test_saved_snapshot_survives_later_planning_changes(self): for value in ('新品', '旧品'): with mock.patch.object(flags, 'query') as sql: self.assertEqual(flags.snapshot_new_old('P1', value), value) sql.assert_not_called() with mock.patch.object(flags, 'query', return_value=[{'planning_new_old': '新品'}]): self.assertEqual(flags.snapshot_new_old('P1'), '新品') def test_single_add_and_quantity_edit_capture_and_preserve_choice(self): for existing, captured in [(None, '新品'), ({'id': 5, 'bom_items': None, 'new_old_product': '旧品'}, '旧品')]: writes = [] def sql(query, params=None): if 'SELECT id, bom_items, new_old_product' in query: return [existing] if existing else [] if query.lstrip().startswith('SELECT'): return [] self.assertEqual(query.count('%s'), len(params)) writes.append((query, params)) return 5 with self.subTest(existing=existing), self.app.test_request_context(json={ 'product_code': 'P1', 'product_name': '商品', 'product_type': '成品', 'formal_order_qty': 10 }), mock.patch.object(purchase, '_require_purchase_operation', return_value=None), \ mock.patch.object(purchase, 'query', side_effect=sql), \ mock.patch.object(purchase, '_build_planning_sales_snapshot', return_value={'昨日销量': 1}), \ mock.patch.object(flags, 'query', return_value=[{'planning_new_old': '新品'}]): response = purchase.purchase_add_pending.__wrapped__() self.assertEqual(response.get_json()['code'], 0) params = writes[0][1] self.assertEqual(params[-4] if existing else params[-3], captured) def test_save_order_modal_creates_hidden_draft_without_pending_order(self): writes = [] def sql(query, params=None): if 'SELECT id, remark, bom_items, formal_order_qty, status FROM pending_orders' in query: return [] if 'SELECT product_name, product_type FROM products' in query: return [{'product_name': '测试成品', 'product_type': '成品'}] writes.append((query, params)) return 42 bom_items = [ {'material_code': '成品_P1', 'actual_quantity': 12, 'remark': '成品备注'}, {'material_code': 'MAT-1', 'material_name': '测试包材', 'quantity': 1, 'actual_quantity': 8, 'remark': '包材备注', 'pre_pack_qty': 5, 'pkg_remark': '余额备注', 'change_manuscript': '是'}, ] with self.app.test_request_context(json={ 'product_code': 'P1', 'remark': '成品备注', 'bom_items': bom_items }), mock.patch.object(purchase, 'operation_required', side_effect=lambda page, op: page == 'planning'), \ mock.patch.object(purchase, 'query', side_effect=sql), \ mock.patch.object(purchase, '_sync_pending_pre_packaging') as sync_pack, \ mock.patch.object(purchase, 'add_operation_log'): response = purchase.purchase_update_pending.__wrapped__() self.assertEqual(response.get_json()['code'], 0) insert = next((q, p) for q, p in writes if q.lstrip().startswith('INSERT INTO pending_orders')) self.assertIn("'draft'", insert[0]) self.assertEqual(insert[1][0:3], ('P1', '测试成品', '成品')) self.assertEqual(insert[1][3], 12) self.assertTrue(any('UPDATE pending_orders SET formal_order_qty=%s, bom_items=%s' in q for q, _ in writes)) sync_pack.assert_called_once() def test_batch_add_captures_each_product_choice(self): writes = [] def sql(query, params=None): if 'SELECT p.product_code' in query: return [{'product_code': 'P1', 'planning_new_old': '新品'}, {'product_code': 'P2', 'planning_new_old': '旧品'}] if query.lstrip().startswith('SELECT'): return [] self.assertEqual(query.count('%s'), len(params)) writes.append(params) return 1 with self.app.test_request_context(json={'product_codes': ['P1', 'P2']}), \ mock.patch.object(purchase, '_require_purchase_operation', return_value=None), \ mock.patch.object(purchase, 'query', side_effect=sql), \ mock.patch.object(purchase, '_build_planning_sales_snapshot', return_value={'昨日销量': 1}), \ mock.patch.object(purchase, 'add_operation_log'): response = purchase.purchase_batch_add_pending.__wrapped__() self.assertEqual(response.get_json()['data']['added'], 2) self.assertEqual([params[-3] for params in writes], ['新品', '旧品']) def test_batch_add_promotes_existing_order_draft(self): writes = [] def sql(query, params=None): if "status='pending'" in query and query.lstrip().startswith('SELECT id'): return [] if "status='draft'" in query and query.lstrip().startswith('SELECT id'): return [{'id': 9, 'new_old_product': '新品', 'sales_snapshot': '{"昨日销量":1}', 'sales_snapshot_at': '2026-10-28 09:00:00'}] if 'SELECT p.product_code' in query: return [{'product_code': 'P1', 'product_name': '商品', 'product_type': '成品', 'planning_new_old': '新品'}] if query.lstrip().startswith('SELECT'): return [] writes.append((query, params)) return 1 with self.app.test_request_context(json={'product_codes': ['P1']}), \ mock.patch.object(purchase, '_require_purchase_operation', return_value=None), \ mock.patch.object(purchase, 'query', side_effect=sql), \ mock.patch.object(purchase, '_build_planning_sales_snapshot') as snapshot, \ mock.patch.object(purchase, 'add_operation_log'): response = purchase.purchase_batch_add_pending.__wrapped__() self.assertEqual(response.get_json()['data'], {'added': 1, 'skipped': 0}) self.assertTrue(any("status='pending'" in q and 'WHERE id=%s' in q for q, _ in writes)) snapshot.assert_not_called() def test_confirm_overseas_uses_snapshot_instead_of_current_product_value(self): for value in ('新品', '旧品'): writes = [] def sql(query, params=None): if 'SELECT product_code, product_name, product_type FROM pending_orders' in query: return [{'product_code': 'P1'}] if 'SELECT brand, product_name FROM products' in query: return [{'brand': 'TEST', 'product_name': '商品'}] if 'SELECT category FROM brand_information' in query: return [{'category': '海外'}] if 'SELECT * FROM pending_orders' in query: return [{'id': 5, 'product_code': 'P1', 'product_name': '商品', 'formal_order_qty': 10, 'new_old_product': value}] if query.lstrip().startswith('SELECT'): return [] self.assertEqual(query.count('%s'), len(params)) writes.append((query, params)) return 1 with self.subTest(value=value), self.app.test_request_context(json={'product_codes': ['P1']}), \ mock.patch.object(purchase, '_require_purchase_operation', return_value=None), \ mock.patch.object(purchase, 'query', side_effect=sql), \ mock.patch.object(purchase, 'execute', return_value=1), \ mock.patch.object(purchase, '_resolve_follower_group', return_value=''), \ mock.patch.object(purchase, '_is_overseas_notify_product', return_value=False), \ mock.patch.object(flags, 'query') as current: response = purchase.purchase_confirm_order.__wrapped__() self.assertEqual(response.get_json()['code'], 0) order = next(params for query, params in writes if 'INSERT INTO overseas_orders' in query) self.assertEqual(order[-1], value) current.assert_not_called() if __name__ == '__main__': unittest.main()