"""开发机：python scripts/unit_test_guard.py scripts/test_batch44_units.py。高股息来源，隔离禁网。""" import unit_test_guard import os,sys,copy,json from datetime import datetime,timedelta from types import SimpleNamespace from unittest.mock import patch from concurrent.futures import ThreadPoolExecutor import unittest sys.path.insert(0,os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from sqlalchemy import text import test_batch43_units as base43 from app.services import dividend_feed as df,plan_recheck as rc,proposal_service as ps,portfolio,command_service,param_store from app.core import plan_revision as pr,action_engine as ae,tradedays as td from app.repo import pms_repo from app.web import main,auth CODE='300693.SZ' REF='a'*64 def source(): now=datetime.now().astimezone() decision={'id':'dividend_'+'b'*64,'source':'dividend','policy_version':'2026.10-1','cutoff_at':now.isoformat()} c={'ts_code':CODE,'name':'合成公司','candidate_source':'dividend','src':'dividend','bucket':'dividend','verdict':'候选', 'decision':decision,'company_review':{'overall':'好','age_days':1},'company_review_text':'合成研究依据', 'review_ref':{'ref':'original-report'},'dividend_research':{'ref':REF,'conclusion':'有支持'},'rank':1,'selection_rank':1, 'risk':['需求可能下降'],'reasons':['高股息研究核验通过']} return {'schema_version':1,'source':'dividend','data_date':str(td.prev_trade_day(now.date())), 'policy_version':'2026.10-1','checked_at':now.isoformat(),'decision':decision,'eligible_count':1, 'rows':[{'ts_code':CODE,'name':'合成公司','eligible':True,'reasons':[],'research_ref':REF,'candidate':c}]} class DividendPathTest(unittest.TestCase): decide=base43.AcceptancePathTest.decide counts=base43.AcceptancePathTest.counts def setUp(self): base43.AcceptancePathTest.setUp(self) self.data=source() self.patch('app.services.param_store.get',side_effect=lambda k,d=None:'http://bridge' if k=='PMS_PLAN_API_BASE' else d) self.http.side_effect=lambda url,**kw:self.response(url,kw) self.patch('app.core.tradedays.calendar_degraded',return_value=False) self.patch('app.core.tradedays.is_trade_day',return_value=True) # Existing switch remains off: new source cannot bypass preparation/acceptance checks. self.patch('app.services.param_store.get_bool',side_effect=lambda k,d=False:False if k=='PMS_PLAN_VERSION_RECHECK' else d) c=df.candidates(record=False,recheck_code=CODE)['items'][0] self.hard.update(candidate_source='dividend',dividend_ref=REF,dividend_policy_version='2026.10-1',plan_basis=c['plan_basis']) self.save(self.hard) def patch(self,target,**kw): p=patch(target,**kw);r=p.start();self.addCleanup(p.stop);return r def response(self,url,kw): self.assertEqual(url,'http://bridge/research/dividend/candidates') self.assertEqual(kw['timeout'],(1,3)) return SimpleNamespace(raise_for_status=lambda:None,json=lambda:copy.deepcopy(self.data)) def save(self,hard): with self.engine.begin() as c:c.execute(text('UPDATE pms_proposal SET hard_numbers_json=:h WHERE proposal_id=:p'),{'h':pms_repo._dumps(hard),'p':self.pid}) def test_ready_accepts_without_research_and_retains_source(self): self.assertTrue(self.decide()['ok']);self.assertEqual(self.counts(),(1,1)) with self.engine.connect() as c: row=c.execute(text('SELECT progress_json FROM pms_instruction')).scalar() self.assertEqual(json.loads(row)['plan']['candidate_source'],'dividend') self.assertEqual(json.loads(row)['plan']['dividend_ref'],REF) def test_only_rank_and_age_change_do_not_reask(self): self.data['rows'][0]['candidate'].update(rank=900,selection_rank=900) self.data['rows'][0]['candidate']['company_review']['age_days']=9 self.assertTrue(self.decide()['ok']) def test_company_research_change_closes_confirmation(self): self.data['rows'][0]['research_ref']='c'*64 self.data['rows'][0]['candidate']['dividend_research']['ref']='c'*64 r=self.decide();self.assertTrue(r['pending']);self.assertEqual(self.counts(),(0,0)) def test_loss_of_eligibility_withdraws(self): self.data['rows'][0].update(eligible=False,reasons=['资料不完整']) self.data['rows'][0]['candidate']['verdict']='仅展示' r=self.decide();self.assertTrue(r['withdrawn']);self.assertEqual(self.counts()[0],0) self.assertEqual(pms_repo.get_proposal(self.pid)['status'],'EXPIRED') def test_other_source_open_commitment_withdraws(self): self.commitments['names'].add(CODE) self.assertTrue(self.decide()['withdrawn']);self.assertEqual(self.counts()[0],0) def test_unavailable_keeps_existing_prepared_suggestion(self): import requests self.http.side_effect=requests.Timeout('network') self.assertTrue(self.decide()['retryable']);self.assertEqual(self.counts(),(0,0)) self.assertEqual(pms_repo.get_proposal(self.pid)['hard_numbers']['preparation']['status'],'ready') def test_pending_not_accepted_when_legacy_switch_off(self): self.hard['preparation']={'status':'pending','reason':'等待准备'};self.save(self.hard) self.assertTrue(self.decide()['pending']);self.assertEqual(self.counts(),(0,0)) self.patch('app.services.tech_service.attach') r=main.api_proposals(status='WAIT_USER',limit=100,include_archived=False) self.assertFalse(r['data'][0]['accept_ready']) def test_concurrent_accepts_use_one_transaction(self): with ThreadPoolExecutor(2) as pool: rows=list(pool.map(lambda _:self.decide(),range(2))) self.assertTrue(all(r['ok'] for r in rows),rows);self.assertEqual(self.counts(),(1,1)) def test_bad_contract_future_date_calendar_or_reference_refuses(self): for change in (lambda d:d.update(data_date='2000-01-01'),lambda d:d['decision'].update(cutoff_at='2099-01-01T00:00:00+08:00'),lambda d:d['rows'][0]['candidate'].update(decision={}),lambda d:d['rows'][0]['candidate']['dividend_research'].update(ref='bad')): self.data=source();change(self.data) with self.assertRaises(df.PlanFeedError):df.read(code=CODE) self.data=source() with patch.object(td,'calendar_degraded',return_value=True): with self.assertRaises(df.PlanFeedError):df.read(code=CODE) def test_readonly_is_not_consumption_and_failed_record_blocks(self): d=df.candidates(record=False,recheck_code=CODE);self.assertNotIn('error',d['consumption']) with patch.object(pms_repo,'insert_plan_snapshot',side_effect=RuntimeError('save failed')): d=df.candidates(record=True,recheck_code=CODE) self.assertEqual(d['items'][0]['consumption_error'],'RuntimeError') def test_snapshot_uses_source_and_compatible_consumption_kind(self): with patch.object(pms_repo,'insert_plan_snapshot',return_value=1) as write: d=df.candidates(record=True,recheck_code=CODE) kw=write.call_args.kwargs;self.assertEqual(kw['meta']['source'],'dividend');self.assertEqual(kw['meta']['kind'],'rechecked') self.assertEqual(kw['meta']['order'],[CODE]);self.assertEqual(kw['roster'][0]['plan_basis'],d['items'][0]['plan_basis']) def test_first_background_prepare_dispatches_original_source(self): h=copy.deepcopy(self.hard);h.update(preparation={'status':'pending','stage':'requested'},plan_basis=None) self.save(h) current=df.candidates(record=False,recheck_code=CODE)['items'][0] fresh={'ts_code':CODE,'action':'OPEN','side':'buy','qty':500,'reason':'高股息建仓','price':34.18, 'hard_numbers':dict(self.hard,plan_basis=current['plan_basis'],consumption_id=current['consumption_id'])} with patch.object(ps,'_scan_open',return_value=[fresh]) as scan, patch.object(ps,'_route_one',return_value={'verdict':{'verdict':'PASS','reason':'合成核验','confidence':0.8}}) as route, patch.object(pms_repo,'insert_plan_snapshot',return_value=1): r=rc.refresh_one(pms_repo.get_proposal(self.pid)) self.assertTrue(r['ok'],r);self.assertEqual(scan.call_args.kwargs['candidate_source'],'dividend') self.assertTrue(route.call_args.kwargs['review_only']);self.assertEqual(self.counts()[0],0) self.assertEqual(pms_repo.get_proposal(self.pid)['hard_numbers']['preparation']['status'],'ready') def test_legacy_off_still_prepares_only_dividend(self): p=pms_repo.get_proposal(self.pid);other=copy.deepcopy(p);other['hard_numbers'].pop('candidate_source');other['proposal_id']='old' with patch.object(pms_repo,'list_proposals',return_value=[p,other]),patch.object(rc,'refresh_one',return_value={'ok':True}) as call: result=rc.refresh_pending() self.assertEqual(len(result),1);self.assertEqual(call.call_args.args[0]['proposal_id'],self.pid) class RequestTest(unittest.TestCase): def setUp(self): self.data=source();self.rows={} self.mock('app.services.dividend_feed.read',side_effect=lambda **kw:copy.deepcopy(self.data)) self.mock('app.core.tradedays.is_trade_day',return_value=True) self.mock('app.core.tradedays.calendar_degraded',return_value=False) fixed=datetime.now().replace(hour=10,minute=0) d=self.mock('app.services.dividend_feed.datetime');d.now.return_value=fixed self.mock('app.services.param_store.get',side_effect=lambda k,d=None:d) self.mock('app.services.param_store.get_bool',side_effect=lambda k,d=False:d) self.mock('app.repo.pms_repo.get_proposal',side_effect=lambda pid:self.rows.get(pid)) self.insert=self.mock('app.repo.pms_repo.insert_proposal',side_effect=self.add) self.mock('app.services.command_service.effective_stock_params',return_value={}) self.mock('app.services.portfolio.positions_view',return_value={'held':[]}) self.mock('app.services.execution_conflicts.buy_blocker',return_value=None) self.mock('app.services.proposal_service._inflight_open_commitments',return_value={'names':set(),'amount':0,'basis':[]}) self.mock('app.services.proposal_service.proposal_expire_at',return_value=fixed.replace(hour=15)) self.mock('app.repo.pms_repo.insert_instruction',side_effect=AssertionError('请求不得生成指令')) self.mock('app.services.proposal_service._route_one',side_effect=AssertionError('请求不得调用研判')) def mock(self,key,**kw): p=patch(key,**kw);v=p.start();self.addCleanup(p.stop);return v def add(self,**kw): if kw['proposal_id'] in self.rows:raise RuntimeError('duplicate') self.rows[kw['proposal_id']]=dict(kw,status='WAIT_USER');return 1 def request(self):return df.request_preparation(CODE,REF,actor='test') def test_request_has_zero_quantity_and_no_approval(self): r=self.request();p=self.rows[r['proposal_id']] self.assertEqual(p['qty'],0);self.assertEqual(p['hard_numbers']['preparation']['stage'],'requested') self.assertEqual(p['hard_numbers']['requested_by'],'test') def test_repeated_request_returns_existing_without_overwrite(self): a=self.request();b=self.request();self.assertEqual(a['proposal_id'],b['proposal_id']);self.assertEqual(self.insert.call_count,1) def test_terminal_proposal_not_revived(self): p=self.request();self.rows[p['proposal_id']]['status']='DECLINED' with self.assertRaises(ValueError):self.request() def test_missing_support_or_changed_reference_does_not_queue(self): self.data['rows'][0]['eligible']=False with self.assertRaises(ValueError):self.request() self.data['rows'][0]['eligible']=True with self.assertRaises(ValueError):df.request_preparation(CODE,'old',actor='test') self.assertEqual(self.insert.call_count,0) def test_held_black_selling_inflight_or_failure_block(self): cases=[('app.services.portfolio.positions_view',{'held':[{'ts_code':CODE}]}), ('app.services.command_service.effective_stock_params',{CODE:{'black':True}}), ('app.services.execution_conflicts.buy_blocker','正在清仓'), ('app.services.proposal_service._inflight_open_commitments',{'names':{CODE}})] for target,value in cases: with patch(target,return_value=value): with self.assertRaises(ValueError):self.request() with patch.object(ps,'_inflight_open_commitments',side_effect=RuntimeError('db failure')): with self.assertRaises(RuntimeError):self.request() self.assertEqual(self.insert.call_count,0) def test_duplicate_insert_race_reuses_winner(self): def raced(**kw):self.add(**kw);raise RuntimeError('duplicate') self.insert.side_effect=raced r=self.request();self.assertTrue(r['already_requested']);self.assertEqual(len(self.rows),1) def test_api_authorization_requires_login_not_admin(self): self.assertFalse(auth.is_public_path('/api/dividend/proposals')) self.assertEqual(auth.authz_decision('POST','/api/dividend/proposals'),'any') def test_original_basis_unchanged(self): r={'score':1,'decision':{'id':'x'}} self.assertNotIn('candidate_source',pr.basis(r));self.assertNotIn('dividend_ref',pr.basis(r)) class AssemblyTest(unittest.TestCase): def test_real_scan_uses_existing_sizer_and_preserves_dividend_basis(self): from contextlib import ExitStack from test_batch20_units import params,caps data=source() view={'held':[],'params':{'max_names':20,'scale':2000000,'portfolio_cap':0.7}, 'totals':{'names_count':0,'portfolio_mv':0,'cash_source':portfolio.CASH_WS,'cash_avail':500000}} p=params(consensus_route=False,plan_by_advice=False) settings=[(df,'read',data),(ps,'_macro_gate',{}),(param_store,'get','plan_api'), (portfolio,'caps_ctx',caps()),(ps,'_inflight_open_commitments',{'names':set(),'amount':0,'basis':[]}), (ps,'_buy_signals_today',{}),(ps,'_scan_params',p),(ps.market,'get_prices',{CODE:10}), (ps.market,'get_ma5',10),(ps.market,'day_snapshot',{}),(ps.industry,'get_many',{CODE:'电子'}), (pms_repo,'insert_plan_snapshot',1)] with ExitStack() as stack: for module,name,value in settings:stack.enter_context(patch.object(module,name,return_value=value)) out={'skipped':[]};rows=ps._scan_open(view,p,{}, {},{},out,review_code=CODE,exclude_proposal='own',candidate_source='dividend') self.assertEqual(len(rows),1,out);c=rows[0] expected,_=ae.eval_open(dict(data['rows'][0]['candidate'],price=10,sector='电子'),p,caps(),500000) self.assertEqual(c['qty'],expected['qty']);self.assertEqual(c['hard_numbers']['dividend_ref'],REF) self.assertEqual(c['hard_numbers']['plan_basis']['candidate_source'],'dividend');self.assertIn('高股息建仓',c['reason']) def test_strict_inflight_read_cannot_hide_db_failure(self): with patch.object(pms_repo,'list_instructions',side_effect=RuntimeError('db')): with self.assertRaises(RuntimeError):ps._inflight_open_commitments({'held':[]},strict=True) def test_normal_source_duplicate_does_not_overwrite_new_request(self): c={'ts_code':CODE,'action':'OPEN','qty':500,'reason':'normal','hard_numbers':{}} with patch.object(param_store,'get_int',return_value=24),patch.object(param_store,'get_bool',return_value=False),patch.object(ps,'_apply_advice_tier'),patch.object(pms_repo,'insert_proposal',side_effect=RuntimeError('duplicate')) as put,patch.object(pms_repo,'get_proposal',return_value={'status':'WAIT_USER'}): pid=ps._make_proposal(c,10,{'verdict':'PASS'}) self.assertEqual(put.call_count,1);self.assertTrue(pid.endswith('_OPEN')) if __name__=='__main__':unittest.main()