diff --git a/DEVLOG.md b/DEVLOG.md index 04bc73a..8aae336 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -2088,3 +2088,15 @@ QMT在线、对账无差异。午休任务按原时段跳过，不制造交易 恢复与异常细节见四仓交接11:40节。 回退标签pms:rollback-before-prepared-20261020。 回退须源码反向提交后拉取并重建。 + + +## 2026-10-21 采纳重请求与落库中断修复（开发验证） + +09:45盛弘股份采纳超时来自3秒整榜读取。桥现场5.481秒、3093481字节。 +新增逐票发布结果接口，采纳不下载整榜。网络暂不可用保留原准备内容，不触发重新研判。 +旧source准备状态恢复时核验原500股，不改确认标识。已知修订仍关闭采纳，资格失效仍撤下。 +普通OPEN确认、审计和指令原子提交。重复请求读回原指令，不重复下发。 +指令标识追加提议身份摘要，避免同日不同提议撞用固定序号。OPEN没有一次性计数器，其他动作路径保持。 +页面只在确认内容确实变化时清空原理由，不把连接失败叫建议更新。 +新增批次43，实际API、来源解析、筛选、快速检查和SQL事务共13例；桥逐票7例。 +47批PMS禁网回归与三页面守卫通过。桥完整版本契约回归通过。部署及真机读数随后追加。 diff --git a/app/db/session.py b/app/db/session.py index 95a1aff..0eb76a6 100644 --- a/app/db/session.py +++ b/app/db/session.py @@ -365,3 +365,38 @@ def publish_prepared_proposal(old, candidate, verdict): 'verdict':'NOTE','price':float(hard.get('price') or 0),'hard':encode(audit),'failed':'[]', 'reason':'后台完成建议更新，等待人工确认','ref':old['proposal_id']}) return 1 + + +def accept_open_proposal(old, hard, instruction, reason): + """普通建仓的确认、动作审计、待执行指令同成同败；每条SQL仍严格单表。""" + import json + from datetime import datetime + if old['action'] != 'OPEN' or hard.get('kind') == 'strategy' or int(old['qty']) <= 0: + raise ValueError('只接受有效的普通建仓建议') + encode = lambda v: json.dumps(v, ensure_ascii=False, default=str) + now = datetime.now() + _check_cooldown('proxy') + with get_engine('proxy').begin() as c: + def query(sql, params): + sql = map_tables(sql) + assert_single_table(sql) + return c.execute(text(sql), params) + n = query("UPDATE pms_proposal SET status='ACCEPTED',decided_at=:at,hard_numbers_json=:hn " + "WHERE proposal_id=:pid AND status='WAIT_USER' AND hard_numbers_json=:old " + "AND expire_at>=:at AND qty=:qty", + {'at':now,'hn':encode(hard),'pid':old['proposal_id'], + 'old':old['hard_numbers_json'],'qty':int(old['qty'])}).rowcount + if not n: + return False + query('INSERT INTO pms_action_ledger (ts_code,decided_at,action,arbiter,verdict,price_at,' + 'hard_numbers_json,failed_checks_json,reason,ref_id,outcome_scored) ' + "VALUES (:code,:at,'OPEN','user','PASS',:px,:hn,'[]',:reason,:pid,0)", + {'code':old['ts_code'],'at':now,'px':float(hard.get('price') or 0), + 'hn':encode(hard),'reason':reason[:500],'pid':old['proposal_id']}) + query('INSERT INTO pms_instruction (instruction_id,origin_type,origin_id,ts_code,action,side,' + 'qty,limit_price,window_tdays,status,exec_qty,progress_json,created_at,updated_at) ' + "VALUES (:iid,'proposal',:pid,:code,'OPEN','buy',:qty,:price,:window,'PROPOSED',0,:pg,:at,:at)", + {'iid':instruction['id'],'pid':old['proposal_id'],'code':old['ts_code'], + 'qty':int(old['qty']),'price':hard.get('price'),'window':instruction['window'], + 'pg':encode(instruction['progress']),'at':now}) + return True diff --git a/app/repo/pms_repo.py b/app/repo/pms_repo.py index 4f0e6ac..c94c31d 100644 --- a/app/repo/pms_repo.py +++ b/app/repo/pms_repo.py @@ -1096,3 +1096,15 @@ def list_op_log(*, limit: int = 200) -> list: def publish_prepared_proposal(old, candidate, verdict): from app.db.session import publish_prepared_proposal as publish return publish(old, candidate, verdict) + + +def accept_open_proposal(old, hard, instruction, reason): + from app.db.session import accept_open_proposal as accept + return accept(old, hard, instruction, reason) + + +def accepted_open_instruction(proposal_id): + rows = fetch_all("SELECT instruction_id FROM pms_instruction " + "WHERE origin_type='proposal' AND origin_id=:pid AND action='OPEN' ORDER BY id", + {'pid': proposal_id}) + return rows[0]['instruction_id'] if len(rows) == 1 else None diff --git a/app/services/plan_feed.py b/app/services/plan_feed.py index a666868..b357408 100644 --- a/app/services/plan_feed.py +++ b/app/services/plan_feed.py @@ -347,7 +347,7 @@ def _rows(raw, bucket: str) -> list: return out -def parse_plan(payload, *, requested=None) -> dict: +def parse_plan(payload, *, requested=None, allow_empty=False) -> dict: """应答 → 内部结构。缺 date 或两档全空都算变形 (抛 PlanFeedError)。 requested: 本次实际发出的 {top, obs_top} (缺省按上游签名的默认值 20/10 算)。 @@ -360,7 +360,7 @@ def parse_plan(payload, *, requested=None) -> dict: raise PlanFeedError("应答缺 date —— 判不了新鲜度, 按取不到处理") main = _rows(payload.get("main"), BUCKET_MAIN) observe = _rows(payload.get("observe"), BUCKET_OBSERVE) - if not main and not observe: + if not main and not observe and not allow_empty: raise PlanFeedError(f"计划 {date} 主榜与观察档都是空的") state = dict(payload.get('refresh_status') or {}) state['pending_codes'] = [normalize_code(c) for c in state.get('pending_codes') or []] @@ -901,11 +901,32 @@ def get_plan(*, force: bool = False, date=None) -> dict: return plan -def read_current_plan(): +def read_current_plan(*, code=None): """采纳只读最新完整发布版，限时取数，不写接收快照或行业映射。""" p = _params() - plan = fetch(base=p['base'], path=p['path'], timeout=min(3, max(1, p['timeout'])), - extra_params=build_query(p)) + if code is None: + plan = fetch(base=p['base'], path=p['path'], timeout=min(3, max(1, p['timeout'])), + extra_params=build_query(p)) + else: + # 不能回退整榜下载：那会把采纳重新变成慢请求。旧桥须先升级兼容接口。 + import requests + try: + r = requests.get(p['base'].rstrip('/') + '/' + p['path'].strip('/') + '/check', + params={'code': code}, timeout=(1, min(3, max(1, p['timeout'])))) + r.raise_for_status() + payload = r.json() + if (not isinstance(payload, dict) or normalize_code(payload.get('check_code')) != code + or not isinstance(payload.get('found'), bool) + or not (payload.get('decision') or {}).get('id') + or not all(isinstance(payload.get(b), list) for b in ('main', 'observe'))): + raise ValueError('逐票核验应答不完整') + rows = payload['main'] + payload['observe'] + if (len(rows) != int(payload['found']) or + any(normalize_code(row.get('code')) != code for row in rows)): + raise ValueError('逐票核验身份不一致') + plan = parse_plan(payload, allow_empty=True) + except Exception as e: + raise PlanFeedError('暂时无法核对最新资料，本次未采纳，请稍后重试') from e assert_fresh(plan, max_stale_tdays=p['stale_tdays']) return plan diff --git a/app/services/plan_recheck.py b/app/services/plan_recheck.py index 8b94672..80e2339 100644 --- a/app/services/plan_recheck.py +++ b/app/services/plan_recheck.py @@ -3,6 +3,7 @@ from datetime import datetime, timedelta import logging import re from app.core.plan_revision import compare, fingerprint, portfolio_context, confirmation_revision +from app.services.plan_feed import PlanFeedError logger = logging.getLogger('pms.proposal.prepare') @@ -12,7 +13,7 @@ def current_basis(hard, code, proposal_id=None, *, record=True): view = portfolio.positions_view() stock = command_service.effective_stock_params() params = ps._scan_params(view) - options = {'force': True} if record else {'record': False, 'plan_override': plan_feed.read_current_plan()} + options = {'force': True} if record else {'record': False, 'plan_override': plan_feed.read_current_plan(code=code)} selected = plan_feed.candidates(held=[x['ts_code'] for x in view['held']], black=[c for c, v in stock.items() if v.get('black')], route_by_verdict=bool(params.get('open_route_by_verdict')), recheck_code=code, **options) @@ -161,6 +162,10 @@ def before_accept(p, expected_revision): checked = check['provenance'] version_check = dict(checked.get('version_check') or {}, checked_decision=checked.get('decision')) return {'ok': True, 'proposal': dict(p, hard_numbers=dict(hard, version_check=version_check))} + except PlanFeedError as e: + # 传输故障没有改变研究或确认内容。保留已准备状态，不制造后台重算。 + logger.warning('采纳来源核验失败 proposal=%s: %s', p.get('proposal_id'), e, exc_info=True) + return {'ok': False, 'retryable': True, 'error': str(e), 'proposal': p} except Exception: logger.exception('采纳快速检查失败 proposal=%s', p.get('proposal_id')) reason = '当前资料暂不可用，建议保留，后台将自动检查' @@ -182,6 +187,9 @@ def refresh_pending(*, now=None, deadline=None): continue try: result = refresh_one(p, now=now, deadline=deadline) + except PlanFeedError as e: + logger.warning('后台核对暂不可用 proposal=%s: %s', p['proposal_id'], e) + result = {'ok': False, 'error': str(e), 'retryable': True} except Exception: logger.exception('后台更新建议失败 proposal=%s', p['proposal_id']) result = _pending_response(p, '建议更新暂未完成，系统将自动重试', 'source') @@ -206,6 +214,17 @@ def refresh_one(p, *, now=None, deadline=None): return _pending_response(p, check.get('error') or '建议资料暂不可用', check.get('stage', 'source')) if check.get('ok') and (hard.get('preparation') or {}).get('status') == 'ready' and not hard.get('last_accept_check'): return {'ok': True} + # 旧的短暂取数失败不应触发扫描、重新选数和新消费；当前原数量仍过闸即可恢复。 + if ((hard.get('preparation') or {}).get('stage') in ('source', 'research_pending') + and hard.get('consumption_id') and hard.get('proposal_revision') + and hard.get('prepared_judgement') and hard.get('plan_basis') is not None + and not hard.get('last_accept_check') and not check.get('research_changed') + and not check.get('params_changed') and not fast_checks(p, check)): + restored = ready_hard(hard, _judgement(p), now) + candidate = {'qty': p['qty'], 'hard_numbers': restored} + if pms_repo.publish_prepared_proposal(p, candidate, _judgement(p)): + return {'ok': True, 'proposal': dict(p, hard_numbers=restored)} + return {'ok': False, 'error': '建议已被其他操作更新'} research_changed = check.get('research_changed', check.get('material', False)) previous_failure = hard.get('last_accept_check') or {} needs_model = research_changed or previous_failure.get('stage') in ('judge', 'system') diff --git a/app/web/main.py b/app/web/main.py index b3782bd..7c3c6b9 100644 --- a/app/web/main.py +++ b/app/web/main.py @@ -519,6 +519,12 @@ def api_decide(request: Request, proposal_id: str, payload: dict = Body(default= p = pms_repo.get_proposal(proposal_id) if not p: return {"ok": False, "error": "提议不存在"} + if (decision == 'ACCEPTED' and p['status'] == 'ACCEPTED' and p['action'] == 'OPEN' + and (p.get('hard_numbers') or {}).get('kind') != 'strategy'): + existing = pms_repo.accepted_open_instruction(proposal_id) + if existing: + return {'ok': True, 'decision': decision, 'instruction_id': existing, 'already_accepted': True} + return {'ok': False, 'error': '该建议已有采纳记录，但指令记录不完整，请联系维护人员核对'} if p["status"] != "WAIT_USER": check = (p.get('hard_numbers') or {}).get('last_accept_check') or {} if p['status'] == 'EXPIRED' and check.get('status') == 'withdrawn': @@ -550,6 +556,27 @@ def api_decide(request: Request, proposal_id: str, payload: dict = Body(default= return checked expected_hard = p.get('hard_numbers_json') p = checked['proposal'] + if (decision == 'ACCEPTED' and p['action'] == 'OPEN' + and (p.get('hard_numbers') or {}).get('kind') != 'strategy'): + # 先完整准备写入内容，再一次提交。不能先改成已采纳，再独立写账与落指令。 + import hashlib + hn = dict(p.get('hard_numbers') or {}, decide_reason_code=reason_code) + window = param_store.get_int('PMS_EXEC_WINDOW_TDAYS', 3) + iid = cs.make_instruction_id(td.ymd(), p['ts_code'], 'OPEN', 1) + iid += '_' + hashlib.sha256(proposal_id.encode()).hexdigest()[:12] + instruction = {'id': iid, 'window': window, 'progress': { + 'from_proposal': proposal_id, 'is_command': False, 'children': [], + 'plan': {k: hn.get(k) for k in ('decision', 'consumption_id', 'review_ref', + 'execution_rank', 'priority_reason', 'version_check')}, + 'deadline': str(td.window_deadline(None, window))}} + if pms_repo.accept_open_proposal(p, hn, instruction, reason): + return {'ok': True, 'decision': decision, 'instruction_id': iid} + latest = pms_repo.get_proposal(proposal_id) + existing = (pms_repo.accepted_open_instruction(proposal_id) + if latest and latest['status'] == 'ACCEPTED' else None) + if existing: + return {'ok': True, 'decision': decision, 'instruction_id': existing, 'already_accepted': True} + return {'ok': False, 'error': '建议已更新或过期，请查看当前内容', 'proposal': latest} changed = (pms_repo.decide_proposal(proposal_id, decision, expected_hard=expected_hard) if expected_hard is not None else pms_repo.decide_proposal(proposal_id, decision)) if not changed: diff --git a/app/web/static/index.html b/app/web/static/index.html index df65023..7bf6a77 100644 --- a/app/web/static/index.html +++ b/app/web/static/index.html @@ -4431,8 +4431,10 @@ createApp({ ? '尚未收到处理结果。请关闭后刷新，先核对待确认提议和当日指令，勿立即重复提交。' : (d.error || '本次未能采纳，请稍后再试'); if (d.proposal) { + const priorRevision = (dd.row.hard_numbers || {}).proposal_revision; dd.row = { ...dd.row, ...d.proposal }; - dd.updated = !dd.withdrawn && !dd.pending && !((d.proposal.hard_numbers||{}).last_accept_check); + dd.updated = !d.retryable && !dd.withdrawn && !dd.pending && + priorRevision !== (d.proposal.hard_numbers || {}).proposal_revision; if (dd.updated) { dd.code = ''; dd.note = ''; } await loadProposals(); } diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 1b1b391..afd845e 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -5,6 +5,7 @@ 运行: docker compose run --rm pms-web python scripts/run_tests.py 包含 (例数按 2026-10-12 隔离回归实跑校正): + test_batch43_units.py 逐票来源核验、实际API至SQL事务、故障回滚与幂等 (13 例) test_batch42_units.py 提议先准备后采纳、模型复用、并发更新与发布审计 (24 例) test_batch41_units.py 跨流程买卖互斥、清仓收尾、出口撤单与真实进展及人工清仓T+1 (24 例) test_batch40_units.py 逐笔历史修复、进度暂停、并发与事务回滚与均价失效 (22 例) @@ -193,7 +194,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py", "test_batch35_units.py", "test_batch36_units.py", "test_batch37_units.py", - "test_batch38_units.py", "test_batch39_units.py", "test_batch40_units.py", "test_batch41_units.py", "test_batch42_units.py", + "test_batch38_units.py", "test_batch39_units.py", "test_batch40_units.py", "test_batch41_units.py", "test_batch42_units.py", "test_batch43_units.py", "test_page_copy_guard.py", "test_page_enum_guard.py", "test_page_wiring_guard.py", diff --git a/scripts/test_batch43_units.py b/scripts/test_batch43_units.py new file mode 100644 index 0000000..4c1a995 --- /dev/null +++ b/scripts/test_batch43_units.py @@ -0,0 +1,145 @@ +"""开发机：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()