175 lines
10 KiB
Python
175 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""本地：python scripts/unit_test_guard.py scripts/test_batch40_units.py。历史修复离线回归。"""
|
||
import unit_test_guard # noqa
|
||
import sys
|
||
from pathlib import Path
|
||
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
|
||
import json
|
||
import unittest
|
||
from copy import deepcopy
|
||
from contextlib import contextmanager
|
||
from unittest.mock import patch
|
||
from app.core.ledger_repair import compile_repair, net_lot_cost, D
|
||
from app.db import session
|
||
|
||
CODE='600001.SH'
|
||
|
||
|
||
def fixture():
|
||
ins=dict(id=1,instruction_id='P',ts_code=CODE,qty=200,exec_qty=200,status='CONFIRMED',
|
||
progress_json='{"children":[{"ref":"P_D01"}]}',origin_id=None,updated_at='2026-10-15 10:00:00')
|
||
order=dict(instruction_id='P_D01',parent_id='P',ts_code=CODE,side='buy',broker_order_id='B',
|
||
status='FILLED',cum_qty=100,created_at='2026-10-15 10:00:00')
|
||
lot=dict(id=1,ts_code=CODE,instruction_id='P',qty=100,closed_qty=0,open_price='10.000',
|
||
open_date='2026-10-15',created_at='2026-10-15 10:01:00',lot_type='BASE',realized_pnl='0.00')
|
||
pos=dict(id=1,ts_code=CODE,total_qty=200,avg_cost='10.000',status='HOLDING')
|
||
payload=dict(ts_code=CODE,instruction_id='P_D01',trade_no='T',broker_order_id='B',side='buy',
|
||
qty=100,price=10,amount=1000,trade_time=1792029600000)
|
||
trade=dict(seq=1,payload_json=json.dumps(payload),processed=1,received_at='2026-10-15 10:00:00',
|
||
processed_at='2026-10-15 10:01:00')
|
||
tables=dict(pms_position=[pos],pms_instruction=[ins],pms_qmt_order=[order],pms_lot=[lot,dict(lot,id=2,
|
||
instruction_id=None,created_at='2026-10-15 10:02:00')],pms_qmt_inbox=[trade],pms_plan=[],pms_strategy=[])
|
||
return dict(instance='155-sim',at='2026-10-15T12:00:00+08:00',tables=tables,
|
||
position_snapshot={'payload':{'items':[dict(ts_code=CODE,total_qty=100,avail_qty=100,market_price=11)]}},
|
||
table_fills=[dict(order_id='P_D01',ts_code=CODE,side='buy',qty=100,price=10,done_time='2026-10-15 10:00:00')])
|
||
|
||
|
||
class CoreTests(unittest.TestCase):
|
||
def plan(self,b=None):return compile_repair(b or fixture(),[CODE])
|
||
def test_quantity_cost_progress_and_terminal_hold(self):
|
||
p=self.plan();s=p['summaries'][0]
|
||
self.assertEqual((s['qty'],s['avg_cost'],s['avail_qty']),(100,'10.000',0))
|
||
self.assertEqual(s['void_lots'],[2]);self.assertEqual(s['instructions'][0]['new_status'],'CANCELLED')
|
||
self.assertEqual(s['instructions'][0]['exec_qty'],100)
|
||
def test_determinism(self):self.assertEqual(self.plan(),self.plan())
|
||
def test_missing_trade_blocks(self):
|
||
b=fixture();b['tables']['pms_qmt_inbox']=[]
|
||
with self.assertRaises(ValueError):self.plan(b)
|
||
def test_duplicate_trade_blocks(self):
|
||
b=fixture();b['tables']['pms_qmt_inbox']*=2
|
||
with self.assertRaises(ValueError):self.plan(b)
|
||
def test_wrong_broker_blocks(self):
|
||
b=fixture();b['tables']['pms_qmt_order'][0]['broker_order_id']='X'
|
||
with self.assertRaises(ValueError):self.plan(b)
|
||
def test_missing_summary_evidence_blocks(self):
|
||
b=fixture();b['table_fills']=[]
|
||
with self.assertRaises(ValueError):self.plan(b)
|
||
def test_ambiguous_lot_blocks(self):
|
||
b=fixture();b['tables']['pms_lot'][1].update(instruction_id='P',created_at='2026-10-15 10:01:00')
|
||
with self.assertRaises(ValueError):self.plan(b)
|
||
def test_position_disagreement_blocks(self):
|
||
b=fixture();b['position_snapshot']['payload']['items'][0]['total_qty']=200
|
||
with self.assertRaises(ValueError):self.plan(b)
|
||
def test_pending_order_blocks(self):
|
||
b=fixture();b['tables']['pms_qmt_order'][0]['status']='PARTIAL'
|
||
with self.assertRaises(ValueError):self.plan(b)
|
||
def test_active_grid_cost_and_qty_follow_actual_lots(self):
|
||
b=fixture();t=b['tables'];t['pms_instruction'][0].update(origin_id='GRID',progress_json='{"leg":"grid_buy:3"}')
|
||
t['pms_lot'][0]['lot_type']='ADD'
|
||
t['pms_strategy']=[dict(id=1,ts_code=CODE,strategy_id='GRID',type='GRID',status='ACTIVE',
|
||
state_json='{"pending":null,"filled_levels":{"3":{"qty":200,"price":10}},"invested":2000}')]
|
||
p=self.plan(b);row=next(c for c in p['changes'] if c['table']=='pms_strategy')
|
||
st=json.loads(row['fields']['state_json'])
|
||
self.assertEqual(st['filled_levels']['3']['qty'],100);self.assertEqual(st['invested'],1000)
|
||
def test_missing_baseline_is_not_assumed_zero(self):
|
||
b=fixture();p=json.loads(b['tables']['pms_qmt_inbox'][0]['payload_json'])
|
||
p.update(trade_no='OLD',trade_time=p['trade_time']-86400000)
|
||
b['tables']['pms_qmt_inbox'].append(dict(b['tables']['pms_qmt_inbox'][0],seq=0,payload_json=json.dumps(p)))
|
||
with self.assertRaises(ValueError):self.plan(b)
|
||
def test_existing_live_parent_stays_live(self):
|
||
b=fixture();b['tables']['pms_instruction'][0]['status']='DISPATCHED'
|
||
self.assertEqual(self.plan(b)['summaries'][0]['instructions'][0]['new_status'],'DISPATCHED')
|
||
def test_real_overfill_is_not_erased(self):
|
||
b=fixture();b['tables']['pms_instruction'][0]['qty']=50
|
||
s=self.plan(b)['summaries'][0]['instructions'][0]
|
||
self.assertEqual((s['exec_qty'],s['overfill_qty']),(100,50))
|
||
def test_new_fill_invalidates_fixed_historical_average(self):
|
||
import sqlite3
|
||
from app.repo import pms_repo
|
||
with sqlite3.connect(':memory:') as db:
|
||
db.execute('CREATE TABLE pms_instruction (instruction_id TEXT, exec_qty INTEGER, exec_avg_price REAL, updated_at TEXT, progress_json TEXT)')
|
||
db.execute("INSERT INTO pms_instruction VALUES ('P',100,10,'old','fixed evidence')")
|
||
with patch.object(pms_repo,'execute',side_effect=lambda sql,params:db.execute(sql,params).rowcount):
|
||
pms_repo.add_instruction_exec('P',100)
|
||
row=db.execute('SELECT exec_qty,exec_avg_price,progress_json FROM pms_instruction').fetchone()
|
||
self.assertEqual(row,(200,None,'fixed evidence'))
|
||
def test_money_precision_does_not_multiply_rounded_average(self):
|
||
lots=[dict(qty=1,closed_qty=3,open_price='10.000',close_avg_price='10.003',realized_pnl='0.01')]
|
||
self.assertEqual(net_lot_cost(lots),D('9.99'))
|
||
def test_sell_rebuilds_pnl_and_remaining_cost(self):
|
||
b=fixture();t=b['tables'];o=dict(t['pms_qmt_order'][0],instruction_id='S_D01',parent_id='S',side='sell',cum_qty=40)
|
||
t['pms_qmt_order'].append(o);t['pms_instruction'].append(dict(t['pms_instruction'][0],id=2,instruction_id='S',qty=40))
|
||
p=json.loads(t['pms_qmt_inbox'][0]['payload_json']);p.update(instruction_id='S_D01',trade_no='T2',side='sell',qty=40,price=11,amount=440,trade_time=p['trade_time']+60000)
|
||
t['pms_qmt_inbox'].append(dict(t['pms_qmt_inbox'][0],seq=2,payload_json=json.dumps(p)))
|
||
b['position_snapshot']['payload']['items'][0]['total_qty']=60
|
||
s=self.plan(b)['summaries'][0]
|
||
self.assertEqual((s['qty'],s['realized_pnl'],s['avg_cost']),(60,'40.00','9.333'))
|
||
|
||
|
||
class Result:
|
||
def __init__(self,rows):self.rows=rows
|
||
def mappings(self):return self
|
||
def fetchall(self):return self.rows
|
||
|
||
|
||
class Engine:
|
||
def __init__(self):self.rows={1:{'id':1,'qty':100,'ts_code':CODE},2:{'id':2,'qty':200,'ts_code':CODE}};self.audit=[];self.fail=False
|
||
@contextmanager
|
||
def begin(self):
|
||
old=deepcopy((self.rows,self.audit))
|
||
try:yield self
|
||
except Exception:
|
||
self.rows,self.audit=old
|
||
raise
|
||
def execute(self,sql,params):
|
||
s=str(sql);session.assert_single_table(s)
|
||
if s.startswith('SELECT id FROM pms_op_log'):
|
||
return Result([{'id':1}] if any(x['op']=='history_repair_summary' and x['ref']==params['r'] for x in self.audit) else [])
|
||
if s.startswith('SELECT * FROM pms_lot'):return Result([deepcopy(self.rows[params['id']])])
|
||
if s.startswith('UPDATE'):
|
||
if self.fail and params['row_id']==2:raise RuntimeError('interrupted')
|
||
self.rows[params['row_id']].update({k:v for k,v in params.items() if k!='row_id'})
|
||
return Result([])
|
||
if s.startswith('INSERT INTO pms_op_log'):self.audit.append(deepcopy(params));return Result([])
|
||
raise AssertionError(s)
|
||
|
||
|
||
class TransactionTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.engine=Engine();self.plan=dict(run_id='TEST',changes=[dict(table='pms_lot',key={'id':i},before=deepcopy(r),fields={'qty':0}) for i,r in self.engine.rows.items()])
|
||
self.p=patch.object(session,'get_engine',return_value=self.engine);self.p.start();self.addCleanup(self.p.stop)
|
||
def test_success_and_idempotency(self):
|
||
self.assertEqual(session.apply_audited_repair(self.plan,[])['changed_rows'],2)
|
||
self.assertTrue(session.apply_audited_repair(self.plan,[])['already_applied'])
|
||
self.assertEqual(len(self.engine.audit),3)
|
||
def test_failure_rolls_back_all_rows_and_audit(self):
|
||
self.engine.fail=True
|
||
with self.assertRaises(RuntimeError):session.apply_audited_repair(self.plan,[])
|
||
self.assertEqual(self.engine.rows[1]['qty'],100);self.assertEqual(self.engine.audit,[])
|
||
def test_changed_before_image_rejects(self):
|
||
self.engine.rows[2]['qty']=201
|
||
with self.assertRaises(ValueError):session.apply_audited_repair(self.plan,[])
|
||
self.assertEqual(self.engine.rows[1]['qty'],100)
|
||
def test_forbidden_table_rejected(self):
|
||
self.plan['changes'][0]['table']='real_pms_lot'
|
||
with self.assertRaises(ValueError):session.apply_audited_repair(self.plan,[])
|
||
def test_audited_inverse_restores_original_values(self):
|
||
for row in self.engine.rows.values():row['updated_at']='2026-10-15 09:00:00'
|
||
for item in self.plan['changes']:item['before']=deepcopy(self.engine.rows[item['key']['id']])
|
||
original=deepcopy(self.engine.rows)
|
||
session.apply_audited_repair(self.plan,[])
|
||
changes=[]
|
||
for row in self.engine.audit:
|
||
if row['op']!='history_repair_row':continue
|
||
item=json.loads(row['params'])
|
||
changes.append(dict(table=item['table'],key=item['key'],before=item['after'],
|
||
fields={k:item['before'][k] for k in set(item['fields'])|{'updated_at'}}))
|
||
session.apply_audited_repair(dict(run_id='TEST',changes=changes),[],rollback=True)
|
||
self.assertEqual(self.engine.rows,original)
|
||
def test_single_table_guard_still_applies(self):
|
||
with self.assertRaises(session.MultiTableSQL):
|
||
session.apply_audited_repair(self.plan,[dict(name='bad',sql='SELECT * FROM pms_lot JOIN pms_position ON 1=1',expected=[])])
|
||
|
||
|
||
if __name__=='__main__':unittest.main()
|