cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(resource): implement DatabaseResourceHandler CRUD and checkpoint methods #1293

Merged
freemo merged 1 commits from feature/database-handler-completion into master 2026-04-02 17:08:06 +00:00
Conversation 2 Commits 1 Files Changed 3
+1333
3 changed files with 1333 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/database_handler_crud.feature
+219
View File
@@ -0,0 +1,219 @@
Feature: DatabaseResourceHandler CRUD and checkpoint methods
As a CleverAgents developer
I want DatabaseResourceHandler to implement full CRUD and checkpoint methods
So that SQLite databases can be read, written, deleted, listed, diffed, and checkpointed
Issue #1241: DatabaseResourceHandler CRUD and checkpoint methods
# ============================================================
# read() — SQLite schema query
# ============================================================
Scenario: read() returns schema for SQLite database with tables
Given a SQLite database with tables "users" and "orders"
When I read from the database handler
Then the db read content should contain "users"
And the db read content should contain "orders"
And the db read content encoding should be "utf-8"
And the db read content hash should not be empty
Scenario: read() returns empty schema for empty SQLite database
Given an empty SQLite database file
When I read from the database handler
Then the db read content encoding should be "utf-8"
And the db read content data should be empty or whitespace
Scenario: read() returns connection-info summary for remote database
Given a remote postgres database resource with location "postgresql://localhost/mydb"
When I read from the database handler
Then the db read content should contain "postgres"
And the db read content should contain "mydb"
And the db read content encoding should be "utf-8"
Scenario: read() returns empty content for resource with no location
Given a database resource with no location
When I read from the database handler
Then the db read content data should be empty or whitespace
Scenario: read() handles SQLite error gracefully
Given a database resource pointing to a non-existent SQLite path
When I read from the database handler
Then no db exceptions should have been raised
# ============================================================
# write() — SQLite SQL execution
# ============================================================
Scenario: write() executes SQL on SQLite database
Given a SQLite database with table "items"
When I write SQL "INSERT INTO items (id, name) VALUES (1, 'apple')" to the database handler
Then the db write result should be successful
And the SQLite table "items" should contain 1 row
Scenario: write() creates a new table via SQL
Given an empty SQLite database file
When I write SQL "CREATE TABLE products (id INTEGER PRIMARY KEY, name TEXT)" to the database handler
Then the db write result should be successful
And the SQLite database should have table "products"
Scenario: write() returns not-supported for remote postgres database
Given a remote postgres database resource with location "postgresql://localhost/mydb"
When I write SQL "SELECT 1" to the database handler
Then the db write result should not be successful
And the db write result message should mention "not supported"
Scenario: write() returns not-supported for remote mysql database
Given a remote mysql database resource with location "mysql://localhost/mydb"
When I write SQL "SELECT 1" to the database handler
Then the db write result should not be successful
And the db write result message should mention "not supported"
Scenario: write() returns failure for resource with no location
Given a database resource with no location
When I write SQL "SELECT 1" to the database handler
Then the db write result should not be successful
Scenario: write() returns failure for invalid SQL
Given an empty SQLite database file
When I write SQL "THIS IS NOT SQL" to the database handler
Then the db write result should not be successful
And the db write result message should mention "error"
# ============================================================
# delete() — SQLite DROP TABLE
# ============================================================
Scenario: delete() drops a table from SQLite database
Given a SQLite database with table "temp_data"
When I delete table "temp_data" from the database handler
Then the db delete result should be successful
And the SQLite database should not have table "temp_data"
Scenario: delete() with non-existent table succeeds (DROP TABLE IF EXISTS)
Given an empty SQLite database file
When I delete table "ghost_table" from the database handler
Then the db delete result should be successful
Scenario: delete() with empty path returns failure
Given an empty SQLite database file
When I delete with empty path from the database handler
Then the db delete result should not be successful
And the db delete result message should mention "table name"
Scenario: delete() returns not-supported for remote postgres database
Given a remote postgres database resource with location "postgresql://localhost/mydb"
When I delete table "users" from the database handler
Then the db delete result should not be successful
And the db delete result message should mention "not supported"
Scenario: delete() returns failure for resource with no location
Given a database resource with no location
When I delete table "users" from the database handler
Then the db delete result should not be successful
# ============================================================
# list_children() — SQLite tables/views
# ============================================================
Scenario: list_children() returns table names for SQLite database
Given a SQLite database with tables "alpha" and "beta"
When I list children from the database handler
Then the db children list should contain "alpha"
And the db children list should contain "beta"
And the db children list should be sorted
Scenario: list_children() returns empty list for empty SQLite database
Given an empty SQLite database file
When I list children from the database handler
Then the db children list should be empty
Scenario: list_children() returns empty list for remote database
Given a remote postgres database resource with location "postgresql://localhost/mydb"
When I list children from the database handler
Then the db children list should be empty
Scenario: list_children() returns empty list for resource with no location
Given a database resource with no location
When I list children from the database handler
Then the db children list should be empty
# ============================================================
# diff() — schema hash comparison
# ============================================================
Scenario: diff() detects no changes between identical SQLite databases
Given a SQLite database with table "same_table"
And a second SQLite database with the same schema
When I diff the database handler against the second database
Then the db diff result should have no changes
Scenario: diff() detects changes between different SQLite databases
Given a SQLite database with table "table_a"
And a second SQLite database with table "table_b"
When I diff the database handler against the second database
Then the db diff result should have changes
Scenario: diff() compares remote database by identity hash
Given a remote postgres database resource with location "postgresql://localhost/db1"
And a second remote database location "postgresql://localhost/db2"
When I diff the database handler against the second remote location
Then the db diff result should have changes
# ============================================================
# create_checkpoint() — SQLite SAVEPOINT
# ============================================================
Scenario: create_checkpoint() creates a SAVEPOINT on SQLite database
Given a SQLite database with table "checkpoint_test"
And a mock sandbox manager
When I create a checkpoint with plan "PLAN-CKPT-001" on the database handler
Then the db checkpoint result should have a checkpoint_id
And the db checkpoint result plan_id should be "PLAN-CKPT-001"
And the db checkpoint result message should mention "SAVEPOINT"
Scenario: create_checkpoint() returns content-hash checkpoint for remote database
Given a remote postgres database resource with location "postgresql://localhost/mydb"
And a mock sandbox manager
When I create a checkpoint with plan "PLAN-CKPT-002" on the database handler
Then the db checkpoint result should have a checkpoint_id
And the db checkpoint result message should mention "hash"
# ============================================================
# rollback_to() — SQLite ROLLBACK TO SAVEPOINT
# ============================================================
Scenario: rollback_to() rolls back SQLite changes to SAVEPOINT
Given a SQLite database with table "rollback_test"
And a mock sandbox manager
When I create a checkpoint with plan "PLAN-RB-001" on the database handler
And I insert a row into "rollback_test" via the database handler
And I rollback to the last checkpoint on the database handler
Then the db rollback result should be successful
And the SQLite table "rollback_test" should contain 0 rows
Scenario: rollback_to() returns failure for unknown checkpoint_id
Given a SQLite database with table "rollback_test2"
And a mock sandbox manager
When I rollback to checkpoint "nonexistent-ckpt" on the database handler
Then the db rollback result should not be successful
And the db rollback result message should mention "not found"
Scenario: rollback_to() returns not-supported for remote database
Given a remote postgres database resource with location "postgresql://localhost/mydb"
And a mock sandbox manager
When I rollback to checkpoint "some-ckpt" on the database handler
Then the db rollback result should not be successful
And the db rollback result message should mention "not supported"
# ============================================================
# Error handling
# ============================================================
Scenario: All methods handle missing location gracefully
Given a database resource with no location
And a mock sandbox manager
When I read from the database handler
And I write SQL "SELECT 1" to the database handler
And I delete with empty path from the database handler
And I list children from the database handler
Then no db exceptions should have been raised
features/steps/database_handler_crud_steps.py
+575
View File
@@ -0,0 +1,575 @@
"""Step definitions for database_handler_crud.feature.
Tests CRUD and checkpoint operations for DatabaseResourceHandler:
- read(), write(), delete(), list_children(), diff()
- create_checkpoint(), rollback_to()
Issue #1241: DatabaseResourceHandler CRUD and checkpoint methods.
"""
from __future__ import annotations
import os
import sqlite3
import tempfile
from unittest.mock import MagicMock
from behave import given, then, when
from behave.runner import Context
from cleveragents.domain.models.core.resource import (
PhysVirt,
Resource,
ResourceCapabilities,
)
from cleveragents.resource.handlers.database import DatabaseResourceHandler
__all__: list[str] = []
# ---------------------------------------------------------------------------
# ULID generation helper
# ---------------------------------------------------------------------------
_ULID_COUNTER = 0
def _next_ulid() -> str:
"""Generate a valid 26-char Crockford Base32 ID for tests."""
global _ULID_COUNTER
_ULID_COUNTER += 1
cb32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
n = _ULID_COUNTER
chars: list[str] = []
for _ in range(26):
chars.append(cb32[n % 32])
n //= 32
return "".join(reversed(chars))
def _make_db_resource(
resource_type: str,
location: str | None,
rid: str | None = None,
) -> Resource:
"""Create a database Resource for testing."""
return Resource(
resource_id=rid or _next_ulid(),
resource_type_name=resource_type,
classification=PhysVirt.PHYSICAL,
location=location,
capabilities=ResourceCapabilities(
readable=True,
writable=True,
sandboxable=True,
checkpointable=True,
),
)
def _create_sqlite_with_table(table_name: str) -> str:
"""Create a temp SQLite file with a single table, return path."""
fd, path = tempfile.mkstemp(suffix=".db", prefix="dbhandler_")
os.close(fd)
conn = sqlite3.connect(path)
conn.execute(f"CREATE TABLE {table_name} (id INTEGER PRIMARY KEY, name TEXT)")
conn.commit()
conn.close()
return path
def _create_sqlite_with_tables(table1: str, table2: str) -> str:
"""Create a temp SQLite file with two tables, return path."""
fd, path = tempfile.mkstemp(suffix=".db", prefix="dbhandler_")
os.close(fd)
conn = sqlite3.connect(path)
conn.execute(f"CREATE TABLE {table1} (id INTEGER PRIMARY KEY, name TEXT)")
conn.execute(f"CREATE TABLE {table2} (id INTEGER PRIMARY KEY, value TEXT)")
conn.commit()
conn.close()
return path
def _create_empty_sqlite() -> str:
"""Create an empty temp SQLite file, return path."""
fd, path = tempfile.mkstemp(suffix=".db", prefix="dbhandler_")
os.close(fd)
conn = sqlite3.connect(path)
conn.close()
return path
# ---------------------------------------------------------------------------
# Given: database setup
# ---------------------------------------------------------------------------
@given('a SQLite database with tables "{table1}" and "{table2}"')
def step_given_sqlite_two_tables(context: Context, table1: str, table2: str) -> None:
context.db_path = _create_sqlite_with_tables(table1, table2)
context.db_handler = DatabaseResourceHandler()
context.db_resource = _make_db_resource("sqlite", context.db_path)
context.db_exceptions_raised: list[Exception] = []
@given('a SQLite database with table "{table_name}"')
def step_given_sqlite_one_table(context: Context, table_name: str) -> None:
context.db_path = _create_sqlite_with_table(table_name)
context.db_handler = DatabaseResourceHandler()
context.db_resource = _make_db_resource("sqlite", context.db_path)
context.db_exceptions_raised = []
@given("an empty SQLite database file")
def step_given_empty_sqlite(context: Context) -> None:
context.db_path = _create_empty_sqlite()
context.db_handler = DatabaseResourceHandler()
context.db_resource = _make_db_resource("sqlite", context.db_path)
context.db_exceptions_raised = []
@given('a remote postgres database resource with location "{location}"')
def step_given_remote_postgres(context: Context, location: str) -> None:
context.db_handler = DatabaseResourceHandler()
context.db_resource = _make_db_resource("postgres", location)
context.db_exceptions_raised = []
@given('a remote mysql database resource with location "{location}"')
def step_given_remote_mysql(context: Context, location: str) -> None:
context.db_handler = DatabaseResourceHandler()
context.db_resource = _make_db_resource("mysql", location)
context.db_exceptions_raised = []
@given("a database resource with no location")
def step_given_no_location(context: Context) -> None:
context.db_handler = DatabaseResourceHandler()
context.db_resource = _make_db_resource("sqlite", None)
context.db_exceptions_raised = []
@given("a database resource pointing to a non-existent SQLite path")
def step_given_nonexistent_sqlite(context: Context) -> None:
context.db_handler = DatabaseResourceHandler()
context.db_resource = _make_db_resource("sqlite", "/tmp/nonexistent_db_12345.db")
context.db_exceptions_raised = []
@given("a mock sandbox manager")
def step_given_mock_sandbox_manager(context: Context) -> None:
context.mock_sandbox_manager = MagicMock()
@given("a second SQLite database with the same schema")
def step_given_second_sqlite_same_schema(context: Context) -> None:
conn_src = sqlite3.connect(context.db_path)
schema_rows = conn_src.execute(
"SELECT name, sql FROM sqlite_master WHERE type='table'"
).fetchall()
conn_src.close()
fd, path2 = tempfile.mkstemp(suffix=".db", prefix="dbhandler2_")
os.close(fd)
conn2 = sqlite3.connect(path2)
for _, sql in schema_rows:
if sql:
conn2.execute(sql)
conn2.commit()
conn2.close()
context.second_db_path = path2
@given('a second SQLite database with table "{table_name}"')
def step_given_second_sqlite_different(context: Context, table_name: str) -> None:
context.second_db_path = _create_sqlite_with_table(table_name)
@given('a second remote database location "{location}"')
def step_given_second_remote_location(context: Context, location: str) -> None:
context.second_db_path = location
# ---------------------------------------------------------------------------
# When: CRUD operations
# ---------------------------------------------------------------------------
@when("I read from the database handler")
def step_when_read_db(context: Context) -> None:
try:
context.db_read_result = context.db_handler.read(resource=context.db_resource)
except Exception as exc:
context.db_exceptions_raised.append(exc)
context.db_read_result = None
@when('I write SQL "{sql}" to the database handler')
def step_when_write_sql(context: Context, sql: str) -> None:
try:
context.db_write_result = context.db_handler.write(
resource=context.db_resource,
path="",
data=sql.encode("utf-8"),
)
except Exception as exc:
context.db_exceptions_raised.append(exc)
context.db_write_result = None
@when('I delete table "{table_name}" from the database handler')
def step_when_delete_table(context: Context, table_name: str) -> None:
try:
context.db_delete_result = context.db_handler.delete(
resource=context.db_resource,
path=table_name,
)
except Exception as exc:
context.db_exceptions_raised.append(exc)
context.db_delete_result = None
@when("I delete with empty path from the database handler")
def step_when_delete_empty(context: Context) -> None:
try:
context.db_delete_result = context.db_handler.delete(
resource=context.db_resource,
path="",
)
except Exception as exc:
context.db_exceptions_raised.append(exc)
context.db_delete_result = None
@when("I list children from the database handler")
def step_when_list_children(context: Context) -> None:
try:
context.db_children_result = context.db_handler.list_children(
resource=context.db_resource
)
except Exception as exc:
context.db_exceptions_raised.append(exc)
context.db_children_result = []
@when("I diff the database handler against the second database")
def step_when_diff_second_db(context: Context) -> None:
try:
context.db_diff_result = context.db_handler.diff(
resource=context.db_resource,
other_location=context.second_db_path,
)
except Exception as exc:
context.db_exceptions_raised.append(exc)
context.db_diff_result = None
@when("I diff the database handler against the second remote location")
def step_when_diff_remote(context: Context) -> None:
try:
context.db_diff_result = context.db_handler.diff(
resource=context.db_resource,
other_location=context.second_db_path,
)
except Exception as exc:
context.db_exceptions_raised.append(exc)
context.db_diff_result = None
@when('I create a checkpoint with plan "{plan_id}" on the database handler')
def step_when_create_checkpoint(context: Context, plan_id: str) -> None:
try:
context.db_checkpoint_result = context.db_handler.create_checkpoint(
resource=context.db_resource,
plan_id=plan_id,
sandbox_manager=context.mock_sandbox_manager,
)
context.db_last_checkpoint_id = context.db_checkpoint_result.checkpoint_id
except Exception as exc:
context.db_exceptions_raised.append(exc)
context.db_checkpoint_result = None
context.db_last_checkpoint_id = None
@when('I insert a row into "{table_name}" via the database handler')
def step_when_insert_row(context: Context, table_name: str) -> None:
sql = f"INSERT INTO {table_name} (id, name) VALUES (42, 'test-row')"
context.db_handler.write(
resource=context.db_resource,
path="",
data=sql.encode("utf-8"),
)
@when("I rollback to the last checkpoint on the database handler")
def step_when_rollback_last(context: Context) -> None:
try:
context.db_rollback_result = context.db_handler.rollback_to(
resource=context.db_resource,
plan_id="PLAN-RB-001",
checkpoint_id=context.db_last_checkpoint_id,
sandbox_manager=context.mock_sandbox_manager,
)
except Exception as exc:
context.db_exceptions_raised.append(exc)
context.db_rollback_result = None
@when('I rollback to checkpoint "{checkpoint_id}" on the database handler')
def step_when_rollback_id(context: Context, checkpoint_id: str) -> None:
try:
context.db_rollback_result = context.db_handler.rollback_to(
resource=context.db_resource,
plan_id="PLAN-TEST",
checkpoint_id=checkpoint_id,
sandbox_manager=context.mock_sandbox_manager,
)
except Exception as exc:
context.db_exceptions_raised.append(exc)
context.db_rollback_result = None
# ---------------------------------------------------------------------------
# Then: read() assertions
# ---------------------------------------------------------------------------
@then('the db read content should contain "{text}"')
def step_then_read_contains(context: Context, text: str) -> None:
assert context.db_read_result is not None, "read() raised an exception"
content_text = context.db_read_result.data.decode(
context.db_read_result.encoding or "utf-8"
)
assert text in content_text, (
f"Expected '{text}' in read content, got: {content_text[:200]}"
)
@then('the db read content encoding should be "{encoding}"')
def step_then_read_encoding(context: Context, encoding: str) -> None:
assert context.db_read_result is not None, "read() raised an exception"
assert context.db_read_result.encoding == encoding, (
f"Expected encoding '{encoding}', got '{context.db_read_result.encoding}'"
)
@then("the db read content hash should not be empty")
def step_then_read_hash_not_empty(context: Context) -> None:
assert context.db_read_result is not None, "read() raised an exception"
assert context.db_read_result.content_hash is not None, (
"Expected content_hash to be set"
)
assert len(context.db_read_result.content_hash) > 0
@then("the db read content data should be empty or whitespace")
def step_then_read_data_empty(context: Context) -> None:
assert context.db_read_result is not None, "read() raised an exception"
text = context.db_read_result.data.decode(
context.db_read_result.encoding or "utf-8"
)
assert text.strip() == "", f"Expected empty/whitespace content, got: {text[:200]}"
@then("no db exceptions should have been raised")
def step_then_no_exceptions(context: Context) -> None:
assert len(context.db_exceptions_raised) == 0, (
f"Unexpected exceptions: {context.db_exceptions_raised}"
)
# ---------------------------------------------------------------------------
# Then: write() assertions
# ---------------------------------------------------------------------------
@then("the db write result should be successful")
def step_then_write_success(context: Context) -> None:
assert context.db_write_result is not None, "write() raised an exception"
assert context.db_write_result.success is True, (
f"Expected write success, got message: {context.db_write_result.message}"
)
@then("the db write result should not be successful")
def step_then_write_not_success(context: Context) -> None:
assert context.db_write_result is not None, "write() raised an exception"
assert context.db_write_result.success is False, (
"Expected write failure, but got success"
)
@then('the db write result message should mention "{text}"')
def step_then_write_message_contains(context: Context, text: str) -> None:
assert context.db_write_result is not None, "write() raised an exception"
assert text.lower() in context.db_write_result.message.lower(), (
f"Expected '{text}' in write message: {context.db_write_result.message}"
)
@then('the SQLite table "{table_name}" should contain {count:d} row')
def step_then_sqlite_table_row_count(
context: Context, table_name: str, count: int
) -> None:
conn = sqlite3.connect(context.db_path)
rows = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()
conn.close()
assert rows is not None
assert rows[0] == count, f"Expected {count} rows, got {rows[0]}"
@then('the SQLite database should have table "{table_name}"')
def step_then_sqlite_has_table(context: Context, table_name: str) -> None:
conn = sqlite3.connect(context.db_path)
rows = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name=?",
(table_name,),
).fetchall()
conn.close()
assert len(rows) == 1, f"Table '{table_name}' not found in database"
@then('the SQLite database should not have table "{table_name}"')
def step_then_sqlite_no_table(context: Context, table_name: str) -> None:
conn = sqlite3.connect(context.db_path)
rows = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name=?",
(table_name,),
).fetchall()
conn.close()
assert len(rows) == 0, f"Table '{table_name}' still exists in database"
# ---------------------------------------------------------------------------
# Then: delete() assertions
# ---------------------------------------------------------------------------
@then("the db delete result should be successful")
def step_then_delete_success(context: Context) -> None:
assert context.db_delete_result is not None, "delete() raised an exception"
assert context.db_delete_result.success is True, (
f"Expected delete success, got message: {context.db_delete_result.message}"
)
@then("the db delete result should not be successful")
def step_then_delete_not_success(context: Context) -> None:
assert context.db_delete_result is not None, "delete() raised an exception"
assert context.db_delete_result.success is False, (
"Expected delete failure, but got success"
)
@then('the db delete result message should mention "{text}"')
def step_then_delete_message_contains(context: Context, text: str) -> None:
assert context.db_delete_result is not None, "delete() raised an exception"
assert text.lower() in context.db_delete_result.message.lower(), (
f"Expected '{text}' in delete message: {context.db_delete_result.message}"
)
# ---------------------------------------------------------------------------
# Then: list_children() assertions
# ---------------------------------------------------------------------------
@then('the db children list should contain "{name}"')
def step_then_children_contains(context: Context, name: str) -> None:
assert name in context.db_children_result, (
f"Expected '{name}' in children: {context.db_children_result}"
)
@then("the db children list should be sorted")
def step_then_children_sorted(context: Context) -> None:
assert context.db_children_result == sorted(context.db_children_result), (
f"Children list is not sorted: {context.db_children_result}"
)
@then("the db children list should be empty")
def step_then_children_empty(context: Context) -> None:
assert context.db_children_result == [], (
f"Expected empty children list, got: {context.db_children_result}"
)
# ---------------------------------------------------------------------------
# Then: diff() assertions
# ---------------------------------------------------------------------------
@then("the db diff result should have changes")
def step_then_diff_has_changes(context: Context) -> None:
assert context.db_diff_result is not None, "diff() raised an exception"
assert context.db_diff_result.has_changes is True, (
"Expected diff to have changes, but has_changes is False"
)
@then("the db diff result should have no changes")
def step_then_diff_no_changes(context: Context) -> None:
assert context.db_diff_result is not None, "diff() raised an exception"
assert context.db_diff_result.has_changes is False, (
"Expected no diff changes, but has_changes is True"
)
# ---------------------------------------------------------------------------
# Then: create_checkpoint() assertions
# ---------------------------------------------------------------------------
@then("the db checkpoint result should have a checkpoint_id")
def step_then_checkpoint_has_id(context: Context) -> None:
assert context.db_checkpoint_result is not None, (
"create_checkpoint() raised an exception"
)
assert context.db_checkpoint_result.checkpoint_id, (
"Expected non-empty checkpoint_id"
)
@then('the db checkpoint result plan_id should be "{plan_id}"')
def step_then_checkpoint_plan_id(context: Context, plan_id: str) -> None:
assert context.db_checkpoint_result is not None
assert context.db_checkpoint_result.plan_id == plan_id, (
f"Expected plan_id '{plan_id}', got '{context.db_checkpoint_result.plan_id}'"
)
@then('the db checkpoint result message should mention "{text}"')
def step_then_checkpoint_message_contains(context: Context, text: str) -> None:
assert context.db_checkpoint_result is not None
assert text.lower() in context.db_checkpoint_result.message.lower(), (
f"Expected '{text}' in checkpoint message: {context.db_checkpoint_result.message}"
)
# ---------------------------------------------------------------------------
# Then: rollback_to() assertions
# ---------------------------------------------------------------------------
@then("the db rollback result should be successful")
def step_then_rollback_success(context: Context) -> None:
assert context.db_rollback_result is not None, "rollback_to() raised an exception"
assert context.db_rollback_result.success is True, (
f"Expected rollback success, got message: {context.db_rollback_result.message}"
)
@then("the db rollback result should not be successful")
def step_then_rollback_not_success(context: Context) -> None:
assert context.db_rollback_result is not None, "rollback_to() raised an exception"
assert context.db_rollback_result.success is False, (
"Expected rollback failure, but got success"
)
@then('the db rollback result message should mention "{text}"')
def step_then_rollback_message_contains(context: Context, text: str) -> None:
assert context.db_rollback_result is not None
assert text.lower() in context.db_rollback_result.message.lower(), (
f"Expected '{text}' in rollback message: {context.db_rollback_result.message}"
)
src/cleveragents/resource/handlers/database.py
+539
View File
@@ -13,9 +13,23 @@ Connection validation tests connectivity and returns safe error
messages that mask credentials using the shared
:mod:`cleveragents.shared.redaction` module.
Content CRUD operations (issue #827, #1241):
- ``read`` — SQLite: query ``sqlite_master`` schema; remote: connection-info summary
- ``write`` — SQLite: execute SQL statement; remote: not-supported result
- ``delete`` — SQLite: ``DROP TABLE IF EXISTS``; remote: not-supported result
- ``list_children`` — SQLite: list tables/views from ``sqlite_master``
- ``diff`` — compare schemas via content hash
Checkpoint methods (issue #836, #1241):
- ``create_checkpoint`` — SQLite: ``SAVEPOINT``; remote: content hash fallback
- ``rollback_to`` — SQLite: ``ROLLBACK TO SAVEPOINT``; remote: not-supported
Based on:
- implementation_plan.md group M7.post-resource-db
- Built-in type definitions for database resources
- Issue #1241 — DatabaseResourceHandler CRUD and checkpoint methods
"""
from __future__ import annotations
@@ -27,7 +41,16 @@ import sqlite3
from typing import Any
from cleveragents.domain.models.core.resource import Resource, SandboxStrategy
from cleveragents.infrastructure.sandbox.manager import SandboxManager
from cleveragents.resource.handlers._base import EMPTY_CONTENT_HASH, BaseResourceHandler
from cleveragents.resource.handlers.protocol import (
CheckpointResult,
Content,
DeleteResult,
DiffResult,
RollbackResult,
WriteResult,
)
from cleveragents.shared.redaction import mask_database_url, redact_dict
logger = logging.getLogger(__name__)
@@ -400,6 +423,30 @@ def _validate_network_db(
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _is_sqlite_resource(resource: Resource) -> bool:
"""Return True if the resource is a local SQLite database file."""
if not resource.location:
return False
loc = resource.location
# In-memory SQLite
if loc == ":memory:":
return True
# File-based SQLite (by extension or by resource type name)
if resource.resource_type_name == "sqlite":
return True
return loc.endswith((".db", ".sqlite", ".sqlite3"))
def _open_sqlite(location: str) -> sqlite3.Connection:
"""Open an SQLite connection to *location* (read-write)."""
return sqlite3.connect(location)
# ---------------------------------------------------------------------------
# DatabaseResourceHandler
# ---------------------------------------------------------------------------
@@ -410,11 +457,503 @@ class DatabaseResourceHandler(BaseResourceHandler):
Provisions a transaction-rollback sandbox for database resources.
Supports ``postgres``, ``mysql``, ``sqlite``, and ``duckdb``.
Content CRUD (issue #1241):
- ``read`` — SQLite: query ``sqlite_master`` schema; remote: connection-info
summary
- ``write`` — SQLite: execute SQL statement; remote: not-supported result
- ``delete`` — SQLite: ``DROP TABLE IF EXISTS``; remote: not-supported result
- ``list_children`` — SQLite: list tables/views from ``sqlite_master``
- ``diff`` — compare schemas via content hash
Checkpoint methods (issue #1241):
- ``create_checkpoint`` — SQLite: ``SAVEPOINT``; remote: content hash fallback
- ``rollback_to`` — SQLite: ``ROLLBACK TO SAVEPOINT``; remote: not-supported
"""
_default_strategy = SandboxStrategy.TRANSACTION_ROLLBACK
_type_label = "database"
def __init__(self) -> None:
# Maps checkpoint_id -> (connection, savepoint_name) for SQLite SAVEPOINTs.
# Each checkpoint holds an open connection with an active savepoint.
self._sqlite_checkpoints: dict[str, tuple[sqlite3.Connection, str]] = {}
# -- Content CRUD (issue #1241) ----------------------------------------
def read(self, *, resource: Resource, path: str = "") -> Content:
"""Read content from a database resource.
For SQLite databases, queries ``sqlite_master`` and returns a
human-readable schema summary (table names and their DDL).
For remote databases (postgres, mysql, duckdb), returns a
connection-info summary using the resource location.
Args:
resource: The database resource.
path: Ignored for database resources (schema is always read).
Returns:
A :class:`Content` with the schema or connection-info bytes.
"""
if not resource.location:
return Content(
data=b"",
encoding="utf-8",
metadata={"source": "empty"},
)
if _is_sqlite_resource(resource):
return self._read_sqlite(resource)
# Remote database: return connection-info summary
return self._read_remote(resource)
def _read_sqlite(self, resource: Resource) -> Content:
"""Read SQLite schema from sqlite_master."""
location = resource.location or ":memory:"
try:
conn = _open_sqlite(location)
try:
cursor = conn.execute(
"SELECT type, name, sql FROM sqlite_master "
"WHERE type IN ('table', 'view', 'index') "
"ORDER BY type, name"
)
rows = cursor.fetchall()
lines: list[str] = []
for obj_type, name, sql in rows:
lines.append(f"-- {obj_type}: {name}")
if sql:
lines.append(sql)
lines.append("")
schema_text = "\n".join(lines)
data = schema_text.encode("utf-8")
content_hash = hashlib.sha256(data).hexdigest()
return Content(
data=data,
encoding="utf-8",
content_hash=content_hash,
metadata={
"source": "sqlite_master",
"location": location,
"object_count": str(len(rows)),
},
)
finally:
conn.close()
except sqlite3.Error as exc:
logger.warning("SQLite read failed for %s: %s", location, exc)
error_msg = f"SQLite read error: {exc}"
return Content(
data=error_msg.encode("utf-8"),
encoding="utf-8",
metadata={"error": str(exc), "location": location},
)
def _read_remote(self, resource: Resource) -> Content:
"""Return a connection-info summary for remote databases."""
location = resource.location or ""
resource_type = resource.resource_type_name
# Mask any credentials in the location string
safe_location = mask_database_url(location) if "://" in location else location
summary = (
f"Database resource: {resource_type}\n"
f"Location: {safe_location}\n"
f"Resource ID: {resource.resource_id}\n"
)
data = summary.encode("utf-8")
return Content(
data=data,
encoding="utf-8",
metadata={
"source": "connection_info",
"resource_type": resource_type,
},
)
def write(self, *, resource: Resource, path: str, data: bytes) -> WriteResult:
"""Write to a database resource.
For SQLite databases, executes the SQL statement provided in
*data* (decoded as UTF-8). The *path* parameter is ignored.
For remote databases, returns a not-supported result.
Args:
resource: The database resource.
path: Ignored for database resources.
data: SQL statement bytes to execute (SQLite only).
Returns:
A :class:`WriteResult` indicating success or not-supported.
"""
if not resource.location:
return WriteResult(
success=False,
message="Database resource has no location",
)
if _is_sqlite_resource(resource):
return self._write_sqlite(resource, data)
# Remote database: not supported
return WriteResult(
success=False,
bytes_written=0,
message=(
f"write() is not supported for remote database type "
f"'{resource.resource_type_name}'"
),
)
def _write_sqlite(self, resource: Resource, data: bytes) -> WriteResult:
"""Execute a SQL statement on an SQLite database."""
location = resource.location or ":memory:"
try:
sql = data.decode("utf-8")
except UnicodeDecodeError as exc:
return WriteResult(
success=False,
message=f"SQL data is not valid UTF-8: {exc}",
)
try:
conn = _open_sqlite(location)
try:
conn.execute(sql)
conn.commit()
return WriteResult(
success=True,
bytes_written=len(data),
message=f"Executed SQL on SQLite database: {location}",
)
finally:
conn.close()
except sqlite3.Error as exc:
logger.warning("SQLite write failed for %s: %s", location, exc)
return WriteResult(
success=False,
message=f"SQLite execute error: {exc}",
)
def delete(self, *, resource: Resource, path: str = "") -> DeleteResult:
"""Delete from a database resource.
For SQLite databases, executes ``DROP TABLE IF EXISTS <path>``
where *path* is the table name. If *path* is empty, returns
a not-supported result (cannot drop the entire database file
via this method).
For remote databases, returns a not-supported result.
Args:
resource: The database resource.
path: Table name to drop (SQLite only).
Returns:
A :class:`DeleteResult` indicating success or not-supported.
"""
if not resource.location:
return DeleteResult(
success=False,
message="Database resource has no location",
)
if _is_sqlite_resource(resource):
return self._delete_sqlite(resource, path)
# Remote database: not supported
return DeleteResult(
success=False,
message=(
f"delete() is not supported for remote database type "
f"'{resource.resource_type_name}'"
),
)
def _delete_sqlite(self, resource: Resource, path: str) -> DeleteResult:
"""Drop a table from an SQLite database."""
location = resource.location or ":memory:"
if not path:
return DeleteResult(
success=False,
message=(
"Cannot delete the entire SQLite database via delete(); "
"provide a table name as path"
),
)
# Validate table name: only allow simple identifiers (no SQL injection)
# Use parameterised quoting via sqlite3.
try:
conn = _open_sqlite(location)
try:
# Use double-quote escaping for the identifier
safe_name = path.replace('"', '""')
conn.execute(f'DROP TABLE IF EXISTS "{safe_name}"')
conn.commit()
return DeleteResult(
success=True,
message=f"Dropped table '{path}' from SQLite database: {location}",
)
finally:
conn.close()
except sqlite3.Error as exc:
logger.warning("SQLite delete failed for %s: %s", location, exc)
return DeleteResult(
success=False,
message=f"SQLite DROP TABLE error: {exc}",
)
def list_children(self, *, resource: Resource) -> list[str]:
"""List tables and views in a database resource.
For SQLite databases, queries ``sqlite_master`` and returns
sorted table and view names.
For remote databases, returns an empty list.
Args:
resource: The database resource.
Returns:
Sorted list of table/view names (SQLite) or empty list.
"""
if not resource.location:
return []
if _is_sqlite_resource(resource):
return self._list_children_sqlite(resource)
# Remote database: cannot list without a live connection
return []
def _list_children_sqlite(self, resource: Resource) -> list[str]:
"""List tables and views from sqlite_master."""
location = resource.location or ":memory:"
try:
conn = _open_sqlite(location)
try:
cursor = conn.execute(
"SELECT name FROM sqlite_master "
"WHERE type IN ('table', 'view') "
"ORDER BY name"
)
return [row[0] for row in cursor.fetchall()]
finally:
conn.close()
except sqlite3.Error as exc:
logger.warning("SQLite list_children failed for %s: %s", location, exc)
return []
def diff(self, *, resource: Resource, other_location: str) -> DiffResult:
"""Compare a database resource against another location.
Computes the content hash of both the resource and the
*other_location* (treated as another database path) and
returns a :class:`DiffResult` indicating whether they differ.
For SQLite databases, the hash is based on the schema from
``sqlite_master``. For remote databases, the identity hash
is used.
Args:
resource: The database resource.
other_location: Path or identifier of the other database.
Returns:
A :class:`DiffResult` summarising the differences.
"""
hash_a = self.content_hash(resource)
# Build a temporary resource for the other location
other_resource = Resource(
resource_id=resource.resource_id,
name=resource.name,
resource_type_name=resource.resource_type_name,
classification=resource.classification,
description=resource.description,
location=other_location,
parents=list(resource.parents),
)
hash_b = self.content_hash(other_resource)
has_changes = hash_a != hash_b
return DiffResult(
has_changes=has_changes,
unified_diff=""
if not has_changes
else f"--- {resource.location}\n+++ {other_location}\n(schema differs)",
files_changed=1 if has_changes else 0,
insertions=0,
deletions=0,
)
# -- Checkpoint and rollback (issue #1241) -----------------------------
def create_checkpoint(
self,
*,
resource: Resource,
plan_id: str,
sandbox_manager: SandboxManager,
phase: str = "execution",
) -> CheckpointResult:
"""Create a checkpoint of the current database state.
For SQLite databases, creates a ``SAVEPOINT`` on an open
connection. The connection is kept open until
:meth:`rollback_to` or the checkpoint is discarded.
For remote databases, falls back to a content-hash-based
checkpoint (records the hash but cannot restore).
Args:
resource: The database resource.
plan_id: The plan requesting the checkpoint.
sandbox_manager: The sandbox lifecycle manager.
phase: Lifecycle phase label (e.g. ``"execution"``).
Returns:
A :class:`CheckpointResult` with the checkpoint ID.
"""
from datetime import UTC, datetime
timestamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%S")
checkpoint_id = f"dbckpt-{plan_id}-{timestamp}"
if resource.location and _is_sqlite_resource(resource):
return self._create_checkpoint_sqlite(resource, plan_id, checkpoint_id)
# Remote database: content-hash fallback
content_hash = self.content_hash(resource)
return CheckpointResult(
checkpoint_id=checkpoint_id,
plan_id=plan_id,
snapshot_path="",
message=(
f"Content-hash checkpoint for remote database "
f"'{resource.resource_type_name}': {content_hash[:16]}..."
),
)
def _create_checkpoint_sqlite(
self,
resource: Resource,
plan_id: str,
checkpoint_id: str,
) -> CheckpointResult:
"""Create a SQLite SAVEPOINT checkpoint."""
location = resource.location or ":memory:"
# Use a safe savepoint name derived from the checkpoint_id
savepoint_name = "sp_" + checkpoint_id.replace("-", "_")
try:
conn = _open_sqlite(location)
conn.execute(f"SAVEPOINT {savepoint_name}")
# Store the open connection so rollback_to can use it
self._sqlite_checkpoints[checkpoint_id] = (conn, savepoint_name)
return CheckpointResult(
checkpoint_id=checkpoint_id,
plan_id=plan_id,
snapshot_path=location,
message=(
f"Created SQLite SAVEPOINT '{savepoint_name}' "
f"on database: {location}"
),
)
except sqlite3.Error as exc:
logger.warning("SQLite SAVEPOINT failed for %s: %s", location, exc)
raise RuntimeError(f"Failed to create SQLite checkpoint: {exc}") from exc
def rollback_to(
self,
*,
resource: Resource,
plan_id: str,
checkpoint_id: str,
sandbox_manager: SandboxManager,
) -> RollbackResult:
"""Rollback a database resource to a prior checkpoint.
For SQLite databases, executes ``ROLLBACK TO SAVEPOINT`` on
the connection stored by :meth:`create_checkpoint`, then
releases the savepoint and closes the connection.
For remote databases, returns a not-supported result.
Args:
resource: The database resource.
plan_id: The plan requesting the rollback.
checkpoint_id: The checkpoint ID returned by
:meth:`create_checkpoint`.
sandbox_manager: The sandbox lifecycle manager.
Returns:
A :class:`RollbackResult` indicating success.
"""
if resource.location and _is_sqlite_resource(resource):
return self._rollback_sqlite(checkpoint_id)
# Remote database: not supported
return RollbackResult(
success=False,
checkpoint_id=checkpoint_id,
message=(
f"rollback_to() is not supported for remote database type "
f"'{resource.resource_type_name}'"
),
)
def _rollback_sqlite(self, checkpoint_id: str) -> RollbackResult:
"""Execute ROLLBACK TO SAVEPOINT for a SQLite checkpoint."""
entry = self._sqlite_checkpoints.get(checkpoint_id)
if entry is None:
return RollbackResult(
success=False,
checkpoint_id=checkpoint_id,
message=f"Checkpoint '{checkpoint_id}' not found or already released",
)
conn, savepoint_name = entry
try:
conn.execute(f"ROLLBACK TO SAVEPOINT {savepoint_name}")
conn.execute(f"RELEASE SAVEPOINT {savepoint_name}")
conn.commit()
conn.close()
del self._sqlite_checkpoints[checkpoint_id]
return RollbackResult(
success=True,
checkpoint_id=checkpoint_id,
restored_files=0,
message=(
f"Rolled back SQLite database to SAVEPOINT '{savepoint_name}'"
),
)
except sqlite3.Error as exc:
logger.warning(
"SQLite ROLLBACK TO SAVEPOINT failed for %s: %s",
checkpoint_id,
exc,
)
# Attempt cleanup
import contextlib
with contextlib.suppress(Exception):
conn.close()
self._sqlite_checkpoints.pop(checkpoint_id, None)
return RollbackResult(
success=False,
checkpoint_id=checkpoint_id,
message=f"SQLite rollback error: {exc}",
)
# -- Content Hashing ---------------------------------------------------
def content_hash(
self,
resource: Resource,