feat(cli): add project context commands
This commit is contained in:
@@ -0,0 +1,161 @@
|
||||
"""Helper script for project context CLI Robot Framework tests.
|
||||
|
||||
Usage: python robot/helper_project_context_cli.py <test-name>
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from cleveragents.domain.models.core.context_policy import (
|
||||
ContextView,
|
||||
ProjectContextPolicy,
|
||||
)
|
||||
from cleveragents.domain.models.core.project import (
|
||||
NamespacedProject,
|
||||
parse_namespaced_name,
|
||||
)
|
||||
from cleveragents.infrastructure.database.models import Base
|
||||
from cleveragents.infrastructure.database.repositories import (
|
||||
NamespacedProjectRepository,
|
||||
)
|
||||
|
||||
|
||||
class _NoClose:
|
||||
"""Session wrapper that no-ops close()."""
|
||||
|
||||
def __init__(self, s: object) -> None:
|
||||
object.__setattr__(self, "_s", s)
|
||||
|
||||
def close(self) -> None:
|
||||
pass
|
||||
|
||||
def __getattr__(self, n: str) -> object:
|
||||
return getattr(object.__getattribute__(self, "_s"), n)
|
||||
|
||||
|
||||
def _setup_db() -> tuple[NamespacedProjectRepository, Any]:
|
||||
"""Create in-memory SQLite DB, return (repo, session_factory)."""
|
||||
engine = create_engine("sqlite:///:memory:", echo=False)
|
||||
Base.metadata.create_all(engine)
|
||||
session = sessionmaker(bind=engine, expire_on_commit=False)()
|
||||
wrapper = _NoClose(session)
|
||||
|
||||
def factory() -> object:
|
||||
return wrapper
|
||||
|
||||
repo = NamespacedProjectRepository(session_factory=factory)
|
||||
return repo, factory
|
||||
|
||||
|
||||
def _create_project(
|
||||
repo: NamespacedProjectRepository,
|
||||
name: str,
|
||||
) -> None:
|
||||
parsed = parse_namespaced_name(name)
|
||||
proj = NamespacedProject(
|
||||
name=parsed.name,
|
||||
namespace=parsed.namespace,
|
||||
)
|
||||
repo.create(proj)
|
||||
|
||||
|
||||
def test_context_set_show() -> None:
|
||||
"""Set a context policy and then read it back."""
|
||||
from cleveragents.cli.commands.project_context import (
|
||||
_read_policy,
|
||||
_write_policy,
|
||||
)
|
||||
|
||||
repo, sf = _setup_db()
|
||||
_create_project(repo, "ctx-robot")
|
||||
|
||||
# Write a policy
|
||||
policy = ProjectContextPolicy(
|
||||
default_view=ContextView(
|
||||
include_resources=["db-*"],
|
||||
exclude_paths=["*.pyc"],
|
||||
max_file_size=1048576,
|
||||
),
|
||||
strategize_view=ContextView(
|
||||
include_resources=["cache-*"],
|
||||
),
|
||||
)
|
||||
_write_policy(sf, "local/ctx-robot", policy)
|
||||
|
||||
# Read it back
|
||||
loaded = _read_policy(sf, "local/ctx-robot")
|
||||
if loaded.default_view.include_resources != ["db-*"]:
|
||||
raise AssertionError(
|
||||
f"include_resources mismatch: {loaded.default_view.include_resources}"
|
||||
)
|
||||
if loaded.strategize_view is None:
|
||||
raise AssertionError("strategize_view is None")
|
||||
if loaded.strategize_view.include_resources != ["cache-*"]:
|
||||
raise AssertionError("strategize include_resources mismatch")
|
||||
|
||||
# Test inheritance
|
||||
resolved = loaded.resolve_view("execute")
|
||||
if resolved.include_resources != ["cache-*"]:
|
||||
raise AssertionError(
|
||||
f"execute should inherit from strategize, got {resolved.include_resources}"
|
||||
)
|
||||
|
||||
print("context-set-show-ok")
|
||||
|
||||
|
||||
def test_context_inspect_stub() -> None:
|
||||
"""Verify inspect raises NotImplementedError."""
|
||||
from cleveragents.cli.commands.project_context import (
|
||||
context_inspect,
|
||||
)
|
||||
|
||||
try:
|
||||
context_inspect(project="local/any")
|
||||
raise AssertionError("Should have raised")
|
||||
except NotImplementedError as exc:
|
||||
if "ACMS" not in str(exc):
|
||||
raise AssertionError(f"Missing ACMS message: {exc}") from exc
|
||||
|
||||
print("context-inspect-stub-ok")
|
||||
|
||||
|
||||
def test_context_simulate_stub() -> None:
|
||||
"""Verify simulate raises NotImplementedError."""
|
||||
from cleveragents.cli.commands.project_context import (
|
||||
context_simulate,
|
||||
)
|
||||
|
||||
try:
|
||||
context_simulate(project="local/any")
|
||||
raise AssertionError("Should have raised")
|
||||
except NotImplementedError as exc:
|
||||
if "ACMS" not in str(exc):
|
||||
raise AssertionError(f"Missing ACMS message: {exc}") from exc
|
||||
|
||||
print("context-simulate-stub-ok")
|
||||
|
||||
|
||||
TESTS: dict[str, Any] = {
|
||||
"set-show": test_context_set_show,
|
||||
"inspect-stub": test_context_inspect_stub,
|
||||
"simulate-stub": test_context_simulate_stub,
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2 or sys.argv[1] not in TESTS:
|
||||
print(
|
||||
f"Usage: {sys.argv[0]} <{'|'.join(TESTS)}>",
|
||||
file=sys.stderr,
|
||||
)
|
||||
sys.exit(2)
|
||||
try:
|
||||
TESTS[sys.argv[1]]()
|
||||
except Exception as exc:
|
||||
print(f"FAIL: {exc}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
Reference in New Issue
Block a user