freemo
7db698b602
- Rewrite TUI session export/import step definitions to use constructor-
based dependency injection (container_factory) instead of
unittest.mock.patch context managers that fail across fork() boundaries
in the parallel test runner.
- Add container_factory parameter to TuiCommandRouter dataclass so tests
can inject a mock container that survives multiprocessing.fork().
- Add use_step_matcher('re') to a2a_jsonrpc_wire_format_steps.py so
regex-based step patterns are matched correctly (fixes 30 errored
scenarios with 56 undefined steps).
- Add plural 'rows' variant to database_handler_crud_steps.py row-count
step matcher (fixes 1 errored scenario).
ISSUES CLOSED: #2597
577 lines
21 KiB
Python
577 lines
21 KiB
Python
"""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} rows')
|
|
@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}"
|
|
)
|