script-build-and-framework-tmp/visualization/backend/tests/test_runtime_bridge.py

from types import SimpleNamespace

import pytest
from sqlalchemy import create_engine

from app import runtime_bridge


def test_database_host_override_is_host_only(monkeypatch):
    monkeypatch.setenv("PATTERN_DB_HOST_OVERRIDE", "192.168.202.204")
    assert runtime_bridge.database_host_override() == "192.168.202.204"

    for invalid in ("mysql://host", "host:3306", "user@host", "host name"):
        monkeypatch.setenv("PATTERN_DB_HOST_OVERRIDE", invalid)
        with pytest.raises(ValueError):
            runtime_bridge.database_host_override()


def test_override_factory_replaces_only_host_and_keeps_parent_database(monkeypatch):
    class FakeDatabaseManager:
        def __init__(self):
            self.engine = create_engine("mysql+pymysql://reader:secret@old-host:3306/open_aigc_pattern")

    monkeypatch.setattr(
        runtime_bridge,
        "load_runtime_modules",
        lambda: (SimpleNamespace(DatabaseManager=FakeDatabaseManager), SimpleNamespace()),
    )
    runtime_bridge._read_only_session_factory.cache_clear()
    factory = runtime_bridge._read_only_session_factory("/runtime", "192.168.202.204")
    engine = factory.kw["bind"]
    try:
        assert engine.url.host == "192.168.202.204"
        assert engine.url.port == 3306
        assert engine.url.database == "open_aigc_pattern"
        assert engine.url.username == "reader"
        assert engine.url.password == "secret"
    finally:
        engine.dispose()
        runtime_bridge._read_only_session_factory.cache_clear()


def test_read_only_factory_keeps_original_host_without_override(monkeypatch):
    class FakeDatabaseManager:
        def __init__(self):
            self.engine = create_engine("mysql+pymysql://reader:secret@db-host:3306/open_aigc_pattern")

    monkeypatch.setattr(
        runtime_bridge,
        "load_runtime_modules",
        lambda: (SimpleNamespace(DatabaseManager=FakeDatabaseManager), SimpleNamespace()),
    )
    runtime_bridge._read_only_session_factory.cache_clear()
    factory = runtime_bridge._read_only_session_factory("/runtime", "")
    engine = factory.kw["bind"]
    try:
        assert engine.url.host == "db-host"
        assert engine.url.database == "open_aigc_pattern"
    finally:
        engine.dispose()
        runtime_bridge._read_only_session_factory.cache_clear()


def test_read_only_hook_executes_only_session_guard():
    statements = []

    class Cursor:
        def execute(self, statement):
            statements.append(statement)

        def close(self):
            statements.append("closed")

    runtime_bridge._set_mysql_session_read_only(SimpleNamespace(cursor=lambda: Cursor()), None)
    assert statements == ["SET SESSION TRANSACTION READ ONLY", "closed"]