207 lines
7.5 KiB
Python
207 lines
7.5 KiB
Python
"""Behave steps for context manager coverage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.reactive.context_manager import ContextManager
|
|
|
|
|
|
def _write_json(path: Path, data: object) -> None:
|
|
path.write_text(json.dumps(data), encoding="utf-8")
|
|
|
|
|
|
@given("a context manager storage directory with valid JSON files")
|
|
def step_valid_storage_directory(context):
|
|
temp_dir = Path(tempfile.mkdtemp())
|
|
context.context_name = "valid_context"
|
|
context.temp_dir = temp_dir
|
|
context.context_root = temp_dir / context.context_name
|
|
context.context_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
context.expected_messages = [
|
|
{
|
|
"role": "user",
|
|
"content": "hello",
|
|
"timestamp": "2024-01-01T00:00:00",
|
|
"metadata": {},
|
|
},
|
|
{
|
|
"role": "assistant",
|
|
"content": "hi",
|
|
"timestamp": "2024-01-01T00:00:01",
|
|
"metadata": {"tone": "friendly"},
|
|
},
|
|
]
|
|
context.expected_metadata = {
|
|
"created_at": "2024-01-01T00:00:00",
|
|
"last_updated": "2024-01-02T00:00:00",
|
|
"context_name": context.context_name,
|
|
}
|
|
context.expected_state = {"stage": "draft"}
|
|
context.expected_global = {"topic": "coverage"}
|
|
|
|
_write_json(context.context_root / "messages.json", context.expected_messages)
|
|
_write_json(context.context_root / "metadata.json", context.expected_metadata)
|
|
_write_json(context.context_root / "state.json", context.expected_state)
|
|
_write_json(context.context_root / "global_context.json", context.expected_global)
|
|
|
|
|
|
@given("a context manager storage directory with invalid JSON files")
|
|
def step_invalid_storage_directory(context):
|
|
temp_dir = Path(tempfile.mkdtemp())
|
|
context.context_name = "invalid_context"
|
|
context.temp_dir = temp_dir
|
|
context.context_root = temp_dir / context.context_name
|
|
context.context_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
for filename in (
|
|
"messages.json",
|
|
"metadata.json",
|
|
"state.json",
|
|
"global_context.json",
|
|
):
|
|
(context.context_root / filename).write_text("{", encoding="utf-8")
|
|
|
|
|
|
@when("I initialize the context manager from that directory")
|
|
def step_initialize_context_manager(context):
|
|
context.manager = ContextManager(context.context_name, context_dir=context.temp_dir)
|
|
|
|
|
|
@then("the stored context data should be loaded")
|
|
def step_verify_loaded_context(context):
|
|
assert context.manager.messages == context.expected_messages
|
|
assert context.manager.metadata == context.expected_metadata
|
|
assert context.manager.state == context.expected_state
|
|
assert context.manager.global_context == context.expected_global
|
|
|
|
|
|
@then("invalid context files should load as empty defaults")
|
|
def step_verify_invalid_defaults(context):
|
|
assert context.manager.messages == []
|
|
assert context.manager.metadata == {}
|
|
assert context.manager.state == {}
|
|
assert context.manager.global_context == {}
|
|
|
|
|
|
@given("a temporary home directory for the context manager")
|
|
def step_temporary_home(context):
|
|
context.temp_home = Path(tempfile.mkdtemp())
|
|
context.context_name = "home_context"
|
|
|
|
|
|
@when("I initialize the context manager without a context directory")
|
|
def step_initialize_default_context(context):
|
|
with patch("pathlib.Path.home", return_value=context.temp_home):
|
|
context.manager = ContextManager(context.context_name)
|
|
|
|
|
|
@then("the context manager should use the default home context path")
|
|
def step_verify_default_path(context):
|
|
expected = context.temp_home / ".cleveragents" / "context" / context.context_name
|
|
assert context.manager.context_dir == expected
|
|
assert context.manager.context_dir.exists()
|
|
|
|
|
|
@then("deleting the context should remove the directory")
|
|
def step_verify_delete(context):
|
|
context.manager.delete()
|
|
assert not context.manager.context_dir.exists()
|
|
|
|
|
|
@given("a context manager for state and metadata updates")
|
|
def step_state_metadata_manager(context):
|
|
context.temp_dir = Path(tempfile.mkdtemp())
|
|
context.context_name = "state_meta_context"
|
|
context.manager = ContextManager(context.context_name, context_dir=context.temp_dir)
|
|
|
|
|
|
@when("I update state and metadata in the context manager")
|
|
def step_update_state_metadata(context):
|
|
context.manager.update_state("mode", "active")
|
|
context.manager.update_metadata({"owner": "qa"})
|
|
context.state_value = context.manager.get_state("mode")
|
|
context.default_state = context.manager.get_state("missing", "default")
|
|
context.exists_value = context.manager.exists()
|
|
|
|
|
|
@then("the context manager should return the updated values")
|
|
def step_verify_updated_values(context):
|
|
assert context.state_value == "active"
|
|
assert context.default_state == "default"
|
|
assert context.manager.metadata["owner"] == "qa"
|
|
|
|
|
|
@then("the context should report it exists on disk")
|
|
def step_verify_exists(context):
|
|
assert context.exists_value is True
|
|
|
|
|
|
@given("a context manager with stored messages")
|
|
def step_context_with_messages(context):
|
|
context.temp_dir = Path(tempfile.mkdtemp())
|
|
context.context_name = "message_context"
|
|
context.manager = ContextManager(context.context_name, context_dir=context.temp_dir)
|
|
context.manager.add_message("user", "Hello")
|
|
context.manager.add_message("assistant", "Hi")
|
|
context.manager.update_state("counter", 2)
|
|
context.manager.global_context = {"mode": "chat"}
|
|
|
|
|
|
@when("I request the last message and clear the context")
|
|
def step_last_message_and_clear(context):
|
|
context.last_messages = context.manager.get_last_n_messages(1)
|
|
context.manager.clear()
|
|
|
|
|
|
@then("the last message should be returned and the context reset")
|
|
def step_verify_last_and_clear(context):
|
|
assert len(context.last_messages) == 1
|
|
assert context.last_messages[0]["role"] == "assistant"
|
|
assert context.manager.messages == []
|
|
assert context.manager.state == {}
|
|
assert context.manager.global_context == {}
|
|
assert context.manager.metadata["context_name"] == context.context_name
|
|
assert "created_at" in context.manager.metadata
|
|
assert "last_updated" in context.manager.metadata
|
|
|
|
|
|
@given("a context manager with exportable data")
|
|
def step_exportable_context(context):
|
|
context.temp_dir = Path(tempfile.mkdtemp())
|
|
context.context_name = "export_context"
|
|
context.manager = ContextManager(context.context_name, context_dir=context.temp_dir)
|
|
context.manager.add_message("user", "Export this")
|
|
context.manager.update_state("flag", True)
|
|
context.manager.global_context = {"mode": "full"}
|
|
context.manager.metadata["tag"] = "export"
|
|
context.export_file = context.temp_dir / "exports" / "context.json"
|
|
|
|
|
|
@when("I export and import the context")
|
|
def step_export_import_context(context):
|
|
context.manager.export_context(context.export_file)
|
|
context.import_manager = ContextManager(
|
|
"import_context", context_dir=context.temp_dir
|
|
)
|
|
context.import_manager.import_context(context.export_file)
|
|
|
|
|
|
@then("the imported context should match the exported data")
|
|
def step_verify_imported_data(context):
|
|
assert context.export_file.exists()
|
|
assert context.import_manager.messages == context.manager.messages
|
|
assert context.import_manager.state == context.manager.state
|
|
assert context.import_manager.global_context == context.manager.global_context
|
|
assert (
|
|
context.import_manager.metadata["context_name"]
|
|
== context.manager.metadata["context_name"]
|
|
)
|
|
assert context.import_manager.metadata.get("tag") == "export"
|