97 lines
2.5 KiB
Python
97 lines
2.5 KiB
Python
# 班次故障审计：一次 executemany，写失败不抛。
|
||
|
||
from __future__ import annotations
|
||
|
||
from contextlib import contextmanager
|
||
|
||
from rpa_oracle.audit import ShiftError, save_shift_errors
|
||
|
||
|
||
# 记下 executemany 的假游标。
|
||
class _Cursor:
|
||
# 挂上连接，方便失败时由连接决定。
|
||
def __init__(self, conn: "_Conn"):
|
||
self.conn = conn
|
||
|
||
# 进入 with，返回自身。
|
||
def __enter__(self):
|
||
return self
|
||
|
||
# 退出 with，不吞异常。
|
||
def __exit__(self, *args):
|
||
return False
|
||
|
||
# 记下 SQL 和 bind。
|
||
def executemany(self, sql, binds):
|
||
self.conn.many.append((sql, list(binds)))
|
||
|
||
|
||
# 可选择在 commit 时失败的假连接。
|
||
class _Conn:
|
||
# fail 为真时 commit 抛错。
|
||
def __init__(self, fail: bool = False):
|
||
self.fail = fail
|
||
self.many: list = []
|
||
self.commits = 0
|
||
self.rollbacks = 0
|
||
|
||
# 返回假游标。
|
||
def cursor(self):
|
||
return _Cursor(self)
|
||
|
||
# 成功则计数，失败则抛带 ORA 的错。
|
||
def commit(self):
|
||
if self.fail:
|
||
raise RuntimeError("ORA-00001 boom")
|
||
self.commits += 1
|
||
|
||
# 记下回滚。
|
||
def rollback(self):
|
||
self.rollbacks += 1
|
||
|
||
|
||
# 借出同一条假连接。
|
||
class _Pool:
|
||
# 挂上连接。
|
||
def __init__(self, conn: _Conn):
|
||
self.conn = conn
|
||
self.acquires = 0
|
||
|
||
# 借出连接并计数。
|
||
@contextmanager
|
||
def acquire(self):
|
||
self.acquires += 1
|
||
yield self.conn
|
||
|
||
|
||
# 多条故障一次 executemany，超长原文截到 4000。
|
||
def test_save_shift_errors_inserts_once_and_clips():
|
||
pool = _Pool(_Conn())
|
||
save_shift_errors(
|
||
pool,
|
||
[
|
||
ShiftError("加购", "11", "QUERY_FAILED", "网关超时"),
|
||
ShiftError("合同", "SUP1", "PAGE_FAILED", "x" * 4001),
|
||
],
|
||
)
|
||
assert pool.acquires == 1
|
||
assert pool.conn.commits == 1
|
||
sql, binds = pool.conn.many[0]
|
||
assert "rpa_shift_error_audit" in sql.lower()
|
||
assert binds[0]["error_text"] == "网关超时"
|
||
assert len(binds[1]["error_text"]) == 4000
|
||
|
||
|
||
# 审计写入失败不向外抛。
|
||
def test_save_shift_errors_swallows_write_failure():
|
||
pool = _Pool(_Conn(fail=True))
|
||
save_shift_errors(pool, [ShiftError("返利", "HT1", "SUBMIT_FAILED", "ORA-12541")])
|
||
assert pool.conn.rollbacks == 1
|
||
|
||
|
||
# 空列表不借连接。
|
||
def test_save_shift_errors_skips_empty():
|
||
pool = _Pool(_Conn())
|
||
save_shift_errors(pool, [])
|
||
assert pool.acquires == 0
|