"""开发机：python scripts/unit_test_guard.py scripts/test_batch43_units.py。隔离禁网，真实SQL事务。""" import unit_test_guard import os,sys,copy,tempfile,json from datetime import datetime,timedelta from types import SimpleNamespace from concurrent.futures import ThreadPoolExecutor from unittest.mock import patch import unittest sys.path.insert(0,os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from sqlalchemy import create_engine,text,event from app.db import session from app.repo import pms_repo from app.services import plan_feed as pf,plan_recheck as rc,proposal_service as ps,portfolio,command_service,param_store,market,execution_conflicts from app.core import plan_revision as pr,tradedays as td from app.web import main import requests CODE='300693.SZ' class AcceptancePathTest(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup) self.engine=create_engine('sqlite:///'+self.tmp.name+'/state.sqlite',connect_args={'check_same_thread':False}) self.addCleanup(self.engine.dispose) def use(target,**kw): p=patch(target,**kw);v=p.start();self.addCleanup(p.stop);return v use('app.db.session.get_engine',return_value=self.engine);use('app.db.session._check_cooldown') use('app.services.param_store.get',side_effect=lambda k,d=None:d) use('app.services.param_store.get_int',side_effect=lambda k,d=0:d) use('app.services.param_store.get_float',side_effect=lambda k,d=0:d) use('app.services.param_store.get_list',side_effect=lambda k,d=None:d or []) use('app.services.param_store.get_bool',side_effect=lambda k,d=False:True if k=='PMS_PLAN_VERSION_RECHECK' else d) use('app.web.main._oplog');use('app.web.main._actor',return_value='test') self.params=pf._params();self.params.update(base='http://bridge',tiers=['强传导']) use('app.services.plan_feed._params',return_value=self.params) self.view={'held':[],'params':{'max_names':40,'scale':2000000,'portfolio_cap':0.7}, 'totals':{'names_count':0,'portfolio_mv':0,'cash_source':portfolio.CASH_WS,'cash_avail':100000}} self.commitments={'names':set(),'amount':0,'basis':[]} self.scan={'open_route_by_verdict':True} use('app.services.portfolio.positions_view',return_value=self.view) use('app.services.command_service.effective_stock_params',return_value={}) use('app.services.proposal_service._scan_params',return_value=self.scan) use('app.services.proposal_service._inflight_open_commitments',return_value=self.commitments) use('app.services.execution_conflicts.buy_blocker',return_value=None) use('app.services.market.get_price',return_value=34.18) use('app.services.market.day_snapshot',return_value={});use('app.services.market.get_ma5',return_value=34) use('app.services.proposal_service.check_candidate_rules',return_value={'passed':True}) use('app.core.tradedays.window_deadline',return_value='2026-10-23') use('app.services.proposal_service._scan_open',side_effect=AssertionError('不能重扫')) use('app.services.proposal_service._route_one',side_effect=AssertionError('不能重问模型')) use('app.repo.pms_repo.insert_plan_snapshot',side_effect=AssertionError('采纳不能写新消费')) self.raw={'date':datetime.now().date().isoformat(),'schema_version':2,'decision':{'id':'live'}, 'check_code':'SZ300693','found':True,'refresh_status':{'pending_codes':[]}, 'main':[{'code':'SZ300693','name':'盛弘股份','score':230,'tier':'强传导','rank':555, 'verdict':'候选','risk':[],'company_review_text':'研究依据'}],'observe':[]} def response(url,**kw): self.assertEqual(url,'http://bridge/plan/check');self.assertEqual(kw['params'],{'code':CODE}) return SimpleNamespace(raise_for_status=lambda:None,json=lambda:copy.deepcopy(self.raw)) self.http=use('requests.get',side_effect=response) selected=pf.candidates(record=False,recheck_code=CODE,plan_override=pf.parse_plan(self.raw),route_by_verdict=True) h={'price':34.18,'sector':'电气设备','preparation':{'status':'ready'},'proposal_revision':'shown', 'consumption_id':'already-saved','decision':{'id':'prepared'},'plan_basis':pr.basis(selected['items'][0]), 'scan_params_hash':pr.fingerprint(self.scan),'portfolio_context_hash':pr.portfolio_context(self.view,self.commitments), 'prepared_judgement':{'verdict':'UNAVAILABLE','reason':'原人工确认依据','degraded':True}} self.hard=h;self.pid='PRP_20261021_300693SZ_OPEN' ddl=[ 'CREATE TABLE pms_proposal(id INTEGER PRIMARY KEY,proposal_id TEXT UNIQUE,ts_code TEXT,action TEXT,qty INTEGER,hard_numbers_json TEXT,status TEXT,expire_at TEXT,decided_at TEXT,judge_verdict TEXT,judge_reason TEXT,archived_at 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)', 'CREATE TABLE pms_instruction(id INTEGER PRIMARY KEY,instruction_id TEXT UNIQUE,origin_type TEXT,origin_id TEXT,ts_code TEXT,action TEXT,side TEXT,qty INTEGER,limit_price REAL,window_tdays INTEGER,status TEXT,exec_qty INTEGER,progress_json TEXT,created_at TEXT,updated_at TEXT)'] with self.engine.begin() as c: for sql in ddl:c.execute(text(sql)) c.execute(text("INSERT INTO pms_proposal(proposal_id,ts_code,action,qty,hard_numbers_json,status,expire_at) VALUES (:pid,:code,'OPEN',500,:hn,'WAIT_USER',:expiry)"), {'pid':self.pid,'code':CODE,'hn':pms_repo._dumps(h),'expiry':datetime.now()+timedelta(hours=1)}) def decide(self,revision='shown'): return main.api_decide(None,self.pid,{'decision':'ACCEPTED','reason':'用户确认','proposal_revision':revision}) def counts(self): with self.engine.connect() as c:return tuple(c.execute(text('SELECT COUNT(*) FROM '+t)).scalar() for t in ('pms_instruction','pms_action_ledger')) def test_full_api_to_sql_accepts_prepared_quantity_and_pairs_original_source(self): r=self.decide();self.assertTrue(r['ok'],r);self.assertEqual(self.counts(),(1,1)) with self.engine.connect() as c: row=dict(c.execute(text('SELECT * FROM pms_instruction')).mappings().one()) self.assertEqual(row['qty'],500);p=json.loads(row['progress_json'])['plan'] self.assertEqual(p['consumption_id'],'already-saved');self.assertEqual(p['decision']['id'],'prepared') self.assertEqual(p['version_check']['checked_decision']['id'],'live') def test_transport_timeout_keeps_ready_and_retry_uses_same_confirmation(self): self.http.side_effect=requests.exceptions.ReadTimeout('3 seconds') r=self.decide();self.assertFalse(r['ok']);self.assertTrue(r['retryable']);self.assertEqual(self.counts(),(0,0)) self.assertEqual(pms_repo.get_proposal(self.pid)['hard_numbers']['preparation']['status'],'ready') self.http.side_effect=lambda *a,**k:SimpleNamespace(raise_for_status=lambda:None,json=lambda:self.raw) self.assertTrue(self.decide()['ok']);self.assertEqual(self.counts(),(1,1)) def test_response_lost_retry_returns_exact_existing_instruction(self): a=self.decide();self.http.side_effect=AssertionError('重试已采纳不再查来源') b=self.decide();self.assertTrue(b['already_accepted']);self.assertEqual(a['instruction_id'],b['instruction_id']) self.assertEqual(self.counts(),(1,1)) def test_two_concurrent_confirmations_create_one_instruction_and_audit(self): with ThreadPoolExecutor(2) as pool:r=list(pool.map(lambda _:self.decide(),range(2))) self.assertTrue(all(x['ok'] for x in r),r);self.assertEqual(len({x['instruction_id'] for x in r}),1) self.assertEqual(self.counts(),(1,1)) def test_instruction_or_audit_failure_rolls_back_confirmation(self): for table in ('pms_instruction','pms_action_ledger'): def fail(conn,cursor,statement,params,context,many): if statement.startswith('INSERT INTO '+table):raise RuntimeError('故障注入') event.listen(self.engine,'before_cursor_execute',fail) try:r=self.decide() finally:event.remove(self.engine,'before_cursor_execute',fail) self.assertFalse(r['ok']);self.assertEqual(self.counts(),(0,0)) self.assertEqual(pms_repo.get_proposal(self.pid)['status'],'WAIT_USER') self.assertTrue(self.decide()['ok']) def test_stale_confirmation_cannot_accept(self): self.assertFalse(self.decide('old')['ok']);self.assertEqual(self.counts(),(0,0)) def test_unknown_shape_or_wrong_stock_does_not_withdraw_or_accept(self): for bad in ({},dict(self.raw,check_code='SH600000'),dict(self.raw,found=False),dict(self.raw,main=None)): self.http.side_effect=lambda *a,**k:SimpleNamespace(raise_for_status=lambda:None,json=lambda:bad) self.assertFalse(self.decide()['ok']);self.assertEqual(self.counts(),(0,0)) self.assertEqual(pms_repo.get_proposal(self.pid)['status'],'WAIT_USER') def test_known_research_pending_is_not_transport_retry(self): self.raw['refresh_status']['pending_codes']=['SZ300693'] r=self.decide();self.assertTrue(r['pending']);self.assertEqual(self.counts(),(0,0)) self.assertEqual(pms_repo.get_proposal(self.pid)['hard_numbers']['preparation']['status'],'pending') def test_missing_qualification_withdraws(self): self.raw.update(found=False,main=[]) r=self.decide();self.assertTrue(r['withdrawn'],r);self.assertEqual(self.counts(),(0,1)) self.assertEqual(pms_repo.get_proposal(self.pid)['status'],'EXPIRED') def test_material_change_waits_for_background_not_accept(self): self.raw['main'][0]['risk']=['重大风险变化'] self.assertTrue(self.decide()['pending']);self.assertEqual(self.counts(),(0,0)) def test_existing_quantity_still_requires_funds(self): self.view['totals']['cash_avail']=100 r=self.decide();self.assertFalse(r['ok']);self.assertIn('资金不足',r['error']);self.assertEqual(self.counts(),(0,0)) def test_source_recovery_restores_original_content_without_scan(self): p=pms_repo.get_proposal(self.pid);h=dict(p['hard_numbers'],preparation={'status':'pending','stage':'source'}) pms_repo.record_proposal_check(p,h) self.assertTrue(rc.refresh_one(pms_repo.get_proposal(self.pid))['ok']) after=pms_repo.get_proposal(self.pid) self.assertEqual(after['qty'],500);self.assertEqual(after['hard_numbers']['proposal_revision'],'shown') self.assertTrue(self.decide()['ok']);self.assertEqual(self.counts(),(1,2)) def test_expiry_during_checks_prevents_commit(self): original=rc.fast_checks def expire(p,c): session.execute("UPDATE pms_proposal SET expire_at=:at WHERE proposal_id=:pid",{'at':datetime.now()-timedelta(seconds=1),'pid':self.pid}) return original(p,c) with patch.object(rc,'fast_checks',side_effect=expire):r=self.decide() self.assertFalse(r['ok']);self.assertEqual(self.counts(),(0,0)) if __name__=='__main__':unittest.main()