254 lines
16 KiB
Python
254 lines
16 KiB
Python
"""开发机：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',
|
||
'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')
|
||
|
||
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=None), 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(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()
|