wangwensheng/AM516
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: deduct available stock before production yield adjustment

Browse Source
This commit is contained in:
ar51
2026-10-12 11:57:24 +08:00
parent de6656dfe0
commit edefe52082
6 changed files with 246 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
tests/test_feishu_bot_bridge.py
View File

@@ -428,40 +428,125 @@ class ProductionQuantityTests(unittest.TestCase):
query.assert_not_called()
record.assert_not_called()
def test_initial_and_follow_up_send_adjusted_quantity_once_to_controlled_script(self) -> None:
request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=5)
completed = SimpleNamespace(stdout=bridge.json.dumps(successful_payload()), stderr="", returncode=0)
for offset, expected_calls in ((0, 5), (bridge.CANDIDATE_BATCH_SIZE, 4)):
def test_stock_is_deducted_before_yield_and_batching(self) -> None:
for demand, stock, expected in ((24, 6, 20), (24, 0, 30), (24, 24, 0), (24, 30, 0), (25, 6, 20), (5, 1, 5), (95, 0, 100), (24, -2, 30)):
with self.subTest(demand=demand, stock=stock):
self.assertEqual(bridge.production_quantity_for_demand(demand, stock), expected)
def test_initial_and_continuation_query_net_production_per_model(self) -> None:
request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=24)
payload = successful_payload()
payload["data"]["库存信息"]["可用库存(扣除了待产数量)"] = 6
completed = SimpleNamespace(stdout=bridge.json.dumps(payload), stderr="", returncode=0)
for offset, model_count in ((0, 5), (bridge.CANDIDATE_BATCH_SIZE, 4)):
with (
self.subTest(offset=offset),
mock.patch.object(bridge.subprocess, "run", return_value=completed) as run,
mock.patch.object(bridge, "write_record") as record,
):
report = bridge.build_report(request, candidate_offset=offset)
self.assertEqual(run.call_count, expected_calls)
by_model = {}
for call in run.call_args_list:
command = call.args[0]
self.assertEqual(command[command.index("--quantity") + 1], "10")
self.assertIn(f"输入需求：{request.model}，5台", report)
self.assertEqual(report.count("- 需求数量：5台"), expected_calls)
self.assertEqual(report.count("- 排产测算数量：10台"), expected_calls)
self.assertEqual(record.call_args.args[0].quantity, 5)
by_model.setdefault(command[command.index("--product") + 1], []).append(command[command.index("--quantity") + 1])
self.assertEqual(len(by_model), model_count)
self.assertTrue(all(quantities == ["24", "20"] for quantities in by_model.values()))
self.assertEqual(report.count("- 扣库存后净需求：18台"), model_count)
self.assertEqual(report.count("- 排产测算数量：20台"), model_count)
self.assertIn("部分覆盖，仍需补齐18台", report)
recorded = bridge.render_record(*record.call_args.args)
self.assertIn("**输入数量**：5台", recorded)
self.assertIn("**排产测算数量（API查询数量）**：10台", recorded)
self.assertIn("**输入数量**：24台", recorded)
self.assertEqual(recorded.count("排产测算数量20台；最终API查询数量20台"), model_count)
def test_stock_coverage_uses_same_quantity_as_api_forecast(self) -> None:
def test_stock_covering_original_demand_needs_no_production_query(self) -> None:
payload = successful_payload()
payload["data"]["库存信息"]["可用库存(扣除了待产数量)"] = 7
with (
mock.patch.object(bridge, "run_controlled_query", return_value=payload),
mock.patch.object(bridge, "run_controlled_query", return_value=payload) as query,
mock.patch.object(bridge, "write_record"),
):
report = bridge.build_report(bridge.RequestInput("eRob70H50I-BHM-18CTC[V5]", 5))
self.assertIn("库存覆盖口径：按排产测算数量判断", report)
self.assertIn("部分覆盖，仍需补齐3台", report)
self.assertIn("是否完全覆盖数量：否", report)
self.assertIn("现货候选：无", report)
self.assertTrue(all(call.args[0].quantity == 5 for call in query.call_args_list))
self.assertIn("库存覆盖口径：按原始需求数量判断", report)
self.assertIn("整机可用库存7台（完全覆盖）", report)
self.assertIn("- 排产测算数量：0台", report)
self.assertIn("是否完全覆盖数量：是", report)
def test_each_candidate_uses_its_own_stock(self) -> None:
request = bridge.RequestInput("eRob70H50I-FS-18CN[V5]", 24)
models = [request.model, *[s['model'] for s in bridge.generate_candidate_specs(request.model)[:4]]]
stocks = dict(zip(models, [6, 0, 24, 5, 30]))
def query(r):
payload = successful_payload()
payload['data']['库存信息']['可用库存(扣除了待产数量)'] = stocks[r.model]
return payload
with (
mock.patch.object(bridge, 'run_controlled_query', side_effect=query) as run,
mock.patch.object(bridge, 'write_record') as record,
):
bridge.build_report(request)
candidates = record.call_args.args[1]
self.assertEqual([c['details']['production_quantity'] for c in candidates], [20, 30, 0, 20, 0])
for model, expected in zip(models, [[24, 20], [24, 30], [24], [24, 20], [24]]):
self.assertEqual([c.args[0].quantity for c in run.call_args_list if c.args[0].model == model], expected)
def test_missing_or_invalid_stock_cannot_be_treated_as_zero(self) -> None:
for stock in (None, True, '6', 1.5, float('nan'), float('inf')):
payload = successful_payload()
payload['data']['库存信息']['可用库存(扣除了待产数量)'] = stock
with (
self.subTest(stock=stock),
mock.patch.object(bridge, 'run_controlled_query', return_value=payload) as query,
):
details = bridge.query_model_for_demand(bridge.RequestInput('model', 24))
self.assertFalse(details['has_lead_time'])
self.assertIn('无法计算净需求', details['unavailable_reason'])
query.assert_called_once()
def test_partial_stock_cannot_trigger_false_stock_delivery(self) -> None:
payload = successful_payload()
payload['data']['库存信息']['可用库存(扣除了待产数量)'] = 20
with mock.patch.object(bridge, 'run_controlled_query', return_value=payload) as query:
details = bridge.query_model_for_demand(bridge.RequestInput('model', 24))
self.assertEqual(details['net_demand'], 4)
self.assertEqual(details['production_quantity'], 5)
self.assertFalse(details['has_lead_time'])
self.assertIn('误走现货分支', details['unavailable_reason'])
query.assert_called_once()
def test_stock_change_between_queries_invalidates_forecast(self) -> None:
first, second = successful_payload(), successful_payload()
first['data']['库存信息']['可用库存(扣除了待产数量)'] = 6
second['data']['库存信息']['可用库存(扣除了待产数量)'] = 7
with mock.patch.object(bridge, 'run_controlled_query', side_effect=[first, second]):
details = bridge.query_model_for_demand(bridge.RequestInput('model', 24))
self.assertFalse(details['has_lead_time'])
self.assertIn('库存发生变化', details['unavailable_reason'])
def test_forecast_uses_production_response_date_and_materials(self) -> None:
first, second = successful_payload(), successful_payload()
for payload in (first, second):
payload['data']['库存信息']['可用库存(扣除了待产数量)'] = 6
second['data']['交期汇总']['AM516工作日交期'] = 8
second['data']['交期汇总']['AM516最终交货日期'] = '2026-10-22'
second['data']['采购信息']['缺料明细'] = [{'物料名称': '测算物料'}]
with mock.patch.object(bridge, 'run_controlled_query', side_effect=[first, second]):
details = bridge.query_model_for_demand(bridge.RequestInput('model', 24))
self.assertTrue(details['has_lead_time'])
self.assertEqual(details['prediction_value'], '8个工作日，最终交货日期：2026-10-22')
self.assertIn('- 缺料物料明细：测算物料', details['evidence_lines'])
def test_unsupported_baseline_stops_before_candidate_queries_and_record(self) -> None:
payload = successful_payload()
payload['data']['库存信息']['可用库存(扣除了待产数量)'] = 20
with (
mock.patch.object(bridge, 'run_controlled_query', return_value=payload) as query,
mock.patch.object(bridge, 'write_record') as record,
):
with self.assertRaisesRegex(RuntimeError, '净需求4台，排产测算数量5台'):
bridge.build_report(bridge.RequestInput('eRob70H50I-FS-18CN[V5]', 24))
query.assert_called_once()
record.assert_not_called()
class CandidateBatchTests(unittest.TestCase):
@@ -871,7 +956,7 @@ class ApiWorkdayOutputTests(unittest.TestCase):
mock.patch.object(bridge, "run_controlled_query", return_value=payload),
mock.patch.object(bridge, "write_record") as write_record,
):
with self.assertRaisesRegex(RuntimeError, "工作日交期和最终交货日期"):
with self.assertRaisesRegex(RuntimeError, "缺少“AM516工作日交期”字段"):
bridge.build_report(request)
write_record.assert_not_called()