tradingSystem/scripts/test_batch42_units.py

"""开发机：python scripts/unit_test_guard.py scripts/test_batch42_units.py。配置隔离且禁网。"""
import unit_test_guard
import os
import sys
import copy
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 app.services import plan_recheck as rc, plan_feed as pf, proposal_service as ps
from app.repo import pms_repo
from app.core import plan_revision as pr

CODE='600183.SH'

def proposal(**kw):
    h={'preparation':{'status':'ready'},'proposal_revision':'shown','consumption_id':'saved',
       'decision':{'id':'prepared'},'plan_basis':{'risk':[]},'price':10,'sector':'电子','prepared_judgement':{'verdict':'PASS','reason':'原研究','degraded':False}}
    p={'proposal_id':'p1','ts_code':CODE,'action':'OPEN','qty':100,'hard_numbers':h,
       'hard_numbers_json':pms_repo._dumps(h),'expire_at':datetime.now()+timedelta(hours=1)}
    p.update(kw)
    return p


def current(**kw):
    r={'ok':True,'view':{'held':[]},'params':{},'stock':{},'commitments':{},
       'provenance':{'decision':{'id':'new'},'consumption_id':'unwritten','version_check':{'status':'passed'}},
       'research_changed':False,'params_changed':False,'portfolio_changed':False}
    r.update(kw)
    return r


class PreparedProposalTest(unittest.TestCase):
    def test_accept_ready_is_model_free_and_does_not_write_consumption(self):
        with patch.object(rc,'current_basis',return_value=current()) as basis, \
             patch.object(rc,'fast_checks',return_value=None), \
             patch.object(ps,'_scan_open',side_effect=AssertionError('采纳不能扫描重算')), \
             patch.object(ps,'_route_one',side_effect=AssertionError('采纳不能研判')), \
             patch.object(pms_repo,'insert_plan_snapshot',side_effect=AssertionError('采纳不能写消费快照')):
            out=rc.before_accept(proposal(),'shown')
        self.assertTrue(out['ok'])
        self.assertFalse(basis.call_args.kwargs['record'])
        self.assertEqual(out['proposal']['hard_numbers']['consumption_id'],'saved')
        self.assertEqual(out['proposal']['hard_numbers']['decision']['id'],'prepared')
        self.assertEqual(out['proposal']['hard_numbers']['version_check']['checked_decision']['id'],'new')

    def test_cash_or_other_position_change_can_accept_existing_quantity(self):
        with patch.object(rc,'current_basis',return_value=current(ok=False,material=True,portfolio_changed=True)), \
             patch.object(rc,'fast_checks',return_value=None), \
             patch.object(ps,'_route_one',side_effect=AssertionError('资金变化不问模型')):
            out=rc.before_accept(proposal(),'shown')
        self.assertTrue(out['ok'])
        self.assertEqual(out['proposal']['qty'],100)

    def test_research_change_marks_pending_without_model(self):
        with patch.object(rc,'current_basis',return_value=current(ok=False,research_changed=True)), \
             patch.object(pms_repo,'record_proposal_check',return_value=1), \
             patch.object(ps,'_route_one',side_effect=AssertionError('只能后台研判')):
            out=rc.before_accept(proposal(),'shown')
        self.assertTrue(out['pending'])
        self.assertEqual(out['proposal']['hard_numbers']['preparation']['status'],'pending')

    def test_pending_or_legacy_proposal_cannot_accept(self):
        for h in ({},{'preparation':{'status':'pending','reason':'研究更新中'}}):
            with patch.object(rc,'current_basis',side_effect=AssertionError('尚未准备不远程取数')):
                out=rc.before_accept(proposal(hard_numbers=h),'shown')
            self.assertTrue(out['pending'])

    def test_expired_proposal_does_not_read_or_accept(self):
        with patch.object(rc,'current_basis',side_effect=AssertionError('已过期')):
            out=rc.before_accept(proposal(expire_at=datetime.now()-timedelta(seconds=1)),'shown')
        self.assertFalse(out['ok'])

    def test_stale_content_is_returned_without_recalculation(self):
        with patch.object(rc,'current_basis',side_effect=AssertionError('旧页面不远程取数')):
            out=rc.before_accept(proposal(),'old')
        self.assertFalse(out['ok'])
        self.assertEqual(out['proposal']['qty'],100)

    def test_source_failure_is_closed_without_discarding_proposal(self):
        with patch.object(rc,'current_basis',side_effect=RuntimeError('offline')), \
             patch.object(pms_repo,'record_proposal_check',return_value=1):
            out=rc.before_accept(proposal(),'shown')
        self.assertFalse(out['ok'])
        self.assertTrue(out['pending'])
        self.assertEqual(out['proposal']['qty'],100)
        self.assertEqual(out['proposal']['hard_numbers']['preparation']['status'],'pending')

    def test_pause_or_money_block_is_saved_with_specific_reason(self):
        with patch.object(rc,'current_basis',return_value=current()), \
             patch.object(rc,'fast_checks',return_value='可用资金不足'), \
             patch.object(pms_repo,'record_proposal_check',return_value=1):
            out=rc.before_accept(proposal(),'shown')
        self.assertEqual(out['error'],'可用资金不足')
        self.assertTrue(out['pending'])

    def test_prepared_research_not_reasked_for_portfolio_or_param_change(self):
        old={'plan_basis':{'risk':[]},'portfolio_context_hash':'a','scan_params_hash':'a'}
        self.assertFalse(pr.needs_new_judgement(dict(old,portfolio_context_hash='b',scan_params_hash='b'),old))
        self.assertTrue(pr.needs_new_judgement(dict(old,plan_basis={'risk':['变化']}),old))

    def test_read_current_plan_has_short_timeout_and_no_snapshot(self):
        params={'base':'http://fixture','path':'/plan','timeout':30,'stale_tdays':1}
        with patch.object(pf,'_params',return_value=params), patch.object(pf,'build_query',return_value={}), \
             patch.object(pf,'fetch',return_value={'date':'2026-10-19'}) as fetch, \
             patch.object(pf,'assert_fresh') as fresh, \
             patch.object(pf,'_snapshot_quiet',side_effect=AssertionError('只读取')):
            pf.read_current_plan()
        self.assertEqual(fetch.call_args.kwargs['timeout'],3)
        fresh.assert_called_once()

    def test_new_proposal_not_ready_when_record_failed(self):
        with patch.object(rc,'current_basis',return_value={'ok':False,'error':'建议保存未完成','stage':'record'}):
            h=rc.prepare_new({'consumption_id':'c'},CODE,{'verdict':'PASS'})
        self.assertEqual(h['preparation']['status'],'pending')
        self.assertIn('保存',h['preparation']['reason'])

    def test_new_proposal_ready_only_with_recorded_consumption(self):
        with patch.object(rc,'current_basis',return_value=current()):
            self.assertEqual(rc.prepare_new({},CODE,{'verdict':'PASS'})['preparation']['status'],'pending')
            self.assertEqual(rc.prepare_new({'consumption_id':'c'},CODE,{'verdict':'PASS'})['preparation']['status'],'ready')

    def _background(self, changed, *, route=None):
        p=proposal()
        c={'ts_code':CODE,'action':'OPEN','qty':100,'hard_numbers':{'consumption_id':'c','price':10,'plan_basis':{'risk':[]}}}
        with patch.object(rc,'current_basis',side_effect=[current(ok=False,material=True,**changed),current()]), \
             patch.object(pms_repo,'record_proposal_check',return_value=1), \
             patch.object(ps,'_scan_open',return_value=[c]), \
             patch.object(ps,'_route_one',return_value={'verdict':{'verdict':'PASS','reason':'原研究'}}) as routed, \
             patch.object(pms_repo,'publish_prepared_proposal',return_value=1), \
             patch.object(rc,'fast_checks',return_value='原数量不再满足资金条件'), patch.object(ps.param_store,'get_int',return_value=0):
            out=rc.refresh_one(p)
        return out,routed

    def test_background_portfolio_change_reuses_judgement(self):
        out,route=self._background({'portfolio_changed':True})
        self.assertTrue(out['ok'])
        self.assertEqual(route.call_args.kwargs['prepared_verdict']['reason'],'原研究')
        self.assertFalse(route.call_args.args[0]['_plan_recheck'])
        self.assertEqual(out['proposal']['hard_numbers']['proposal_revision'],'shown')

    def test_background_research_change_uses_original_judge_pipeline(self):
        out,route=self._background({'research_changed':True})
        self.assertTrue(out['ok'])
        self.assertIsNone(route.call_args.kwargs['prepared_verdict'])
        self.assertTrue(route.call_args.args[0]['_plan_recheck'])
        self.assertNotEqual(out['proposal']['hard_numbers']['proposal_revision'],'shown')

    def test_active_update_lease_prevents_duplicate_judgement(self):
        p=proposal();p['hard_numbers']['preparation']['lease_until']=(datetime.now().astimezone()+timedelta(seconds=100)).isoformat()
        with patch.object(rc,'current_basis',side_effect=AssertionError('租约内不能重评')):
            self.assertFalse(rc.refresh_one(p)['ok'])

    def test_budget_exhausted_returns_to_pending_without_model(self):
        with patch.object(rc,'current_basis',return_value=current(ok=False,research_changed=True)), \
             patch.object(pms_repo,'record_proposal_check',return_value=1), \
             patch.object(ps,'_judge_budget_left',return_value=False), \
             patch.object(ps,'_scan_open',side_effect=AssertionError('预算不足不扫描')):
            out=rc.refresh_one(proposal(),deadline=0)
        self.assertTrue(out['pending'])
        self.assertNotIn('lease_until',out['proposal']['hard_numbers']['preparation'])

    def test_background_cas_loser_never_calls_model(self):
        with patch.object(rc,'current_basis',return_value=current(ok=False,research_changed=True)), \
             patch.object(pms_repo,'record_proposal_check',return_value=0), \
             patch.object(ps,'_scan_open',side_effect=AssertionError('失去所有权')):
            self.assertFalse(rc.refresh_one(proposal())['ok'])

    def test_sell_and_strategy_proposals_keep_existing_path(self):
        with patch.object(rc,'current_basis',side_effect=AssertionError('无选股重评')):
            self.assertTrue(rc.before_accept(proposal(action='EXIT'),None)['ok'])
            self.assertTrue(rc.before_accept(proposal(hard_numbers={'kind':'strategy'}),None)['ok'])

    def test_publication_and_audit_rollback_and_concurrent_accept(self):
        from sqlalchemy import create_engine,text
        from app.db import session
        engine=create_engine('sqlite:///:memory:');self.addCleanup(engine.dispose)
        p=proposal();h=rc.ready_hard({'consumption_id':'c','price':10},{'verdict':'PASS'})
        c={'qty':200,'hard_numbers':h};v={'verdict':'PASS'}
        with engine.begin() as db:
            db.execute(text('CREATE TABLE pms_proposal(proposal_id TEXT,status TEXT,qty INTEGER,hard_numbers_json TEXT,judge_verdict TEXT,judge_reason TEXT,decided_at TEXT)'))
            db.execute(text("INSERT INTO pms_proposal VALUES ('p1','WAIT_USER',100,:old,NULL,NULL,NULL)"),{'old':p['hard_numbers_json']})
        with patch.object(session,'get_engine',return_value=engine),patch.object(session,'_check_cooldown'):
            with self.assertRaises(Exception):pms_repo.publish_prepared_proposal(p,c,v)
            with engine.begin() as db:
                self.assertEqual(db.execute(text('SELECT qty FROM pms_proposal')).scalar(),100)
                db.execute(text('CREATE TABLE pms_action_ledger(ts_code TEXT,decided_at TEXT,action TEXT,arbiter TEXT,verdict TEXT,price_at REAL,hard_numbers_json TEXT,failed_checks_json TEXT,reason TEXT,ref_id TEXT,outcome_scored INTEGER)'))
            self.assertEqual(pms_repo.publish_prepared_proposal(p,c,v),1)
            self.assertEqual(pms_repo.publish_prepared_proposal(p,c,v),0)
            self.assertEqual(pms_repo.decide_proposal('p1','ACCEPTED',expected_hard=p['hard_numbers_json']),0)
        with engine.begin() as db:
            self.assertEqual(db.execute(text('SELECT COUNT(*) FROM pms_action_ledger')).scalar(),1)

    def test_quantity_changed_requires_new_confirmation_identity(self):
        h={'plan_basis':{'risk':[]}};v={'verdict':'PASS'}
        self.assertNotEqual(pr.confirmation_revision(100,h,v),pr.confirmation_revision(200,h,v))
        self.assertEqual(pr.confirmation_revision(100,h,v),pr.confirmation_revision(100,dict(h,decision={'id':'v2'},research_rank=99),v))

    def test_real_fast_checks_capacity_and_funds_without_changing_quantity(self):
        from app.services import market,portfolio,param_store,execution_conflicts
        p=proposal()
        view={'held':[],'params':{'max_names':5,'portfolio_cap':0.7,'scale':10000},
              'totals':{'names_count':1,'portfolio_mv':1000,'cash_source':portfolio.CASH_WS,'cash_avail':2000}}
        check=current(view=view,commitments={'names':set(),'amount':0})
        with patch.object(execution_conflicts,'buy_blocker',return_value=None), \
             patch.object(param_store,'get',return_value='propose_only'), \
             patch.object(param_store,'get_int',return_value=0), \
             patch.object(param_store,'get_bool',return_value=True), \
             patch.object(market,'get_price',return_value=10), \
             patch.object(market,'day_snapshot',return_value={}), \
             patch.object(market,'get_ma5',return_value=10), \
             patch.object(ps,'check_candidate_rules',return_value={'passed':True}) as gate:
            self.assertIsNone(rc.fast_checks(p,check))
            view['totals']['cash_avail']=900
            self.assertIn('资金不足',rc.fast_checks(p,check))
            view['totals']['cash_avail']=2000
            check['commitments']['names']={'a','b','c','d'}
            self.assertIn('名额不足',rc.fast_checks(p,check))
        self.assertEqual(p['qty'],100)
        gate.assert_called_once()

    def test_route_reuses_prepared_verdict_without_judge(self):
        from app.services import judge,param_store
        c={'ts_code':CODE,'action':'OPEN','side':'buy','qty':100,'price':10,'reason':'原依据','hard_numbers':{}}
        with patch.object(ps,'check_candidate_rules',return_value={'passed':True}), \
             patch.object(ps,'_pos_of',return_value={}), \
             patch.object(ps,'_self_decide',return_value={}), \
             patch.object(ps,'_apply_advice_tier'), \
             patch.object(judge,'request',side_effect=AssertionError('不许重复调用模型')), \
             patch.object(param_store,'get_bool',return_value=False):
            out=ps._route_one(c,{}, {},{},False,datetime.now(),True,{'autonomy':'propose_only'},
                             review_only=True,prepared_verdict={'verdict':'PASS','reason':'原结论'})
        self.assertEqual(out['verdict']['reason'],'原结论')

    def test_background_failure_stays_pending_and_does_not_publish(self):
        with patch.object(rc,'current_basis',return_value=current(ok=False,portfolio_changed=True)), \
             patch.object(rc,'fast_checks',return_value='资金不足'), \
             patch.object(pms_repo,'record_proposal_check',return_value=1), \
             patch.object(ps,'_scan_open',return_value=[]), \
             patch.object(pms_repo,'publish_prepared_proposal',side_effect=AssertionError('无结果不能发布')):
            out=rc.refresh_one(proposal())
        self.assertFalse(out['ok'])
        self.assertEqual(out['proposal']['hard_numbers']['preparation']['status'],'pending')

    def test_prepare_final_source_failure_cannot_publish(self):
        c={'ts_code':CODE,'qty':100,'hard_numbers':{'consumption_id':'c'}}
        with patch.object(rc,'current_basis',side_effect=[current(ok=False,research_changed=True),{'ok':False,'error':'研究更新中'}]), \
             patch.object(pms_repo,'record_proposal_check',return_value=1), \
             patch.object(ps,'_scan_open',return_value=[c]), \
             patch.object(ps,'_route_one',return_value={'verdict':{'verdict':'PASS'}}), \
             patch.object(ps.param_store,'get_int',return_value=0), \
             patch.object(pms_repo,'publish_prepared_proposal',side_effect=AssertionError('旧研究不能发布')):
            out=rc.refresh_one(proposal())
        self.assertTrue(out['pending'])

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