tradingSystem/scripts/test_batch41_units.py

"""本地：python scripts/unit_test_guard.py scripts/test_batch41_units.py。配置隔离、禁止联网。"""
import unit_test_guard
import os
import sys
import unittest
from datetime import datetime, timedelta
from unittest.mock import patch
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from test_wiring import install_fakes
from app.repo import pms_repo, qmt_repo
from app.services import executor, dispatcher, proposal_service, signal_service
from app.core import signal_rules

CODE = '002518.SZ'

def ins(fake, iid, action, side, qty=600, done=0):
    fake.insert_instruction(instruction_id=iid, origin_type='proposal', origin_id=None,
        ts_code=CODE, action=action, side=side, qty=qty, status='DISPATCHED',
        progress={'children': [], 'deadline': '2026-10-21', 'is_command': False})
    fake.instructions[iid]['exec_qty'] = done
    return fake.instructions[iid]

class ConflictsTest(unittest.TestCase):
    def setUp(self):
        self.fake=install_fakes(params={'PMS_DISPATCH_MODE':'ws'})
        self.fake.qmt.set_channel()

    def test_old_partial_buy_stopped_before_exit(self):
        from app.services import execution_conflicts as c
        buy=ins(self.fake,'old','OPEN','buy',600,400)
        sell=ins(self.fake,'exit','EXIT','sell',400,200)
        result=c.enforce_exit(sell)
        self.assertTrue(result['ok'])
        self.assertEqual(buy['status'],'CANCELLED')
        self.assertEqual((buy['qty'],buy['exec_qty']),(600,400))
        self.assertEqual(sell['exec_qty'],200)
        n=len(self.fake.ledger)
        c.enforce_exit(sell)
        self.assertEqual(len(self.fake.ledger),n)

    def test_trim_pauses_without_cancelling_buy(self):
        from app.services import execution_conflicts as c
        buy=ins(self.fake,'old','OPEN','buy')
        sell=ins(self.fake,'trim','TRIM','sell')
        self.assertTrue(c.buy_blocker(CODE))
        c.enforce_exit(sell)
        self.assertEqual(buy['status'],'DISPATCHED')
        sell['status']='CONFIRMED'
        self.assertFalse(c.buy_blocker(CODE))

    def test_t0_pair_is_not_exit(self):
        from app.services import execution_conflicts as c
        ins(self.fake,'t0','T0_ROUND','sell')
        self.assertFalse(c.buy_blocker(CODE))

    def test_dispatch_rejects_buy_with_active_sell(self):
        ins(self.fake,'exit','EXIT','sell',400)
        result=dispatcher.dispatch(instruction_id='buy_D01',parent_id='buy',ts_code=CODE,
            side='buy',qty=200,limit_price=28,valid_until=datetime.now()+timedelta(minutes=10))
        self.assertFalse(result['ok'])
        self.assertEqual(self.fake.qmt.orders,{})

    def test_lookup_failure_cannot_dispatch_buy(self):
        with patch.object(pms_repo,'list_instructions',side_effect=RuntimeError('source down')):
            r=dispatcher.dispatch(instruction_id='buy_D01',ts_code=CODE,side='buy',qty=200,limit_price=28)
        self.assertFalse(r['ok'])
        self.assertEqual(self.fake.qmt.orders,{})

    def test_pending_sell_blocks_new_buy_proposals(self):
        self.fake.insert_proposal(proposal_id='sell',ts_code=CODE,action='EXIT',qty=200,
            hard_numbers={},expire_at=datetime.now()+timedelta(hours=1))
        keys=proposal_service._inflight_keys()
        for action in ('OPEN','ADD','FILL','DCA'):
            self.assertIn((CODE,action),keys)

    def test_signal_reason_preserved(self):
        parsed=signal_rules.parse_intraday({'ts_code':CODE,'action':'SELL','confidence':'0.95',
            'audit_reason':'支撑失守，资金流出','component_scores':'{"reason":"另一份理由"}'})
        self.assertEqual(parsed['reason'],'支撑失守，资金流出')
        parsed=signal_rules.parse_intraday({'component_scores':'{"reason":"资金流出"}'})
        self.assertEqual(parsed['reason'],'资金流出')

    def test_cancel_failure_keeps_stop_after_exit_completes(self):
        from app.services import execution_conflicts as c
        buy=ins(self.fake,'old','OPEN','buy',600,400)
        sell=ins(self.fake,'exit','EXIT','sell',400,200)
        with patch.object(dispatcher,'cancel',return_value={'ok':False,'error':'offline'}):
            result=c.enforce_exit(sell)
        self.assertFalse(result['ok'])
        self.assertEqual(buy['status'],'DISPATCHED')
        self.assertEqual(buy['progress']['stopped_by_exit'],'exit')
        sell['status']='CONFIRMED'
        r=dispatcher.dispatch(instruction_id='old_D03',parent_id='old',ts_code=CODE,
            side='buy',qty=200,limit_price=28)
        self.assertFalse(r['ok'])
        self.assertEqual(self.fake.qmt.orders,{})

    def test_cleanup_cancels_gated_plans_and_strategy_buy_proposal(self):
        from app.services import execution_conflicts as c
        self.fake.plans=[{'plan_id':'p1','ts_code':CODE,'action':'ADD','status':'GATED','filled_qty':0}]
        self.fake.insert_proposal(proposal_id='t0buy',ts_code=CODE,action='T0_ROUND',qty=200,
            hard_numbers={'kind':'strategy','side':'buy'},expire_at=datetime.now()+timedelta(hours=1))
        result=c.enforce_exit(ins(self.fake,'exit','EXIT','sell',400))
        self.assertTrue(result['ok'],result)
        self.assertEqual(self.fake.plans[0]['status'],'CANCELLED')
        self.assertEqual(self.fake.proposals['t0buy']['status'],'DECLINED')
        self.assertTrue(any(x['arbiter']=='system' and x['ref_id']=='t0buy' for x in self.fake.ledger))

    def test_all_exit_origins_share_executor_cleanup(self):
        from app.services import execution_conflicts as c, portfolio, exec_advisor
        for origin in ('system','proposal','plan','strategy'):
            with self.subTest(origin=origin):
                self.fake.instructions.clear()
                buy=ins(self.fake,'old','OPEN','buy',600,400)
                sell=ins(self.fake,'exit','EXIT','sell',400,200)
                sell['origin_type']=origin
                with patch.object(portfolio,'positions_view',return_value={'positions':[],'params':{}}), \
                     patch.object(executor.td,'trade_days_left',return_value=2), \
                     patch.object(exec_advisor,'decide',return_value={'action':'WAIT','reason':'等待可卖'}):
                    result=executor.run_tick()
                self.assertEqual(buy['status'],'CANCELLED',result)
                self.assertFalse(result['fired'])
                self.assertEqual(sell['exec_qty'],200)

    def test_dry_run_does_not_cancel(self):
        from app.services import portfolio, exec_advisor
        buy=ins(self.fake,'old','OPEN','buy',600,400)
        ins(self.fake,'exit','EXIT','sell',400,200)
        with patch.object(portfolio,'positions_view',return_value={'positions':[],'params':{}}), \
             patch.object(executor.td,'trade_days_left',return_value=2), \
             patch.object(exec_advisor,'decide',return_value={'action':'WAIT','reason':'等待可卖'}):
            executor.run_tick(dry_run=True)
        self.assertEqual(buy['status'],'DISPATCHED')
        self.assertFalse(buy['progress'].get('stopped_by_exit'))
        self.assertFalse(self.fake.ledger)

    def test_queued_buy_rechecks_parent_and_exit(self):
        from app.services import execution_conflicts as c
        buy=ins(self.fake,'old','OPEN','buy')
        row={'parent_id':'old','ts_code':CODE,'side':'buy'}
        self.assertIsNone(c.queued_buy_blocker(row))
        ins(self.fake,'exit','EXIT','sell')
        self.assertTrue(c.queued_buy_blocker(row))
        self.fake.instructions['exit']['status']='CONFIRMED'
        buy['status']='CANCELLED'
        self.assertTrue(c.queued_buy_blocker(row))
        self.assertIsNone(c.queued_buy_blocker(dict(row,side='sell')))

    def test_missing_parent_keeps_queue_unverified(self):
        from app.services import execution_conflicts as c
        with self.assertRaises(RuntimeError):
            c.queued_buy_blocker({'parent_id':'missing','ts_code':CODE,'side':'buy'})

    def test_finished_children_show_wait_instead_of_pending_order(self):
        from app.services import execution_conflicts as c
        row=ins(self.fake,'old','OPEN','buy',600,400)
        row['progress']['last_decision']={'reason':'当日配额已出完'}
        with patch.object(qmt_repo,'orders_for_parent',return_value=[{'status':'FILLED'}]):
            self.assertEqual(c.instruction_progress([row])[0]['execution_label'],'等待下一批')
        row['progress']['last_decision']={'reason':'可卖量不足, 顺延'}
        with patch.object(qmt_repo,'orders_for_parent',return_value=[]):
            self.assertEqual(c.instruction_progress([row])[0]['execution_label'],'等待可卖')

    def test_pending_cancel_never_shown_as_finished(self):
        from app.services import execution_conflicts as c
        row=ins(self.fake,'old','OPEN','buy');row['status']='CANCELLED'
        with patch.object(qmt_repo,'orders_for_parent',return_value=[{'status':'PARTIAL','cancel_state':'REQUESTED'}]):
            self.assertEqual(c.instruction_progress([row])[0]['execution_label'],'撤单处理中')
        with patch.object(qmt_repo,'orders_for_parent',side_effect=RuntimeError('unavailable')):
            self.assertEqual(c.instruction_progress([row])[0]['execution_label'],'委托状态待核实')

    def test_other_stock_not_blocked(self):
        from app.services import execution_conflicts as c
        ins(self.fake,'exit','EXIT','sell')
        self.assertIsNone(c.buy_blocker('600000.SH'))

    def test_claim_rejects_cancel_requested_atomically(self):
        with patch.object(qmt_repo,'execute',return_value=0) as write:
            self.assertFalse(qmt_repo.claim_order('child'))
            self.assertIn("cancel_state = 'NONE'",write.call_args.args[0])

    def test_outbox_drops_new_conflict_without_sending(self):
        import asyncio
        from unittest.mock import AsyncMock
        from app.ws.runner import WsRunner
        buy=ins(self.fake,'old','OPEN','buy')
        ins(self.fake,'exit','EXIT','sell')
        row={'instruction_id':'old_D03','parent_id':'old','ts_code':CODE,'side':'buy',
            'qty':200,'limit_price':28,'valid_until':int((datetime.now()+timedelta(minutes=10)).timestamp()*1000),
            'cancel_state':'NONE','status':'SENDING'}
        loop=asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        self.addCleanup(loop.close)
        runner=WsRunner()
        with patch.object(qmt_repo,'next_queued',return_value=[row]), \
             patch.object(qmt_repo,'claim_order',return_value=True), \
             patch.object(qmt_repo,'get_order',return_value=row), \
             patch.object(qmt_repo,'abort_order') as abort, \
             patch.object(qmt_repo,'mark_sent'), \
             patch.object(runner,'_send',new_callable=AsyncMock) as send:
            asyncio.run(runner._drain_outbox('unused'))
            abort.assert_called_once()
            send.assert_not_called()

    def test_outbox_source_failure_preserves_queue(self):
        import asyncio
        from unittest.mock import AsyncMock
        from app.ws.runner import WsRunner
        row={'instruction_id':'old_D03','parent_id':'old','ts_code':CODE,'side':'buy',
             'valid_until':int((datetime.now()+timedelta(minutes=10)).timestamp()*1000), 'cancel_state':'NONE'}
        loop=asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        self.addCleanup(loop.close)
        runner=WsRunner()
        with patch.object(qmt_repo,'next_queued',return_value=[row]), \
             patch.object(qmt_repo,'claim_order',return_value=True), \
             patch.object(qmt_repo,'get_order',return_value=row), \
             patch.object(qmt_repo,'requeue_order') as retry, \
             patch.object(qmt_repo,'mark_sent'), \
             patch.object(runner,'_send',new_callable=AsyncMock) as send:
            asyncio.run(runner._drain_outbox('unused'))
            retry.assert_called_once()
            self.assertIs(retry.call_args.args[-1],False)
            send.assert_not_called()

    def test_stale_progress_cannot_restore_stopped_buy(self):
        from app.services import execution_conflicts as c
        buy=ins(self.fake,'old','OPEN','buy',600,400)
        sell=ins(self.fake,'exit','EXIT','sell',400,200)
        with patch.object(dispatcher,'cancel',return_value={'ok':False,'error':'offline'}):
            c.enforce_exit(sell)
        buy['progress']={'children':[]}  # 模拟并发旧进度覆盖，停止事实仍在追加账本中。
        sell['status']='CONFIRMED'
        result=dispatcher.dispatch(instruction_id='old_D03',parent_id='old',ts_code=CODE,side='buy',qty=200,limit_price=28)
        self.assertFalse(result['ok'])

    def test_terminal_parent_cannot_be_revived_by_old_dispatch_write(self):
        buy=ins(self.fake,'old','OPEN','buy');buy['status']='CANCELLED'
        self.assertEqual(pms_repo.update_instruction('old',status='DISPATCHED'),0)
        self.assertEqual(buy['status'],'CANCELLED')

    def test_human_accept_is_blocked_before_changing_proposal(self):
        from app.web import main
        ins(self.fake,'exit','EXIT','sell')
        self.fake.insert_proposal(proposal_id='p',ts_code=CODE,action='OPEN',qty=200,
            hard_numbers={},expire_at=datetime.now()+timedelta(hours=1))
        with patch.object(main,'ok_logged',side_effect=lambda name,fn,**kw:fn()), \
             patch.object(main,'_actor',return_value='test'):
            result=main.api_decide(None,'p',{'decision':'ACCEPTED','reason':'用户确认'})
        self.assertFalse(result['ok'])
        self.assertEqual(self.fake.proposals['p']['status'],'WAIT_USER')
        self.assertEqual(len(self.fake.instructions),1)

if __name__=='__main__':unittest.main()