CoreRasurae
798db9088e
CI / lint (pull_request) Successful in 15s
CI / typecheck (pull_request) Successful in 36s
CI / security (pull_request) Successful in 25s
CI / quality (pull_request) Successful in 19s
CI / build (pull_request) Successful in 23s
CI / integration_tests (pull_request) Successful in 6m49s
CI / coverage (pull_request) Successful in 12m1s
CI / unit_tests (pull_request) Successful in 16m28s
CI / docker (pull_request) Successful in 51s
123 lines
3.3 KiB
Python
123 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Helper script for Robot Framework tool registry smoke tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from cleveragents.infrastructure.database.models import Base
|
|
from cleveragents.infrastructure.database.repositories import (
|
|
DuplicateToolError,
|
|
ToolRegistryRepository,
|
|
ValidationAttachmentRepository,
|
|
)
|
|
|
|
|
|
def _setup():
|
|
"""Create in-memory DB and return (session, tool_repo, attachment_repo)."""
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
Base.metadata.create_all(engine)
|
|
factory = sessionmaker(bind=engine, expire_on_commit=False)
|
|
session = factory()
|
|
tool_repo = ToolRegistryRepository(session_factory=lambda: session)
|
|
attachment_repo = ValidationAttachmentRepository(session_factory=lambda: session)
|
|
return session, tool_repo, attachment_repo
|
|
|
|
|
|
def _make_tool(name, source="builtin", tool_type="tool", mode=None):
|
|
now = datetime.now().isoformat()
|
|
d = {
|
|
"name": name,
|
|
"description": f"Test tool {name}",
|
|
"tool_type": tool_type,
|
|
"source": source,
|
|
"timeout": 300,
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
"resource_bindings": [],
|
|
}
|
|
if mode:
|
|
d["mode"] = mode
|
|
return d
|
|
|
|
|
|
def cmd_register_and_get():
|
|
session, repo, _ = _setup()
|
|
tool = _make_tool("smoke/read-file")
|
|
repo.create(tool)
|
|
session.commit()
|
|
got = repo.get_by_name("smoke/read-file")
|
|
assert got is not None
|
|
assert got["name"] == "smoke/read-file"
|
|
print("tool-registry-ok")
|
|
|
|
|
|
def cmd_list_filter():
|
|
session, repo, _ = _setup()
|
|
repo.create(_make_tool("core/tool-a"))
|
|
repo.create(_make_tool("local/tool-b", source="custom"))
|
|
session.commit()
|
|
core_tools = repo.list_all(namespace="core")
|
|
assert len(core_tools) == 1
|
|
all_tools = repo.list_all()
|
|
assert len(all_tools) == 2
|
|
print("tool-list-ok")
|
|
|
|
|
|
def cmd_attach_detach():
|
|
session, repo, att_repo = _setup()
|
|
repo.create(_make_tool("smoke/val-check", tool_type="validation", mode="required"))
|
|
session.commit()
|
|
att = att_repo.attach(
|
|
validation_name="smoke/val-check",
|
|
resource_id="res-001",
|
|
mode="required",
|
|
)
|
|
session.commit()
|
|
assert att["attachment_id"]
|
|
result = att_repo.detach(att["attachment_id"])
|
|
session.commit()
|
|
assert result is True
|
|
print("attach-detach-ok")
|
|
|
|
|
|
def cmd_duplicate_reject():
|
|
session, repo, _ = _setup()
|
|
repo.create(_make_tool("smoke/dup-test"))
|
|
session.commit()
|
|
try:
|
|
repo.create(_make_tool("smoke/dup-test"))
|
|
session.commit()
|
|
print("ERROR: should have raised")
|
|
sys.exit(1)
|
|
except DuplicateToolError:
|
|
print("duplicate-reject-ok")
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print("Usage: helper_tool_registry.py <command>", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
cmd = sys.argv[1]
|
|
commands = {
|
|
"register-and-get": cmd_register_and_get,
|
|
"list-filter": cmd_list_filter,
|
|
"attach-detach": cmd_attach_detach,
|
|
"duplicate-reject": cmd_duplicate_reject,
|
|
}
|
|
|
|
if cmd not in commands:
|
|
print(f"Unknown command: {cmd}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
commands[cmd]()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|