2b6ac00263
Ensure that when database_url is a relative path (like sqlite:///cleveragents.db), it resolves relative to CLEVERAGENTS_HOME instead of the current working directory. This fixes E2E test isolation where data from previous runs persisted across runs at CWD, causing UNIQUE constraint failures. Changes: - Added _resolve_database_urls model validator in Settings to rewrite relative SQLite paths to absolute paths under CLEVERAGENTS_HOME - Updated get_database_url() in container.py to use CLEVERAGENTS_HOME as base directory and create parent directories when CLEVERAGENTS_HOME is explicitly set - Updated _check_database() in system.py to walk up directory tree for writable ancestor check, handling cases where intermediate directories have not yet been created - Removed @tdd_expected_fail from TDD test (now passes genuinely) ISSUES CLOSED: #1024
216 lines
7.5 KiB
Python
216 lines
7.5 KiB
Python
"""Helper script for tdd_sqlite_url_cwd.robot integration tests.
|
|
|
|
Each subcommand exercises the real application path (no mocks) to
|
|
reproduce bug #1024. The helper reports the **real** outcome: it exits
|
|
0 and prints the sentinel when the database resolves inside
|
|
CLEVERAGENTS_HOME (bug is fixed), and exits 1 when it resolves to CWD
|
|
(bug still present). The ``tdd_expected_fail_listener`` on the Robot
|
|
side handles pass/fail inversion while the bug remains open.
|
|
|
|
Bug: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/1024
|
|
TDD: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/1034
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import NoReturn
|
|
|
|
# Ensure local source tree is importable.
|
|
_ROOT = Path(__file__).resolve().parents[1]
|
|
_SRC = str(_ROOT / "src")
|
|
if _SRC not in sys.path:
|
|
sys.path.insert(0, _SRC)
|
|
|
|
|
|
def _fail(message: str) -> NoReturn:
|
|
"""Print an error message to stderr and exit with code 1."""
|
|
print(message, file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def _extract_sqlite_path(url: str) -> Path | None:
|
|
"""Extract the file path from a SQLite URL.
|
|
|
|
Note: intentionally duplicated in features/steps/tdd_sqlite_url_cwd_steps.py
|
|
to keep the Robot and Behave test suites independently runnable
|
|
without shared test utility coupling.
|
|
"""
|
|
prefix = "sqlite:///"
|
|
if not url.startswith(prefix):
|
|
return None
|
|
raw_path = url[len(prefix) :]
|
|
if not raw_path:
|
|
return None
|
|
return Path(raw_path).resolve()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommands
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def check_db_url_resolution() -> None:
|
|
"""Verify that get_database_url() resolves inside CLEVERAGENTS_HOME.
|
|
|
|
Creates a temporary CLEVERAGENTS_HOME, removes any overriding env
|
|
vars for database URLs, resets singletons, and checks whether the
|
|
resolved database URL points inside the temporary home directory.
|
|
"""
|
|
from cleveragents.application.container import (
|
|
get_database_url,
|
|
reset_container,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
|
|
original_cwd = Path.cwd().resolve()
|
|
tmpdir = tempfile.mkdtemp(prefix="tdd_sqlite_url_1024_robot_")
|
|
|
|
# Save and remove any overriding env vars
|
|
saved_db_url = os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
|
|
saved_test_db_url = os.environ.pop("CLEVERAGENTS_TEST_DATABASE_URL", None)
|
|
|
|
try:
|
|
os.environ["CLEVERAGENTS_HOME"] = tmpdir
|
|
reset_container()
|
|
Settings._instance = None
|
|
|
|
resolved_url = get_database_url()
|
|
db_path = _extract_sqlite_path(resolved_url)
|
|
|
|
home = Path(tmpdir).resolve()
|
|
|
|
if db_path is None:
|
|
_fail(f"Could not extract file path from database URL: {resolved_url}")
|
|
|
|
if not str(db_path).startswith(str(home)):
|
|
_fail(
|
|
f"Database path {db_path} is NOT inside "
|
|
f"CLEVERAGENTS_HOME {home}.\n"
|
|
f"Full URL: {resolved_url}\n"
|
|
f"CWD was: {original_cwd}"
|
|
)
|
|
|
|
print("tdd-sqlite-url-cwd-resolution-ok")
|
|
|
|
finally:
|
|
# Restore env vars
|
|
if saved_db_url is not None:
|
|
os.environ["CLEVERAGENTS_DATABASE_URL"] = saved_db_url
|
|
else:
|
|
os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
|
|
if saved_test_db_url is not None:
|
|
os.environ["CLEVERAGENTS_TEST_DATABASE_URL"] = saved_test_db_url
|
|
else:
|
|
os.environ.pop("CLEVERAGENTS_TEST_DATABASE_URL", None)
|
|
os.environ.pop("CLEVERAGENTS_HOME", None)
|
|
reset_container()
|
|
Settings._instance = None
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
|
|
def check_cli_db_location() -> None:
|
|
"""Verify that a CLI command creates the DB inside CLEVERAGENTS_HOME.
|
|
|
|
Invokes ``session list`` via the Typer CLI runner and then checks
|
|
whether any NEW database files ended up in CWD rather than
|
|
CLEVERAGENTS_HOME. Records which files exist before the command
|
|
runs so pre-existing files from earlier test runs are not flagged.
|
|
"""
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.application.container import reset_container
|
|
from cleveragents.cli.commands.session import app as session_app
|
|
from cleveragents.config.settings import Settings
|
|
|
|
original_cwd = Path.cwd().resolve()
|
|
tmpdir = tempfile.mkdtemp(prefix="tdd_sqlite_url_1024_robot_cli_")
|
|
|
|
saved_db_url = os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
|
|
saved_test_db_url = os.environ.pop("CLEVERAGENTS_TEST_DATABASE_URL", None)
|
|
|
|
try:
|
|
os.environ["CLEVERAGENTS_HOME"] = tmpdir
|
|
reset_container()
|
|
Settings._instance = None
|
|
|
|
# Record which suspect files already exist before running the command,
|
|
# so we only flag files that are newly created by the CLI invocation.
|
|
suspect_files = [
|
|
original_cwd / "cleveragents.db",
|
|
original_cwd / "cleveragents_test.db",
|
|
original_cwd / ".cleveragents" / "db.sqlite",
|
|
]
|
|
pre_existing = {str(f) for f in suspect_files if f.exists()}
|
|
|
|
runner = CliRunner()
|
|
runner.invoke(session_app, ["list"])
|
|
|
|
home = Path(tmpdir).resolve()
|
|
|
|
# Check for NEW DB files in CWD that should not be there
|
|
newly_created = [
|
|
f for f in suspect_files if f.exists() and str(f) not in pre_existing
|
|
]
|
|
|
|
if newly_created:
|
|
_fail(
|
|
f"Database file(s) newly created in CWD instead of "
|
|
f"CLEVERAGENTS_HOME:\n"
|
|
f" New CWD files: {newly_created}\n"
|
|
f" CLEVERAGENTS_HOME: {home}"
|
|
)
|
|
|
|
# Verify the DB was actually created inside CLEVERAGENTS_HOME.
|
|
# If no DB file exists anywhere, the test isn't exercising the
|
|
# database path and the result would be vacuously true.
|
|
home_db_files = list(home.rglob("*.db")) + list(home.rglob("*.sqlite"))
|
|
if not home_db_files:
|
|
_fail(
|
|
f"No database file was created inside CLEVERAGENTS_HOME "
|
|
f"({home}). The CLI command did not trigger database "
|
|
f"creation, so the test result is inconclusive.\n"
|
|
f"CWD was: {original_cwd}"
|
|
)
|
|
|
|
print("tdd-sqlite-url-cwd-cli-ok")
|
|
|
|
finally:
|
|
if saved_db_url is not None:
|
|
os.environ["CLEVERAGENTS_DATABASE_URL"] = saved_db_url
|
|
else:
|
|
os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
|
|
if saved_test_db_url is not None:
|
|
os.environ["CLEVERAGENTS_TEST_DATABASE_URL"] = saved_test_db_url
|
|
else:
|
|
os.environ.pop("CLEVERAGENTS_TEST_DATABASE_URL", None)
|
|
os.environ.pop("CLEVERAGENTS_HOME", None)
|
|
reset_container()
|
|
Settings._instance = None
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dispatcher
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_COMMANDS: dict[str, Callable[[], None]] = {
|
|
"check-db-url-resolution": check_db_url_resolution,
|
|
"check-cli-db-location": check_cli_db_location,
|
|
}
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
|
|
print(
|
|
f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
cmd = _COMMANDS[sys.argv[1]]
|
|
cmd()
|