98 lines
3.6 KiB
Python
98 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试采购需求状态更新为"已下单"后期初库存的更新功能
|
|
"""
|
|
|
|
import sqlite3
|
|
import sys
|
|
import os
|
|
|
|
# 添加项目路径
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
DB_PATH = 'server/data.db'
|
|
|
|
def test_purchase_demand_ordered_update():
|
|
"""测试采购需求状态更新为已下单后期初库存自动更新"""
|
|
|
|
# 连接数据库
|
|
conn = sqlite3.connect(DB_PATH)
|
|
c = conn.cursor()
|
|
|
|
try:
|
|
# 1. 创建测试物料
|
|
material_code = 'TEST001'
|
|
material_name = '测试物料'
|
|
|
|
# 插入期初库存记录
|
|
c.execute('''INSERT OR REPLACE INTO initial_stock
|
|
(material_code, material_name, stock_qty, created_at, updated_at)
|
|
VALUES (?, ?, ?, datetime('now', 'localtime'), datetime('now', 'localtime'))''',
|
|
(material_code, material_name, 100))
|
|
|
|
# 2. 创建采购需求记录
|
|
c.execute('''INSERT INTO purchase_demand
|
|
(demand_no, material_code, material_name, order_qty, bom_unit_qty,
|
|
total_demand, initial_stock, net_demand, min_package, actual_purchase_qty,
|
|
status, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now', 'localtime'), datetime('now', 'localtime'))''',
|
|
('TEST001', material_code, material_name, 10, 5, 50, 100, 0, 10, 50, 'pending'))
|
|
|
|
demand_id = c.lastrowid
|
|
|
|
# 3. 更新状态为已下单
|
|
c.execute('''UPDATE purchase_demand
|
|
SET status = 'ordered', updated_at = datetime('now', 'localtime')
|
|
WHERE id = ?''', (demand_id,))
|
|
|
|
# 4. 模拟计算新的期初库存：实际采购 - 总需求
|
|
actual_purchase_qty = 50
|
|
total_demand = 50
|
|
expected_stock = actual_purchase_qty - total_demand
|
|
|
|
# 更新期初库存
|
|
c.execute('''INSERT OR REPLACE INTO initial_stock
|
|
(material_code, material_name, stock_qty, updated_at)
|
|
VALUES (?, ?, ?, datetime('now', 'localtime'))''',
|
|
(material_code, material_name, expected_stock))
|
|
|
|
conn.commit()
|
|
|
|
# 5. 验证结果
|
|
c.execute('SELECT stock_qty FROM initial_stock WHERE material_code = ?', (material_code,))
|
|
result = c.fetchone()
|
|
|
|
if result:
|
|
actual_stock = result[0]
|
|
print(f"✅ 测试通过！")
|
|
print(f" - 物料编码: {material_code}")
|
|
print(f" - 实际采购数量: {actual_purchase_qty}")
|
|
print(f" - 总需求: {total_demand}")
|
|
print(f" - 期望期初库存: {expected_stock}")
|
|
print(f" - 实际期初库存: {actual_stock}")
|
|
|
|
if actual_stock == expected_stock:
|
|
print(f" - 期初库存更新正确！")
|
|
else:
|
|
print(f" - ❌ 期初库存更新错误！")
|
|
else:
|
|
print("❌ 测试失败：未找到期初库存记录")
|
|
|
|
# 6. 清理测试数据
|
|
c.execute('DELETE FROM purchase_demand WHERE id = ?', (demand_id,))
|
|
c.execute('DELETE FROM initial_stock WHERE material_code = ?', (material_code,))
|
|
conn.commit()
|
|
|
|
except Exception as e:
|
|
print(f"❌ 测试出错: {e}")
|
|
conn.rollback()
|
|
finally:
|
|
conn.close()
|
|
|
|
if __name__ == '__main__':
|
|
print("开始测试采购需求状态更新功能...")
|
|
print("=" * 50)
|
|
test_purchase_demand_ordered_update()
|
|
print("=" * 50)
|
|
print("测试完成！")
|