hamza.khyari
b028c80cab
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 14s
CI / build (pull_request) Successful in 15s
CI / quality (pull_request) Successful in 18s
CI / security (pull_request) Successful in 33s
CI / typecheck (pull_request) Successful in 41s
CI / unit_tests (pull_request) Successful in 2m29s
CI / integration_tests (pull_request) Successful in 3m2s
CI / docker (pull_request) Successful in 39s
CI / coverage (pull_request) Successful in 4m28s
CI / benchmark-regression (pull_request) Successful in 29m5s
Add project-resource isolation to the ACMS context tier system via ScopedBackendView, ResourceScope, alias resolution, DAG expansion, and enforcement hooks. Core types (scoped_view.py): ResourceScope, ScopeViolationError, ScopedBackendView, ScopedBackendSet, create_scoped_backend_set. Scope resolution (scope_resolution.py): ResourceAliasResolver, validate_project_scope, validate_resource_scope, resolve_resource_scope with registry_lookup callback for DAG expansion. Tier integration (scoped_tiers.py): ScopedTierMixin with get_scoped, get_scoped_by_resource, validate_fragment_scope, store_with_scope_check, get_scoped_metrics mixed into ContextTierService. Backward compat: tiers.py re-exports ScopedBackendView. Tests: 74 Behave scenarios (235 steps), 8 Robot Framework cases, 7 ASV benchmark suites. Lint and typecheck clean. ISSUES CLOSED: #193
274 lines
9.6 KiB
Python
274 lines
9.6 KiB
Python
"""Robot Framework helper for scoped backend view smoke tests.
|
|
|
|
Provides a CLI-style interface for Robot to invoke scoped view creation,
|
|
resource scope resolution, backend proxying, and scope validation.
|
|
Exit code 0 = success, 1 = failure.
|
|
|
|
Usage:
|
|
python robot/helper_scoped_view.py <command>
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Ensure the src directory is on the import path.
|
|
_SRC = str(Path(__file__).resolve().parents[1] / "src")
|
|
if _SRC not in sys.path:
|
|
sys.path.insert(0, _SRC)
|
|
|
|
from cleveragents.domain.models.acms.scope_resolution import ( # noqa: E402
|
|
ResourceAliasResolver,
|
|
resolve_resource_scope,
|
|
validate_project_scope,
|
|
validate_resource_scope,
|
|
)
|
|
from cleveragents.domain.models.acms.scoped_view import ( # noqa: E402
|
|
ResourceScope,
|
|
ScopedBackendView,
|
|
ScopeViolationError,
|
|
create_scoped_backend_set,
|
|
)
|
|
from cleveragents.domain.models.acms.stubs import ( # noqa: E402
|
|
InMemoryGraphBackend,
|
|
InMemoryTextBackend,
|
|
InMemoryVectorBackend,
|
|
)
|
|
from cleveragents.domain.models.acms.tiers import ( # noqa: E402
|
|
ContextTier,
|
|
TieredFragment,
|
|
)
|
|
from cleveragents.domain.models.core.project import ( # noqa: E402
|
|
LinkedResource,
|
|
NamespacedProject,
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
"""Entry point called by Robot Framework ``Run Process``."""
|
|
if len(sys.argv) < 2:
|
|
print("Usage: helper_scoped_view.py <command>")
|
|
return 1
|
|
|
|
command: str = sys.argv[1]
|
|
|
|
if command == "resource-scope":
|
|
try:
|
|
scope = ResourceScope(
|
|
resource_ids=frozenset({"R1", "R2"}),
|
|
project_names=frozenset({"local/api"}),
|
|
)
|
|
assert scope.contains_resource("R1")
|
|
assert scope.contains_resource("R2")
|
|
assert not scope.contains_resource("R99")
|
|
assert scope.contains_project("local/api")
|
|
assert not scope.contains_project("local/frontend")
|
|
assert scope.temporal_scope == "current"
|
|
print("scoped-view-resource-scope-ok")
|
|
return 0
|
|
except Exception as exc:
|
|
print(f"scoped-view-resource-scope-fail: {exc}")
|
|
return 1
|
|
|
|
if command == "path-filtering":
|
|
try:
|
|
scope = ResourceScope(
|
|
resource_ids=frozenset({"R1"}),
|
|
project_names=frozenset({"p1"}),
|
|
include_paths=("src/**/*.py",),
|
|
exclude_paths=("**/test_*",),
|
|
)
|
|
assert scope.matches_path("src/main.py")
|
|
assert not scope.matches_path("src/test_main.py")
|
|
assert not scope.matches_path("docs/readme.md")
|
|
print("scoped-view-path-filtering-ok")
|
|
return 0
|
|
except Exception as exc:
|
|
print(f"scoped-view-path-filtering-fail: {exc}")
|
|
return 1
|
|
|
|
if command == "fragment-filtering":
|
|
try:
|
|
view = ScopedBackendView(
|
|
allowed_projects=frozenset({"proj-a"}),
|
|
resource_ids=frozenset({"R1"}),
|
|
)
|
|
frag_ok = TieredFragment(
|
|
fragment_id="f1",
|
|
content="c",
|
|
tier=ContextTier("hot"),
|
|
project_name="proj-a",
|
|
resource_id="R1",
|
|
)
|
|
frag_bad_proj = TieredFragment(
|
|
fragment_id="f2",
|
|
content="c",
|
|
tier=ContextTier("hot"),
|
|
project_name="proj-b",
|
|
resource_id="R1",
|
|
)
|
|
frag_bad_res = TieredFragment(
|
|
fragment_id="f3",
|
|
content="c",
|
|
tier=ContextTier("hot"),
|
|
project_name="proj-a",
|
|
resource_id="R99",
|
|
)
|
|
assert view.is_visible(frag_ok)
|
|
assert not view.is_visible(frag_bad_proj)
|
|
assert not view.is_visible(frag_bad_res)
|
|
print("scoped-view-fragment-filtering-ok")
|
|
return 0
|
|
except Exception as exc:
|
|
print(f"scoped-view-fragment-filtering-fail: {exc}")
|
|
return 1
|
|
|
|
if command == "backend-proxying":
|
|
try:
|
|
scope = ResourceScope(
|
|
resource_ids=frozenset({"R1"}),
|
|
project_names=frozenset({"local/api"}),
|
|
)
|
|
view = ScopedBackendView.from_resource_scope(scope)
|
|
text_results = view.search_text(InMemoryTextBackend(), "test")
|
|
assert text_results == []
|
|
vec_results = view.search_vector(InMemoryVectorBackend(), [0.1, 0.2, 0.3])
|
|
assert vec_results == []
|
|
graph_result = view.search_graph(
|
|
InMemoryGraphBackend(), "SELECT ?s WHERE { ?s a uko:X }"
|
|
)
|
|
assert graph_result.triples == []
|
|
print("scoped-view-backend-proxying-ok")
|
|
return 0
|
|
except Exception as exc:
|
|
print(f"scoped-view-backend-proxying-fail: {exc}")
|
|
return 1
|
|
|
|
if command == "backend-set":
|
|
try:
|
|
scope = ResourceScope(
|
|
resource_ids=frozenset({"R1"}),
|
|
project_names=frozenset({"local/api"}),
|
|
)
|
|
bset = create_scoped_backend_set(
|
|
scope,
|
|
text_backend=InMemoryTextBackend(),
|
|
vector_backend=InMemoryVectorBackend(),
|
|
graph_backend=InMemoryGraphBackend(),
|
|
)
|
|
assert bset.search_text("q") == []
|
|
assert bset.search_vector([0.1]) == []
|
|
assert bset.search_graph("SELECT ?s WHERE { ?s a uko:X }").triples == []
|
|
# None backend returns empty
|
|
bset_no_text = create_scoped_backend_set(scope)
|
|
assert bset_no_text.search_text("q") == []
|
|
print("scoped-view-backend-set-ok")
|
|
return 0
|
|
except Exception as exc:
|
|
print(f"scoped-view-backend-set-fail: {exc}")
|
|
return 1
|
|
|
|
if command == "scope-validation":
|
|
try:
|
|
# validate_project_scope: valid
|
|
validate_project_scope(
|
|
frozenset({"proj-a"}), frozenset({"proj-a", "proj-b"})
|
|
)
|
|
# validate_project_scope: invalid
|
|
raised = False
|
|
try:
|
|
validate_project_scope(
|
|
frozenset({"proj-c"}), frozenset({"proj-a", "proj-b"})
|
|
)
|
|
except ScopeViolationError:
|
|
raised = True
|
|
assert raised, "Expected ScopeViolationError for out-of-scope project"
|
|
|
|
# validate_resource_scope: valid
|
|
scope = ResourceScope(
|
|
resource_ids=frozenset({"R1", "R2"}),
|
|
project_names=frozenset({"p1"}),
|
|
)
|
|
validate_resource_scope(frozenset({"R1"}), scope)
|
|
# validate_resource_scope: invalid
|
|
raised = False
|
|
try:
|
|
validate_resource_scope(frozenset({"R99"}), scope)
|
|
except ScopeViolationError:
|
|
raised = True
|
|
assert raised, "Expected ScopeViolationError for out-of-scope resource"
|
|
|
|
print("scoped-view-scope-validation-ok")
|
|
return 0
|
|
except Exception as exc:
|
|
print(f"scoped-view-scope-validation-fail: {exc}")
|
|
return 1
|
|
|
|
if command == "alias-resolver":
|
|
try:
|
|
ulid1 = "01HQ8ZDRX50000000000000001"
|
|
ulid2 = "01HQ8ZDRX50000000000000002"
|
|
proj = NamespacedProject(
|
|
name="myproj",
|
|
namespace="local",
|
|
linked_resources=[
|
|
LinkedResource(resource_id=ulid1, alias="repo"),
|
|
LinkedResource(resource_id=ulid2, alias="docs"),
|
|
],
|
|
)
|
|
resolver = ResourceAliasResolver()
|
|
# Resolve by alias
|
|
assert resolver.resolve(proj, "repo") == ulid1
|
|
# Resolve by ID
|
|
assert resolver.resolve(proj, ulid1) == ulid1
|
|
# Unknown returns None
|
|
assert resolver.resolve(proj, "nonexistent") is None
|
|
# resolve_many
|
|
resolved = resolver.resolve_many(proj, ("repo", "docs"))
|
|
assert ulid1 in resolved
|
|
assert ulid2 in resolved
|
|
# validate_uniqueness: no dupes
|
|
assert resolver.validate_uniqueness(proj) == []
|
|
print("scoped-view-alias-resolver-ok")
|
|
return 0
|
|
except Exception as exc:
|
|
print(f"scoped-view-alias-resolver-fail: {exc}")
|
|
return 1
|
|
|
|
if command == "resolve-scope":
|
|
try:
|
|
ulid1 = "01HQ8ZDRX50000000000000001"
|
|
ulid2 = "01HQ8ZDRX50000000000000002"
|
|
proj = NamespacedProject(
|
|
name="api",
|
|
namespace="local",
|
|
linked_resources=[
|
|
LinkedResource(resource_id=ulid1, alias="repo"),
|
|
LinkedResource(resource_id=ulid2, alias="docs"),
|
|
],
|
|
)
|
|
# Full scope
|
|
scope = resolve_resource_scope([proj])
|
|
assert len(scope.resource_ids) == 2
|
|
# Include filter
|
|
scope_inc = resolve_resource_scope([proj], include_resources=("repo",))
|
|
assert len(scope_inc.resource_ids) == 1
|
|
assert ulid1 in scope_inc.resource_ids
|
|
# Exclude filter
|
|
scope_exc = resolve_resource_scope([proj], exclude_resources=("docs",))
|
|
assert len(scope_exc.resource_ids) == 1
|
|
assert ulid1 in scope_exc.resource_ids
|
|
print("scoped-view-resolve-scope-ok")
|
|
return 0
|
|
except Exception as exc:
|
|
print(f"scoped-view-resolve-scope-fail: {exc}")
|
|
return 1
|
|
|
|
print(f"Unknown command: {command}")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|