124 lines
5.0 KiB
Python
124 lines
5.0 KiB
Python
"""本地：python scripts/unit_test_guard.py scripts/test_batch39_units.py。禁止联网。"""
|
||
import unit_test_guard # noqa: F401
|
||
import json
|
||
import unittest
|
||
from unittest.mock import patch
|
||
from test_wiring import install_fakes
|
||
from app.services import ledger_service as ls
|
||
from app.repo import downstream_repo as ds, qmt_repo as q
|
||
|
||
CODE = '600000.SH'
|
||
IID = 'INS_R_D01'
|
||
|
||
|
||
class ReplayOwnershipTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.fake = install_fakes(prices={CODE: 10.0})
|
||
self.fake.insert_instruction(instruction_id='INS_R', origin_type='plan', origin_id='P1',
|
||
ts_code=CODE, action='OPEN', side='buy', qty=100, status='DISPATCHED')
|
||
self.fake.instructions['INS_R']['created_at'] = '2026-10-15 09:00:00'
|
||
self.fake.qmt.enqueue_order(instruction_id=IID, parent_id='INS_R', ts_code=CODE,
|
||
side='buy', qty=100, limit_price=10, valid_until=0)
|
||
self.fake.qmt.orders[IID]['broker_order_id'] = 'BROKER-1'
|
||
self.fills = [dict(order_id=IID, ts_code=CODE, side='buy', qty=100, price=11,
|
||
done_time='2026-10-15 10:00:00', raw={'qmt_order_id': 'BROKER-1'})]
|
||
self.patcher = patch.object(ds, 'fetch_filled_orders', side_effect=lambda **kw: self.fills)
|
||
self.patcher.start()
|
||
self.addCleanup(self.patcher.stop)
|
||
|
||
def trade(self, seq=1, qty=100, side='buy'):
|
||
self.fake.qmt.put_trade(seq=seq, instruction_id=IID, ts_code=CODE,
|
||
side=side, qty=qty, price=10, fee=0)
|
||
|
||
def qty(self):
|
||
return self.fake.positions.get(CODE, {}).get('total_qty', 0)
|
||
|
||
def test_buy_in_both_sources_is_booked_once(self):
|
||
self.trade()
|
||
result = ls.replay_fills()
|
||
self.assertTrue(result['ok'], result)
|
||
self.assertEqual(self.qty(), 100)
|
||
self.assertEqual(self.fake.instructions['INS_R']['exec_qty'], 100)
|
||
self.assertEqual(ls.replay_fills()['actions'], 0)
|
||
self.assertEqual(self.qty(), 100)
|
||
|
||
def test_sell_in_both_sources_is_booked_once(self):
|
||
ls._apply_action(dict(kind='BUY', ts_code=CODE, qty=300, price=9))
|
||
self.fake.qmt.orders[IID]['side'] = 'sell'
|
||
self.fake.instructions['INS_R'].update(side='sell', action='EXIT')
|
||
self.fills[0]['side'] = 'sell'
|
||
self.trade(side='sell')
|
||
self.assertTrue(ls.replay_fills()['ok'])
|
||
self.assertEqual(self.qty(), 200)
|
||
|
||
def test_table_before_ws_waits_for_exact_trades(self):
|
||
self.assertTrue(ls.replay_fills()['ok'])
|
||
self.assertEqual(self.qty(), 0)
|
||
self.trade()
|
||
self.assertTrue(ls.replay_fills()['ok'])
|
||
self.assertEqual(self.qty(), 100)
|
||
|
||
def test_partial_ws_then_delayed_remainder(self):
|
||
self.trade(qty=73)
|
||
ls.replay_fills()
|
||
self.assertEqual(self.qty(), 73)
|
||
self.trade(seq=2, qty=27)
|
||
ls.replay_fills()
|
||
self.assertEqual(self.qty(), 100)
|
||
|
||
def test_external_fill_still_books(self):
|
||
self.trade()
|
||
self.fills.append(dict(self.fills[0], order_id='MANUAL-1', qty=200, raw={}))
|
||
result = ls.replay_fills()
|
||
self.assertTrue(result['ok'], result)
|
||
self.assertEqual(self.qty(), 300)
|
||
|
||
def test_broker_identity_works_when_table_id_differs(self):
|
||
self.fills[0]['order_id'] = 'DOWNSTREAM-1'
|
||
self.trade()
|
||
self.assertTrue(ls.replay_fills()['ok'])
|
||
self.assertEqual(self.qty(), 100)
|
||
|
||
def test_lookup_failure_leaves_fill_retryable(self):
|
||
with patch.object(q, 'orders_for_fill', side_effect=RuntimeError('unavailable')):
|
||
result = ls.replay_fills()
|
||
self.assertFalse(result['ok'])
|
||
self.assertEqual(self.qty(), 0)
|
||
self.assertNotIn(IID, json.loads(self.fake.params[ls.CURSOR_KEY])['seen'])
|
||
self.trade()
|
||
self.assertTrue(ls.replay_fills()['ok'])
|
||
self.assertEqual(self.qty(), 100)
|
||
|
||
def test_conflicting_code_does_not_fall_back_to_external(self):
|
||
self.fills[0]['ts_code'] = '600001.SH'
|
||
result = ls.replay_fills()
|
||
self.assertFalse(result['ok'])
|
||
self.assertFalse(self.fake.lots)
|
||
self.assertNotIn(IID, json.loads(self.fake.params[ls.CURSOR_KEY])['seen'])
|
||
|
||
def test_conflicting_broker_identity_is_held(self):
|
||
self.fills[0]['raw']['qmt_order_id'] = 'DIFFERENT'
|
||
self.assertFalse(ls.replay_fills()['ok'])
|
||
self.assertEqual(self.qty(), 0)
|
||
|
||
def test_smoke_table_mirror_cannot_reenter_book(self):
|
||
self.fake.qmt.orders[IID]['parent_id'] = 'SMOKE_1'
|
||
self.trade()
|
||
self.assertTrue(ls.replay_fills()['ok'])
|
||
self.assertEqual(self.qty(), 0)
|
||
|
||
def test_ownership_survives_dispatch_mode_change(self):
|
||
self.fake.params['PMS_DISPATCH_MODE'] = 'shadow'
|
||
self.trade()
|
||
self.assertTrue(ls.replay_fills()['ok'])
|
||
self.assertEqual(self.qty(), 100)
|
||
|
||
def test_two_owners_fail_closed(self):
|
||
self.fake.qmt.orders['OTHER'] = dict(self.fake.qmt.orders[IID], instruction_id='OTHER')
|
||
self.assertFalse(ls.replay_fills()['ok'])
|
||
self.assertEqual(self.qty(), 0)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|