382 lines
12 KiB
Python
382 lines
12 KiB
Python
# 连接池：thick client 只初始化一次；缺配置抛固定文案；acquire 会归还。
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from rpa_oracle.errors import RepoUnavailable
|
||
|
||
|
||
# 配齐建池所需的三项环境变量。
|
||
def _creds(monkeypatch) -> None:
|
||
monkeypatch.setenv("RPA_ORACLE_USER", "rpainit")
|
||
monkeypatch.setenv("RPA_ORACLE_PASSWORD", "not-for-git")
|
||
monkeypatch.setenv("RPA_ORACLE_HOST", "secret-host.example")
|
||
monkeypatch.setenv("RPA_ORACLE_PORT", "1521")
|
||
monkeypatch.setenv("RPA_ORACLE_SERVICE", "rpa")
|
||
monkeypatch.setenv("RPA_ORACLE_THICK", "1")
|
||
|
||
|
||
# 把假 oracledb 塞进 sys.modules，供 pool 函数内 import。
|
||
def _install_fake_oracledb(monkeypatch, *, init=None, create=None):
|
||
fake = SimpleNamespace(
|
||
init_oracle_client=init or (lambda **kwargs: None),
|
||
create_pool=create or (lambda **kwargs: SimpleNamespace(acquire=lambda: SimpleNamespace(close=lambda: None))),
|
||
)
|
||
monkeypatch.setitem(sys.modules, "oracledb", fake)
|
||
return fake
|
||
|
||
|
||
# 每个用例把 thick client 已初始化标记清掉。
|
||
@pytest.fixture
|
||
def reset_client(monkeypatch):
|
||
import rpa_oracle.pool as pool_mod
|
||
|
||
monkeypatch.setattr(pool_mod, "_client_ready", False)
|
||
yield pool_mod
|
||
monkeypatch.setattr(pool_mod, "_client_ready", False)
|
||
|
||
|
||
# OMS 角色默认池大小 1–4。
|
||
def test_pool_limits_oms_defaults(monkeypatch):
|
||
monkeypatch.delenv("RPA_ORACLE_POOL_MIN", raising=False)
|
||
monkeypatch.delenv("RPA_ORACLE_POOL_MAX", raising=False)
|
||
from rpa_oracle.pool import pool_limits
|
||
|
||
assert pool_limits("oms") == (1, 4)
|
||
|
||
|
||
# 班次角色默认池大小 1–1。
|
||
def test_pool_limits_job_defaults(monkeypatch):
|
||
monkeypatch.delenv("RPA_ORACLE_POOL_MIN", raising=False)
|
||
monkeypatch.delenv("RPA_ORACLE_POOL_MAX", raising=False)
|
||
from rpa_oracle.pool import pool_limits
|
||
|
||
assert pool_limits("job") == (1, 1)
|
||
|
||
|
||
# RPA_ORACLE_POOL_MIN / MAX 覆盖角色默认值。
|
||
def test_pool_limits_env_overrides(monkeypatch):
|
||
monkeypatch.setenv("RPA_ORACLE_POOL_MIN", "2")
|
||
monkeypatch.setenv("RPA_ORACLE_POOL_MAX", "8")
|
||
from rpa_oracle.pool import pool_limits
|
||
|
||
assert pool_limits("oms") == (2, 8)
|
||
assert pool_limits("job") == (2, 8)
|
||
|
||
|
||
# 缺用户时抛「数据库不可用」，文案不含 host。
|
||
def test_missing_user_raises_fixed_message(monkeypatch):
|
||
monkeypatch.delenv("RPA_ORACLE_USER", raising=False)
|
||
monkeypatch.setenv("RPA_ORACLE_PASSWORD", "not-for-git")
|
||
monkeypatch.setenv("RPA_ORACLE_HOST", "secret-host.example")
|
||
from rpa_oracle.pool import create_pool
|
||
|
||
with pytest.raises(RepoUnavailable) as caught:
|
||
create_pool()
|
||
assert caught.value.message == "数据库不可用"
|
||
assert "secret-host" not in str(caught.value)
|
||
assert "ORA-" not in str(caught.value)
|
||
|
||
|
||
# 缺密码时抛「数据库不可用」，文案不含 DSN。
|
||
def test_missing_password_raises_fixed_message(monkeypatch):
|
||
monkeypatch.setenv("RPA_ORACLE_USER", "rpainit")
|
||
monkeypatch.delenv("RPA_ORACLE_PASSWORD", raising=False)
|
||
monkeypatch.setenv("RPA_ORACLE_HOST", "secret-host.example")
|
||
from rpa_oracle.pool import create_pool
|
||
|
||
with pytest.raises(RepoUnavailable) as caught:
|
||
create_pool()
|
||
assert caught.value.message == "数据库不可用"
|
||
assert "secret-host" not in str(caught.value)
|
||
|
||
|
||
# create_pool 调用两次时 init_oracle_client 只跑一次。
|
||
def test_thick_init_runs_once(monkeypatch, reset_client):
|
||
_creds(monkeypatch)
|
||
calls: list[dict] = []
|
||
|
||
# 记下 thick client 初始化参数。
|
||
def init_oracle_client(**kwargs):
|
||
calls.append(kwargs)
|
||
|
||
created: list[dict] = []
|
||
|
||
# 记下建池参数并返回空池替身。
|
||
def create_pool_raw(**kwargs):
|
||
created.append(kwargs)
|
||
return SimpleNamespace(acquire=lambda: SimpleNamespace(close=lambda: None))
|
||
|
||
_install_fake_oracledb(monkeypatch, init=init_oracle_client, create=create_pool_raw)
|
||
from rpa_oracle.pool import create_pool
|
||
|
||
create_pool()
|
||
create_pool()
|
||
assert len(calls) == 1
|
||
assert len(created) == 2
|
||
|
||
|
||
# thick client 初始化失败时对外仍是「数据库不可用」，不含库路径。
|
||
def test_thick_init_failure_hides_lib_path(monkeypatch, reset_client):
|
||
_creds(monkeypatch)
|
||
monkeypatch.setenv("RPA_ORACLE_CLIENT_LIB", "/secret/instantclient")
|
||
|
||
# 假装 Instant Client 加载失败，并把路径写进异常。
|
||
def init_oracle_client(**kwargs):
|
||
raise OSError(f"cannot load {kwargs.get('lib_dir')}")
|
||
|
||
_install_fake_oracledb(monkeypatch, init=init_oracle_client)
|
||
from rpa_oracle.pool import create_pool
|
||
|
||
with pytest.raises(RepoUnavailable) as caught:
|
||
create_pool()
|
||
assert caught.value.message == "数据库不可用"
|
||
assert "instantclient" not in str(caught.value)
|
||
assert "ORA-" not in str(caught.value)
|
||
|
||
|
||
# OMS 建池把 min=1 max=4 传给 oracledb.create_pool。
|
||
def test_create_pool_passes_oms_limits(monkeypatch, reset_client):
|
||
_creds(monkeypatch)
|
||
monkeypatch.delenv("RPA_ORACLE_POOL_MIN", raising=False)
|
||
monkeypatch.delenv("RPA_ORACLE_POOL_MAX", raising=False)
|
||
seen: dict = {}
|
||
|
||
# 记下 OMS 建池参数。
|
||
def create_pool_raw(**kwargs):
|
||
seen.update(kwargs)
|
||
return SimpleNamespace(acquire=lambda: SimpleNamespace(close=lambda: None))
|
||
|
||
_install_fake_oracledb(monkeypatch, create=create_pool_raw)
|
||
from rpa_oracle.pool import create_pool
|
||
|
||
create_pool(role="oms")
|
||
assert seen["min"] == 1
|
||
assert seen["max"] == 4
|
||
assert "secret-host.example" in seen["dsn"]
|
||
assert seen["user"] == "rpainit"
|
||
|
||
|
||
def test_create_pool_passes_job_limits(monkeypatch, reset_client):
|
||
"""班次建池保持 min=1 max=1，并带 60 秒探活。"""
|
||
_creds(monkeypatch)
|
||
monkeypatch.delenv("RPA_ORACLE_POOL_MIN", raising=False)
|
||
monkeypatch.delenv("RPA_ORACLE_POOL_MAX", raising=False)
|
||
seen: dict = {}
|
||
|
||
# 记下班次建池参数。
|
||
def create_pool_raw(**kwargs):
|
||
seen.update(kwargs)
|
||
return SimpleNamespace(acquire=lambda: SimpleNamespace(close=lambda: None))
|
||
|
||
_install_fake_oracledb(monkeypatch, create=create_pool_raw)
|
||
from rpa_oracle.pool import create_pool
|
||
|
||
create_pool(role="job")
|
||
assert seen["min"] == 1
|
||
assert seen["max"] == 1
|
||
assert seen["ping_interval"] == 60
|
||
|
||
|
||
# with acquire() 结束时把连接 close 归还。
|
||
def test_acquire_releases_on_exit(monkeypatch, reset_client):
|
||
_creds(monkeypatch)
|
||
|
||
# 记下 close 是否被调用。
|
||
class FakeConn:
|
||
# 连接尚未归还。
|
||
def __init__(self):
|
||
self.closed = False
|
||
|
||
# 标记已归还。
|
||
def close(self):
|
||
self.closed = True
|
||
|
||
conn = FakeConn()
|
||
|
||
# 每次 acquire 都给出同一条假连接。
|
||
class FakeRawPool:
|
||
# 返回预先准备的假连接。
|
||
def acquire(self):
|
||
return conn
|
||
|
||
_install_fake_oracledb(monkeypatch, create=lambda **kwargs: FakeRawPool())
|
||
from rpa_oracle.pool import create_pool
|
||
|
||
pool = create_pool()
|
||
with pool.acquire() as held:
|
||
assert held is conn
|
||
assert conn.closed is False
|
||
assert conn.closed is True
|
||
|
||
|
||
# with 块里抛错时连接仍归还。
|
||
def test_acquire_releases_after_error(monkeypatch, reset_client):
|
||
_creds(monkeypatch)
|
||
|
||
# 记下 close 是否被调用。
|
||
class FakeConn:
|
||
# 连接尚未归还。
|
||
def __init__(self):
|
||
self.closed = False
|
||
|
||
# 标记已归还。
|
||
def close(self):
|
||
self.closed = True
|
||
|
||
conn = FakeConn()
|
||
|
||
# 每次 acquire 都给出同一条假连接。
|
||
class FakeRawPool:
|
||
# 返回预先准备的假连接。
|
||
def acquire(self):
|
||
return conn
|
||
|
||
_install_fake_oracledb(monkeypatch, create=lambda **kwargs: FakeRawPool())
|
||
from rpa_oracle.pool import create_pool
|
||
|
||
pool = create_pool()
|
||
with pytest.raises(RuntimeError):
|
||
with pool.acquire():
|
||
raise RuntimeError("boom")
|
||
assert conn.closed is True
|
||
|
||
|
||
def test_failed_ping_or_dead_connection_borrows_a_new_one():
|
||
"""探活失败或执行前已死时，下一次借到新连接。"""
|
||
from rpa_oracle.pool import OraclePool
|
||
|
||
class Conn:
|
||
"""假连接。ping 失败或 dead 表示执行前不可用。"""
|
||
|
||
def __init__(self, name, *, alive=True, dead=False):
|
||
"""记下名字和是否还活着。"""
|
||
self.name = name
|
||
self._alive = alive
|
||
self.dead = dead
|
||
self.closed = False
|
||
|
||
def ping(self):
|
||
"""探活。不可用时抛错。"""
|
||
if not self._alive:
|
||
raise OSError("ping failed")
|
||
|
||
def close(self):
|
||
"""标记已归还。"""
|
||
self.closed = True
|
||
|
||
live = Conn("live")
|
||
live_after_dead = Conn("live-after-dead")
|
||
|
||
class Raw:
|
||
"""按队列发连接，并记下被丢掉的名字。"""
|
||
|
||
def __init__(self):
|
||
"""先给出坏连接，再给出活连接。"""
|
||
self.queue = [
|
||
Conn("ping-dead", alive=False),
|
||
live,
|
||
Conn("flag-dead", dead=True),
|
||
live_after_dead,
|
||
]
|
||
self.dropped: list[str] = []
|
||
|
||
def acquire(self):
|
||
"""取出下一条连接。"""
|
||
return self.queue.pop(0)
|
||
|
||
def drop(self, conn):
|
||
"""记下丢掉的连接。"""
|
||
self.dropped.append(conn.name)
|
||
|
||
raw = Raw()
|
||
pool = OraclePool(raw)
|
||
with pool.acquire() as held:
|
||
assert held.name == "live"
|
||
with pool.acquire() as held:
|
||
assert held.name == "live-after-dead"
|
||
assert raw.dropped == ["ping-dead", "flag-dead"]
|
||
assert live.closed is True
|
||
assert live_after_dead.closed is True
|
||
|
||
|
||
def test_uncommitted_failure_retries_that_statement_only():
|
||
"""未提交失败再借一次只重试这一笔。已经 commit 的不重做。"""
|
||
from rpa_oracle.pool import OraclePool
|
||
|
||
class Conn:
|
||
"""假连接，记下 commit 次数。"""
|
||
|
||
def __init__(self, name):
|
||
"""按借出顺序命名。"""
|
||
self.name = name
|
||
self.closed = False
|
||
self.commits = 0
|
||
|
||
def commit(self):
|
||
"""记一次提交。"""
|
||
self.commits += 1
|
||
|
||
def close(self):
|
||
"""标记已归还。"""
|
||
self.closed = True
|
||
|
||
class Raw:
|
||
"""每次借出一条新连接，并记下被丢掉的名字。"""
|
||
|
||
def __init__(self):
|
||
"""从 1 开始编号。"""
|
||
self.n = 0
|
||
self.dropped: list[int] = []
|
||
self.conns: list[Conn] = []
|
||
|
||
def acquire(self):
|
||
"""借出下一条新连接。"""
|
||
self.n += 1
|
||
conn = Conn(self.n)
|
||
self.conns.append(conn)
|
||
return conn
|
||
|
||
def drop(self, conn):
|
||
"""记下丢掉的连接。"""
|
||
self.dropped.append(conn.name)
|
||
|
||
raw = Raw()
|
||
pool = OraclePool(raw)
|
||
seen: list[int] = []
|
||
|
||
def work(conn):
|
||
"""第一笔未提交就失败，重试的那笔才提交。"""
|
||
seen.append(conn.name)
|
||
if conn.name == 1:
|
||
raise RuntimeError("not committed")
|
||
conn.commit()
|
||
|
||
pool.run(work)
|
||
assert seen == [1, 2]
|
||
assert raw.dropped == [1]
|
||
assert raw.conns[0].commits == 0
|
||
assert raw.conns[1].commits == 1
|
||
assert raw.conns[1].closed is True
|
||
|
||
committed_seen: list[str] = []
|
||
|
||
def committed_then_fail(conn):
|
||
"""先提交再失败。这一笔不能重做。"""
|
||
committed_seen.append(f"run-{conn.name}")
|
||
conn.commit()
|
||
committed_seen.append(f"commit-{conn.name}")
|
||
raise RuntimeError("after commit")
|
||
|
||
before = raw.n
|
||
with pytest.raises(RuntimeError, match="after commit"):
|
||
pool.run(committed_then_fail)
|
||
name = before + 1
|
||
assert committed_seen == [f"run-{name}", f"commit-{name}"]
|
||
assert raw.dropped == [1, name]
|
||
assert raw.n == name
|