"""Independent graph/timestamp checks against original and proposed full rows."""
|
import json
|
from pathlib import Path
|
from datetime import datetime,date
|
from decimal import Decimal
|
P=Path(__file__).parent
|
def validate(before,after,plan,now):
|
changes={(x['table'],x['id']):x['after'] for x in plan['changes']}
|
ids=lambda t:{r['id']:r for r in after[t]}
|
for t,rows in before.items():
|
new=ids(t);assert set(new)=={r['id'] for r in rows},('Rows added or removed',t)
|
for old in rows:
|
proposed={**old,**changes.get((t,old['id']),{})}
|
# Database datetime objects vs serializable saved plan strings.
|
canon=lambda x:json.loads(json.dumps(x,default=str))
|
assert canon(new[old['id']])==canon(proposed),('Unexpected change',t,old['id'])
|
hol={date(2026,9,d) for d in range(25,28)}|{date(2026,10,d) for d in range(1,8)}
|
dt=lambda x:datetime.fromisoformat(str(x))
|
for change in plan['changes']:
|
for k,v in change['after'].items():
|
if v is None:continue
|
d=dt(v);assert d<=now and d.weekday()<5 and d.date() not in hol,(change['table'],change['id'],k,d)
|
row=ids(change['table'])[change['id']];assert dt(row['create_time'])<=dt(row['update_time'])
|
selected={x['order_id'] for x in plan['chains']}
|
actual={r['order_id'] for r in after['erp_sale_order_items'] if r['need_production'] in (1,True,b'\x01',"b'\\x01'")}
|
assert selected==actual
|
for chain in plan['chains']:
|
oid=chain['order_id'];order=ids('erp_sale_order')[oid];start=dt(order['order_time']);out=dt(chain['outbound_time']);production=dt(order['production_finish_time']);q=Decimal(str(order['total_count']))
|
n=len(chain['workdays']);assert (1<=n<=2 if q<=30 else 3<=n<=5 if q<=1000 else n>=6 and (out-start).days>=7)
|
assert start<production<out<=now
|
for mid in chain['mps_ids']:
|
m=ids('mes_pro_mps')[mid];assert m['sale_order_id']==oid and start<=dt(m['create_time'])<=dt(m['issue_date'])<=production
|
assert dt(m['finish_date'])==production
|
for wid in chain['work_order_ids']:
|
w=ids('mes_pro_work_order')[wid];assert start<=dt(w['create_time'])<=dt(w['finish_date'])==production
|
ts=sorted([r for r in after['mes_pro_task'] if r['work_order_id']==wid],key=lambda r:dt(r['start_time']))
|
assert ts
|
for i,t in enumerate(ts):
|
assert dt(w['create_time'])<=dt(t['start_time'])<dt(t['end_time'])<=dt(t['finish_date'])<=production
|
if i:assert dt(ts[i-1]['finish_date'])<=dt(t['start_time'])
|
for f in after['mes_pro_feedback']:
|
if f['task_id']==t['id']:assert dt(t['start_time'])<=dt(f['feedback_time'])<=dt(t['finish_date'])
|
for sale in after['mes_wm_product_sales']:
|
if sale['sale_order_id']!=oid:continue
|
assert dt(sale['sales_date'])==out
|
for line in after['mes_wm_product_sales_line']:
|
if line['sales_id']!=sale['id'] or not line['oqc_id']:continue
|
qc=ids('mes_qc_oqc')[line['oqc_id']];assert production<=dt(qc['inspect_date'])<out and dt(qc['out_date'])==out
|
for qc in after['mes_qc_ipqc']:
|
if ('mes_qc_ipqc',qc['id']) not in changes:continue
|
task=ids('mes_pro_task')[qc['task_id']];assert dt(task['end_time'])<=dt(qc['inspect_date'])<=dt(task['finish_date'])
|
for receipt in after['mes_wm_item_receipt']:
|
if ('mes_wm_item_receipt',receipt['id']) not in changes:continue
|
iqc=ids('mes_qc_iqc')[receipt['iqc_id']];notice=ids('mes_wm_arrival_notice')[receipt['notice_id']]
|
assert dt(notice['arrival_date'])<=dt(iqc['inspect_date'])<=dt(receipt['receipt_date'])
|
for produce in after['mes_wm_product_produce']:
|
if ('mes_wm_product_produce',produce['id']) not in changes:continue
|
if produce['feedback_id']:
|
f=ids('mes_pro_feedback')[produce['feedback_id']];assert dt(f['feedback_time'])<=dt(produce['produce_date'])
|
# Shared inventory and compensation receipts must not acquire new timing deficits.
|
affected={r['material_stock_id'] for r in after['mes_wm_transaction'] if ('mes_wm_transaction',r['id']) in changes}
|
def minimum(rows,sid):
|
balance=Decimal(0);smallest=Decimal(0)
|
for r in sorted([x for x in rows if x['material_stock_id']==sid],key=lambda x:(dt(x['transaction_time']),x['id'])):
|
balance+=Decimal(str(r['quantity']));smallest=min(smallest,balance)
|
return smallest
|
for sid in affected:assert minimum(after['mes_wm_transaction'],sid)>=minimum(before['mes_wm_transaction'],sid),('New chronological stock deficit',sid)
|
return {'passed':True,'orders':len(selected),'changed_rows':len(changes),'changed_tables':len({t for t,id in changes}),'cycles':[{'no':c['order_no'],'quantity':c['quantity'],'workdays':c['cycle_workdays'],'start':c['order_time'],'end':c['outbound_time']} for c in plan['chains']],'outside_scope_unchanged':True,'quantities_money_states_results_unchanged':True,'no_new_inventory_timing_deficits':True}
|
if __name__=='__main__':
|
s=json.loads((P/'snapshot.json').read_text(encoding='utf8'));p=json.loads((P/'plan.json').read_text(encoding='utf8'));before={t:x['rows'] for t,x in s['tables'].items()};changes={(x['table'],x['id']):x['after'] for x in p['changes']}
|
after={t:[{**r,**changes.get((t,r['id']),{})} for r in rs] for t,rs in before.items()}
|
print(json.dumps(validate(before,after,p,datetime.fromisoformat(s['state']['now'])),ensure_ascii=False))
|
