122 lines
6.7 KiB
Python
122 lines
6.7 KiB
Python
"""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 query.lstrip().startswith('SELECT id, bom_items'):
|
|
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(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[-2] if existing else params[-1], captured)
|
|
|
|
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, 'add_operation_log'):
|
|
response = purchase.purchase_batch_add_pending.__wrapped__()
|
|
self.assertEqual(response.get_json()['data']['added'], 2)
|
|
self.assertEqual([params[-1] for params in writes], ['是', '否'])
|
|
|
|
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 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()
|