harness_zbt/backend/tests/integration/test_real_redis.py

from datetime import UTC, datetime

import pytest
from redis import Redis

from zbt.commands.reset_cache import clear_project_cache
from zbt.core.config import Settings
from zbt.harness.events import RuntimeEvent
from zbt.infrastructure.redis.agent_state import RedisAgentStateStore
from zbt.infrastructure.redis.keys import RedisKeyBuilder

settings = Settings()
pytestmark = pytest.mark.skipif(
    settings.app_env != "test" or not settings.redis_prefix.startswith("ins:s3:test:"),
    reason="真实Redis集成测试只允许使用白名单测试前缀",
)


def test_real_redis_prefix_round_trip_and_scoped_cleanup() -> None:
    client = Redis.from_url(settings.redis_url, decode_responses=True)
    keys = RedisKeyBuilder(settings.redis_prefix)
    session_key = keys.auth_session("integration-session")
    product_key = keys.product_cache("integration-products")
    adjacent_key = "ins:s3:test:adjacent:sentinel"

    try:
        client.set(session_key, "active", ex=60)
        client.set(product_key, "cached", ex=60)
        client.set(adjacent_key, "keep", ex=60)

        assert client.get(session_key) == "active"
        assert client.get(product_key) == "cached"
        assert clear_project_cache(settings) == 2
        assert client.get(session_key) is None
        assert client.get(product_key) is None
        assert client.get(adjacent_key) == "keep"
    finally:
        client.delete(session_key, product_key, adjacent_key)
        client.close()


def test_real_redis_agent_hot_state_rate_limit_and_events() -> None:
    store = RedisAgentStateStore(settings)
    keys = RedisKeyBuilder(settings.redis_prefix)
    thread_id = "integration-thread"
    run_id = "integration-run"
    principal_id = "integration-user"
    thread_key = keys.agent_thread_hot(thread_id)
    rate_key = keys.agent_rate(principal_id)
    events_key = keys.agent_run_events(run_id)
    client = Redis.from_url(settings.redis_url, decode_responses=True)
    try:
        store.touch_thread(
            thread_id,
            owner_type="H5_USER",
            owner_id=principal_id,
            persona="customer",
        )
        assert store.allow_request(principal_id)
        store.append_event(
            RuntimeEvent(
                event="run.started",
                run_id=run_id,
                sequence=1,
                occurred_at=datetime.now(UTC),
                data={"persona": "customer"},
            )
        )

        assert client.ttl(thread_key) > 0
        assert client.ttl(rate_key) > 0
        events = store.list_events(run_id)
        assert len(events) == 1
        assert events[0].event == "run.started"
        assert client.ttl(events_key) > 0
    finally:
        client.delete(thread_key, rate_key, events_key)
        store.close()
        client.close()