# -*- 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()