muse-agent-example/tests/单元/test_数据库连接生命周期.py

"""provider的装配、借还与失败合同；真实池并发另由隔离PG验证。"""

from types import SimpleNamespace

import pytest

from muse.共享.调用身份 import 用途
from muse.基础设施.数据库.连接 import 数据库工厂, 连接池未就绪, 连接池耗尽
from muse.配置 import 数据库引用


class 假连接:
    def __init__(self):
        self.closed = False
        self.read_only = False
        self.autocommit = True
        self.isolation_level = None
        self.deferrable = False
        self.commits = self.rollbacks = 0
        self.sql = []

    def execute(self, sql, params=()):
        self.sql.append((sql, params))
        return self

    def fetchone(self):
        return ("muse_app", "muse_app")

    def commit(self):
        self.commits += 1

    def rollback(self):
        self.rollbacks += 1


class 假驱动池:
    def __init__(self, **kwargs):
        self.kwargs = kwargs
        self.conn = 假连接()
        self.closed = False
        self.borrowed = 0

    def open(self):
        self.kwargs["configure"](self.conn)

    def close(self):
        self.closed = True

    def getconn(self, *, timeout):
        assert timeout == 2.0
        self.borrowed += 1
        return self.conn

    def putconn(self, conn):
        self.borrowed -= 1
        self.kwargs["reset"](conn)


@pytest.fixture
def provider(monkeypatch):
    monkeypatch.setenv("O06_DB", "synthetic-not-a-dsn")
    monkeypatch.setattr(
        "muse.基础设施.数据库.连接.import_module",
        lambda _: SimpleNamespace(ConnectionPool=假驱动池),
    )
    return 数据库工厂(数据库引用("环境变量", "O06_DB"), 用途.生产, 常驻=True)


@pytest.mark.case_id("NC-O06-POOL-LIFECYCLE")
def test_构建不解析凭据且生命周期有界配置(provider, monkeypatch):
    monkeypatch.delenv("O06_DB")
    assert provider._池 is None
    with pytest.raises(连接池未就绪):
        with provider.连接():
            pass
    monkeypatch.setenv("O06_DB", "synthetic-not-a-dsn")
    provider.打开()
    pool = provider._池
    provider.打开()
    assert provider._池 is pool
    assert (pool.kwargs["min_size"], pool.kwargs["max_size"], pool.kwargs["timeout"]) == (0, 4, 2)
    with provider.连接(只读=True) as conn:
        assert conn.read_only and not conn.autocommit
        assert pool.borrowed == 1
    assert pool.borrowed == 0
    assert not conn.read_only and conn.autocommit
    with provider.连接(autocommit=True) as conn:
        assert not conn.read_only and conn.autocommit
    provider.关闭()
    assert pool.closed
    assert provider._池 is None


@pytest.mark.case_id("NC-O06-POOL-ROLLBACK")
def test_业务异常回滚并归还用途状态(provider):
    provider.打开()
    pool = provider._池
    with pytest.raises(ValueError, match="business"):
        with provider.连接(只读=True) as conn:
            conn.isolation_level = "modified"
            conn.deferrable = True
            raise ValueError("business")
    assert pool.borrowed == 0 and conn.rollbacks >= 1
    assert (conn.read_only, conn.autocommit, conn.isolation_level, conn.deferrable) == (
        False,
        True,
        None,
        False,
    )
    assert conn.sql[-1] == ("SELECT set_config('app.purpose', %s, false)", ("production",))
    provider.关闭()


@pytest.mark.case_id("NC-O06-POOL-TIMEOUT")
def test_借用超时独立错误且不翻译业务错误(provider, monkeypatch):
    class PoolTimeout(Exception):
        pass

    provider.打开()

    def exhausted(**_):
        raise PoolTimeout()

    monkeypatch.setattr(provider._池, "getconn", exhausted)
    with pytest.raises(连接池耗尽):
        with provider.连接():
            pass
    assert provider._池.borrowed == 0
    provider.关闭()


@pytest.mark.case_id("NC-O06-ASSEMBLY-OWNER")
def test_运行助手共用单个数据库owner(monkeypatch):
    from muse.启动 import 构建
    from muse.配置 import 应用配置

    app = 构建(应用配置(数据库引用("环境变量", "MISSING_O06"), "synthetic"))
    db = app.要求数据库()
    assert app.要求评测().数据库 is db
    assert app.要求证据().数据库 is db
    assert app.要求原文().数据库 is db
    assert app.作品.数据库 is db and app.正文.数据库 is db and app.任务运行.数据库 is db
    assert db._池 is None


@pytest.mark.case_id("NC-O06-LIFESPAN-CLEANUP")
def test_装配生命周期异常也归还并关闭(provider):
    from muse.启动 import 构建
    from muse.配置 import 应用配置

    app = 构建(应用配置(provider.引用, "synthetic"))
    app.数据库 = provider
    with pytest.raises(ValueError, match="execution"):
        with app.生命周期():
            pool = provider._池
            with provider.连接():
                raise ValueError("execution")
    assert pool.closed and pool.borrowed == 0
    assert provider._池 is None


@pytest.mark.case_id("NC-O06-POOL-INIT-FAILURE")
def test_池打开失败关闭已创建对象(provider, monkeypatch):
    pools = []

    class BrokenPool(假驱动池):
        def open(self):
            pools.append(self)
            raise RuntimeError("initialization")

    monkeypatch.setattr(
        "muse.基础设施.数据库.连接.import_module",
        lambda _: SimpleNamespace(ConnectionPool=BrokenPool),
    )
    with pytest.raises(RuntimeError, match="initialization"):
        provider.打开()
    assert len(pools) == 1 and pools[0].closed
    assert provider._池 is None


@pytest.mark.case_id("NC-O06-POOL-CREDENTIAL-FREEZE")
def test_打开空池不读凭据且首次借用冻结引用(provider, monkeypatch):
    reads = []

    def credential(_):
        reads.append("first")
        return "synthetic-first"

    monkeypatch.setattr("muse.基础设施.数据库.连接.解析连接串", credential)
    provider.打开()
    pool = provider._池
    assert reads == []
    with provider.连接():
        assert pool.kwargs["conninfo"]() == "synthetic-first"
    monkeypatch.setattr(
        "muse.基础设施.数据库.连接.解析连接串", lambda _: pytest.fail("重复读取凭据")
    )
    with provider.连接():
        assert pool.kwargs["conninfo"]() == "synthetic-first"
    assert reads == ["first"]
    provider.关闭()