454 lines
17 KiB
Python
454 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.gateway import api_read_cache
|
|
|
|
|
|
class FakeRedis:
|
|
def __init__(self) -> None:
|
|
self.values: dict[str, bytes] = {}
|
|
self.expirations: dict[str, int] = {}
|
|
|
|
async def get(self, key: str) -> bytes | None:
|
|
return self.values.get(key)
|
|
|
|
async def set(self, key: str, value: bytes | str, *, ex: int, nx: bool = False) -> bool | None:
|
|
assert ex > 0
|
|
if nx and key in self.values:
|
|
return None
|
|
self.values[key] = value if isinstance(value, bytes) else value.encode("utf-8")
|
|
self.expirations[key] = ex
|
|
return True
|
|
|
|
async def incr(self, key: str) -> int:
|
|
value = int(self.values.get(key, b"0")) + 1
|
|
self.values[key] = str(value).encode("ascii")
|
|
return value
|
|
|
|
async def eval(self, _script: str, _num_keys: int, key: str, token: str) -> int:
|
|
if self.values.get(key) == token.encode("utf-8"):
|
|
del self.values[key]
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def _config(*, enabled: bool = True) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
api_read_cache=SimpleNamespace(
|
|
enabled=enabled,
|
|
url=None,
|
|
password=None,
|
|
namespace="test:homepage",
|
|
ttl_seconds=300,
|
|
max_value_bytes=524288,
|
|
socket_connect_timeout_ms=50,
|
|
socket_timeout_ms=50,
|
|
),
|
|
concurrency=SimpleNamespace(
|
|
redis=SimpleNamespace(
|
|
enabled=enabled,
|
|
url="redis://cache.test/0",
|
|
password=None,
|
|
namespace="test:deerflow",
|
|
socket_connect_timeout_ms=50,
|
|
socket_timeout_ms=50,
|
|
)
|
|
)
|
|
)
|
|
|
|
|
|
def test_redis_cache_ttl_defaults_to_five_minutes_and_stays_bounded() -> None:
|
|
assert api_read_cache._cache_ttl_seconds(SimpleNamespace(api_read_cache=SimpleNamespace())) == 300
|
|
assert api_read_cache._cache_ttl_seconds(
|
|
SimpleNamespace(api_read_cache=SimpleNamespace(ttl_seconds=9999))
|
|
) == 3600
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_round_trip_and_generation_invalidation(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
redis = FakeRedis()
|
|
monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: None)
|
|
monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
|
|
monkeypatch.setattr(api_read_cache, "_client", lambda _config, **_kwargs: (redis, "test:cache"))
|
|
config = _config()
|
|
|
|
await api_read_cache.set_cached_json(config, "agents", "user:u1", {"items": [1, 2]})
|
|
assert await api_read_cache.get_cached_json(config, "agents", "user:u1") == {"items": [1, 2]}
|
|
payload = await api_read_cache.get_cached_json_payload(config, "agents", "user:u1")
|
|
assert payload == b'{"items":[1,2]}'
|
|
|
|
await api_read_cache.invalidate_cached_collection(config, "agents")
|
|
assert await api_read_cache.get_cached_json(config, "agents", "user:u1") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_cache_misses_wait_for_single_fill(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
redis = FakeRedis()
|
|
monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: None)
|
|
monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
|
|
monkeypatch.setattr(api_read_cache, "_client", lambda _config, **_kwargs: (redis, "test:cache"))
|
|
config = _config()
|
|
_, owner_token = await api_read_cache.wait_for_cached_json_or_claim_fill(
|
|
config, "agents", "user:u1", version="0"
|
|
)
|
|
assert owner_token is not None
|
|
|
|
waiter = asyncio.create_task(
|
|
api_read_cache.wait_for_cached_json_or_claim_fill(
|
|
config, "agents", "user:u1", version="0", wait_ms=500
|
|
)
|
|
)
|
|
await asyncio.sleep(0.05)
|
|
await api_read_cache.set_cached_json(
|
|
config, "agents", "user:u1", {"items": ["shared fill"]}, expected_version="0"
|
|
)
|
|
await api_read_cache.release_cache_fill(config, "agents", "user:u1", owner_token)
|
|
cached, waiter_token = await waiter
|
|
assert cached == {"items": ["shared fill"]}
|
|
assert waiter_token is None
|
|
assert not any(":fill:" in key for key in redis.values)
|
|
|
|
stale_version = await api_read_cache.get_cache_version(config, "agents")
|
|
await api_read_cache.invalidate_cached_collection(config, "agents")
|
|
await api_read_cache.set_cached_json(
|
|
config,
|
|
"agents",
|
|
"user:u1",
|
|
{"items": ["stale"]},
|
|
expected_version=stale_version,
|
|
)
|
|
assert await api_read_cache.get_cached_json(config, "agents", "user:u1") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_http_payload_misses_wait_for_single_fill(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
redis = FakeRedis()
|
|
monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: None)
|
|
monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
|
|
monkeypatch.setattr(api_read_cache, "_client", lambda _config, **_kwargs: (redis, "test:cache"))
|
|
config = _config()
|
|
_, owner_token = await api_read_cache.wait_for_cached_json_payload_or_claim_fill(
|
|
config, "skills", "user:u1", version="0"
|
|
)
|
|
assert owner_token is not None
|
|
|
|
waiter = asyncio.create_task(
|
|
api_read_cache.wait_for_cached_json_payload_or_claim_fill(
|
|
config, "skills", "user:u1", version="0", wait_ms=500
|
|
)
|
|
)
|
|
await asyncio.sleep(0.05)
|
|
await api_read_cache.set_cached_json(
|
|
config,
|
|
"skills",
|
|
"user:u1",
|
|
{"skills": [{"name": "one"}]},
|
|
expected_version="0",
|
|
)
|
|
await api_read_cache.release_cache_fill(config, "skills", "user:u1", owner_token)
|
|
payload, waiter_token = await waiter
|
|
assert payload == b'{"skills":[{"name":"one"}]}'
|
|
assert waiter_token is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_redis_uses_the_database_path() -> None:
|
|
config = _config(enabled=True)
|
|
config.api_read_cache = SimpleNamespace(enabled=False)
|
|
assert api_read_cache._redis_settings(config) is None
|
|
api_read_cache.set_api_read_cache_enabled_local(False)
|
|
assert await api_read_cache.get_cached_json(config, "skills", "user:u1") is None
|
|
await api_read_cache.set_cached_json(config, "skills", "user:u1", {"items": []})
|
|
|
|
|
|
def test_homepage_cache_can_reuse_a_separate_redis_connection() -> None:
|
|
config = _config(enabled=False)
|
|
config.api_read_cache = SimpleNamespace(
|
|
enabled=True,
|
|
url="redis://homepage-cache.test/2",
|
|
password="secret",
|
|
namespace="test:homepage",
|
|
)
|
|
assert api_read_cache._redis_settings(config) == (
|
|
"redis://homepage-cache.test/2",
|
|
"secret",
|
|
"test:homepage",
|
|
)
|
|
|
|
|
|
def test_enabled_homepage_cache_reuses_concurrency_redis() -> None:
|
|
config = _config(enabled=True)
|
|
config.api_read_cache = SimpleNamespace(
|
|
enabled=True,
|
|
url=None,
|
|
password=None,
|
|
namespace="test:homepage",
|
|
)
|
|
assert api_read_cache._redis_settings(config) == (
|
|
"redis://cache.test/0",
|
|
None,
|
|
"test:homepage",
|
|
)
|
|
|
|
|
|
def test_separate_cache_redis_does_not_inherit_an_unrelated_password() -> None:
|
|
config = _config(enabled=True)
|
|
config.concurrency.redis.password = "concurrency-secret"
|
|
config.api_read_cache = SimpleNamespace(
|
|
enabled=True,
|
|
url="redis://homepage-cache.test/2",
|
|
password=None,
|
|
namespace="test:homepage",
|
|
)
|
|
assert api_read_cache._redis_settings(config) == (
|
|
"redis://homepage-cache.test/2",
|
|
None,
|
|
"test:homepage",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_cache_switch_and_ttl_are_persisted_and_survive_process_refresh(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from deerflow.persistence.api_read_cache_settings.model import ApiReadCacheSettingRow
|
|
|
|
class FakeSession:
|
|
def __init__(self) -> None:
|
|
self.row: ApiReadCacheSettingRow | None = None
|
|
self.added: list[ApiReadCacheSettingRow] = []
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
async def get(self, _model, key: str, **_kwargs):
|
|
assert key == "default"
|
|
return self.row
|
|
|
|
def add(self, row: ApiReadCacheSettingRow) -> None:
|
|
self.added.append(row)
|
|
self.row = row
|
|
|
|
async def commit(self) -> None:
|
|
return None
|
|
|
|
session = FakeSession()
|
|
monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: lambda: session)
|
|
monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
|
|
config = _config(enabled=False)
|
|
|
|
assert await api_read_cache.get_api_read_cache_settings(config, force_refresh=True) == (False, 300)
|
|
await api_read_cache.save_api_read_cache_settings(True, 900, config=config)
|
|
assert session.row is not None and session.row.enabled is True
|
|
assert session.row.ttl_seconds == 900
|
|
assert await api_read_cache.get_api_read_cache_settings(config) == (True, 900)
|
|
|
|
# A new process has no in-memory value and reads the persisted MySQL row.
|
|
monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
|
|
assert await api_read_cache.get_api_read_cache_settings(config, force_refresh=True) == (True, 900)
|
|
await api_read_cache.save_api_read_cache_enabled(False)
|
|
assert session.row.enabled is False
|
|
assert session.row.ttl_seconds == 900
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_toggle_mysql_read_is_singleflight_under_concurrency(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
class FakeSession:
|
|
def __init__(self) -> None:
|
|
self.get_calls = 0
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
async def get(self, _model, _key):
|
|
self.get_calls += 1
|
|
await asyncio.sleep(0.01)
|
|
return None
|
|
|
|
session = FakeSession()
|
|
monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: lambda: session)
|
|
monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
|
|
config = _config(enabled=False)
|
|
|
|
results = await asyncio.gather(
|
|
*(api_read_cache.is_api_read_cache_enabled(config) for _ in range(70))
|
|
)
|
|
|
|
assert results == [False] * 70
|
|
assert session.get_calls == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_configured_mysql_ttl_is_used_for_agent_and_skill_cache_writes(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
redis = FakeRedis()
|
|
|
|
class FakeSession:
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_args):
|
|
return False
|
|
|
|
async def get(self, _model, _key):
|
|
return SimpleNamespace(enabled=True, ttl_seconds=900)
|
|
|
|
monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: lambda: FakeSession())
|
|
monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
|
|
monkeypatch.setattr(api_read_cache, "_client", lambda _config, **_kwargs: (redis, "test:cache"))
|
|
config = _config()
|
|
|
|
for collection in ("agents", "skills"):
|
|
await api_read_cache.set_cached_json(config, collection, "user:u1", {"items": []})
|
|
expirations = [
|
|
expiry
|
|
for key, expiry in redis.expirations.items()
|
|
if f":{collection}:v0:" in key
|
|
]
|
|
assert expirations == [900]
|
|
|
|
|
|
def test_existing_api_cache_setting_receives_five_minute_ttl(tmp_path: Path) -> None:
|
|
import sqlalchemy as sa
|
|
from alembic import command
|
|
from alembic.config import Config as AlembicConfig
|
|
|
|
database = tmp_path / "api-read-cache.db"
|
|
engine = sa.create_engine(f"sqlite:///{database.resolve().as_posix()}")
|
|
with engine.begin() as connection:
|
|
connection.exec_driver_sql(
|
|
"CREATE TABLE api_read_cache_settings ("
|
|
"`key` VARCHAR(32) PRIMARY KEY, enabled BOOLEAN NOT NULL, updated_at DATETIME NOT NULL)"
|
|
)
|
|
connection.exec_driver_sql(
|
|
"INSERT INTO api_read_cache_settings (`key`, enabled, updated_at) "
|
|
"VALUES ('default', 1, CURRENT_TIMESTAMP)"
|
|
)
|
|
connection.exec_driver_sql(
|
|
"CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL PRIMARY KEY)"
|
|
)
|
|
connection.exec_driver_sql(
|
|
"INSERT INTO alembic_version (version_num) VALUES ('20261026_02')"
|
|
)
|
|
engine.dispose()
|
|
|
|
migrations = Path(__file__).parents[1] / "packages/harness/deerflow/persistence/migrations"
|
|
config = AlembicConfig(str(migrations / "alembic.ini"))
|
|
config.set_main_option(
|
|
"sqlalchemy.url",
|
|
f"sqlite+aiosqlite:///{database.resolve().as_posix()}",
|
|
)
|
|
command.upgrade(config, "head")
|
|
|
|
engine = sa.create_engine(f"sqlite:///{database.resolve().as_posix()}")
|
|
with engine.connect() as connection:
|
|
setting = connection.execute(
|
|
sa.text("SELECT enabled, ttl_seconds, collection_overrides, browser_status_enabled, cache_epoch FROM api_read_cache_settings WHERE `key`='default'")
|
|
).one()
|
|
revision = connection.execute(sa.text("SELECT version_num FROM alembic_version")).scalar_one()
|
|
engine.dispose()
|
|
assert bool(setting.enabled) is True
|
|
assert setting.ttl_seconds == 300
|
|
assert setting.collection_overrides is None
|
|
assert bool(setting.browser_status_enabled) is True
|
|
assert setting.cache_epoch == "0"
|
|
assert revision == "20261102_02"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_catalog_projection_uses_compact_rows_without_long_fields(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from sqlalchemy import update
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.gateway import skill_catalog_cache
|
|
from deerflow.persistence.skills.model import SkillRow
|
|
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(SkillRow.__table__.create)
|
|
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
monkeypatch.setattr(skill_catalog_cache, "get_session_factory", lambda: session_factory)
|
|
monkeypatch.setattr(skill_catalog_cache, "get_or_new_skill_storage", lambda **_kwargs: SimpleNamespace())
|
|
monkeypatch.setattr(skill_catalog_cache, "_load_usage_entries", lambda: {})
|
|
|
|
skill_file = tmp_path / "SKILL.md"
|
|
skill_file.write_text("---\nname: catalog-demo\n---\n", encoding="utf-8")
|
|
skill = SimpleNamespace(
|
|
name="catalog-demo",
|
|
description="compact metadata",
|
|
license="MIT",
|
|
category="public",
|
|
enabled=True,
|
|
always_on=False,
|
|
skill_file=skill_file,
|
|
)
|
|
|
|
assert await skill_catalog_cache.refresh_skill_catalog_projection(skills=[skill]) == 1
|
|
async with session_factory() as session:
|
|
await session.execute(update(SkillRow).where(SkillRow.name == "catalog-demo").values(detail="x" * 5000))
|
|
await session.commit()
|
|
|
|
rows = await skill_catalog_cache.load_skill_catalog_projection(user_id="user-1")
|
|
assert rows is not None and len(rows) == 1
|
|
assert rows[0]["name"] == "catalog-demo"
|
|
assert rows[0]["description"] == "compact metadata"
|
|
assert "detail" not in rows[0]
|
|
assert "detail" not in {column.key for column in skill_catalog_cache._SKILL_LIST_COLUMNS}
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_catalog_schema_adds_columns_to_existing_table_without_losing_data(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from sqlalchemy import inspect
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from app.gateway import skill_catalog_cache
|
|
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
async with engine.begin() as connection:
|
|
await connection.exec_driver_sql(
|
|
"CREATE TABLE skills ("
|
|
"name VARCHAR(64) PRIMARY KEY, owner_user_id VARCHAR(64), "
|
|
"published BOOLEAN NOT NULL DEFAULT 0, detail TEXT, name_zh VARCHAR(128), "
|
|
"always_on BOOLEAN NOT NULL DEFAULT 0, square_id VARCHAR(64) NOT NULL DEFAULT '', "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)"
|
|
)
|
|
await connection.exec_driver_sql(
|
|
"INSERT INTO skills (name, detail, created_at, updated_at) "
|
|
"VALUES ('existing-skill', 'keep this annotation', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
|
|
)
|
|
monkeypatch.setattr(skill_catalog_cache, "get_engine", lambda: engine)
|
|
|
|
await skill_catalog_cache.ensure_skill_catalog_schema()
|
|
# The migration must be safe to retry after process restarts.
|
|
await skill_catalog_cache.ensure_skill_catalog_schema()
|
|
|
|
async with engine.connect() as connection:
|
|
columns = await connection.run_sync(
|
|
lambda sync_connection: {column["name"] for column in inspect(sync_connection).get_columns("skills")}
|
|
)
|
|
preserved = (
|
|
await connection.exec_driver_sql(
|
|
"SELECT detail, available, category, detail_preview FROM skills WHERE name='existing-skill'"
|
|
)
|
|
).one()
|
|
assert "detail_preview" in columns
|
|
assert preserved[0] == "keep this annotation"
|
|
assert bool(preserved[1]) is True
|
|
assert preserved[2] == "custom"
|
|
assert preserved[3] == "keep this annotation"
|
|
await engine.dispose()
|