258 lines
9.5 KiB
Python
258 lines
9.5 KiB
Python
"""asf-context tests + relationship_service walk using in-memory repo mocks.
|
|
|
|
Covers:
|
|
* In-memory FakeRelationshipRepo (outgoing / incoming / for_artifact / create / delete)
|
|
* ContextEngine BFS 2-hop with whitelist + cap via fake services
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
|
|
from asf_domain import (
|
|
Artifact,
|
|
ArtifactType,
|
|
ArtifactRelationship,
|
|
ArtifactRelationshipCreate,
|
|
RelationshipType,
|
|
AuditAction,
|
|
AuditEventCreate,
|
|
)
|
|
|
|
CONTEXT_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(CONTEXT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(CONTEXT_ROOT))
|
|
|
|
TENANT_ID = UUID("11111111-0000-0000-0000-000000000001")
|
|
ACTOR_ID = UUID("22222222-0000-0000-0000-000000000002")
|
|
PROJ_ID = UUID("33333333-0000-0000-0000-000000000003")
|
|
|
|
|
|
# =========================================================================
|
|
# In-memory fake relationship + audit repos (matches interface subset used)
|
|
# =========================================================================
|
|
|
|
class FakeRelationshipRepo:
|
|
def __init__(self) -> None:
|
|
self._rels: list[ArtifactRelationship] = []
|
|
|
|
async def create(self, tenant_id: UUID, body: ArtifactRelationshipCreate) -> ArtifactRelationship:
|
|
from datetime import datetime, timezone
|
|
now = datetime.now(tz=timezone.utc)
|
|
r = ArtifactRelationship(
|
|
id=uuid4(),
|
|
tenant_id=tenant_id,
|
|
relationship_type=body.relationship_type,
|
|
source_artifact_id=body.source_artifact_id,
|
|
target_artifact_id=body.target_artifact_id,
|
|
source_project_id=PROJ_ID,
|
|
target_project_id=PROJ_ID,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self._rels.append(r)
|
|
return r
|
|
|
|
async def outgoing(
|
|
self, tenant_id, artifact_id, *, relationship_type=None, limit=100
|
|
) -> list[ArtifactRelationship]:
|
|
out = [r for r in self._rels if r.tenant_id == tenant_id and r.source_artifact_id == artifact_id]
|
|
if relationship_type is not None:
|
|
out = [r for r in out if r.relationship_type == relationship_type]
|
|
return out[:limit]
|
|
|
|
async def incoming(
|
|
self, tenant_id, artifact_id, *, relationship_type=None, limit=100
|
|
) -> list[ArtifactRelationship]:
|
|
out = [r for r in self._rels if r.tenant_id == tenant_id and r.target_artifact_id == artifact_id]
|
|
if relationship_type is not None:
|
|
out = [r for r in out if r.relationship_type == relationship_type]
|
|
return out[:limit]
|
|
|
|
async def list_for_artifact(self, tenant_id, artifact_id, *, limit=200) -> list[ArtifactRelationship]:
|
|
out = [
|
|
r
|
|
for r in self._rels
|
|
if r.tenant_id == tenant_id and (r.source_artifact_id == artifact_id or r.target_artifact_id == artifact_id)
|
|
]
|
|
return out[:limit]
|
|
|
|
async def delete(self, tenant_id, relationship_id) -> bool:
|
|
before = len(self._rels)
|
|
self._rels = [r for r in self._rels if not (r.tenant_id == tenant_id and r.id == relationship_id)]
|
|
return len(self._rels) < before
|
|
|
|
|
|
class FakeAuditRepo:
|
|
def __init__(self) -> None:
|
|
self.events: list[tuple[UUID, AuditEventCreate]] = []
|
|
|
|
async def append(self, tenant_id: UUID, event: AuditEventCreate) -> None:
|
|
self.events.append((tenant_id, event))
|
|
|
|
|
|
# =========================================================================
|
|
# Lightweight "service-like" wrapper (same signatures as RelationshipService)
|
|
# =========================================================================
|
|
|
|
class FakeRelService:
|
|
"""Drop-in, dependency-free replacement for RelationshipService used by tests."""
|
|
|
|
def __init__(self, rels: FakeRelationshipRepo, audit: FakeAuditRepo) -> None:
|
|
self._tenant = TENANT_ID
|
|
self._rels = rels
|
|
self._audit = audit
|
|
|
|
async def create(self, body: ArtifactRelationshipCreate) -> ArtifactRelationship:
|
|
r = await self._rels.create(self._tenant, body)
|
|
await self._audit.append(
|
|
self._tenant,
|
|
AuditEventCreate(
|
|
action=AuditAction.CREATED,
|
|
resource_type="artifact_relationship",
|
|
resource_id=r.id,
|
|
actor_id=ACTOR_ID,
|
|
metadata={
|
|
"source": str(r.source_artifact_id),
|
|
"type": r.relationship_type,
|
|
"target": str(r.target_artifact_id),
|
|
},
|
|
),
|
|
)
|
|
return r
|
|
|
|
async def outgoing(self, artifact_id, *, relationship_type=None, limit=100):
|
|
return await self._rels.outgoing(self._tenant, artifact_id, relationship_type=relationship_type, limit=limit)
|
|
|
|
async def incoming(self, artifact_id, *, relationship_type=None, limit=100):
|
|
return await self._rels.incoming(self._tenant, artifact_id, relationship_type=relationship_type, limit=limit)
|
|
|
|
async def for_artifact(self, artifact_id, *, limit=200):
|
|
return await self._rels.list_for_artifact(self._tenant, artifact_id, limit=limit)
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_rels_repo():
|
|
return FakeRelationshipRepo()
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_audit_repo():
|
|
return FakeAuditRepo()
|
|
|
|
|
|
@pytest.fixture
|
|
def rel_service(fake_rels_repo, fake_audit_repo):
|
|
return FakeRelService(fake_rels_repo, fake_audit_repo)
|
|
|
|
|
|
@pytest.fixture
|
|
def chain_artifacts_ids():
|
|
return uuid4(), uuid4(), uuid4(), uuid4()
|
|
|
|
|
|
@pytest.fixture
|
|
async def seeded_service(rel_service, chain_artifacts_ids):
|
|
"""REQ --implements--> TASK --implements--> PR --generated_from--> BUILD (4 nodes, 3 edges, 2 hops from REQ)."""
|
|
a, b, c, d = chain_artifacts_ids
|
|
for rt, src, tgt in [
|
|
(RelationshipType.IMPLEMENTS, a, b),
|
|
(RelationshipType.IMPLEMENTS, b, c),
|
|
(RelationshipType.GENERATED_FROM, c, d),
|
|
]:
|
|
await rel_service.create(
|
|
ArtifactRelationshipCreate(
|
|
source_artifact_id=src, target_artifact_id=tgt, relationship_type=rt,
|
|
)
|
|
)
|
|
return rel_service, chain_artifacts_ids
|
|
|
|
|
|
# =========================================================================
|
|
# Tests
|
|
# =========================================================================
|
|
|
|
def _mk_artifact(aid: UUID, t: ArtifactType, title: str) -> Artifact:
|
|
from datetime import datetime, timezone
|
|
now = datetime.now(tz=timezone.utc)
|
|
return Artifact(
|
|
id=aid,
|
|
tenant_id=TENANT_ID,
|
|
project_id=PROJ_ID,
|
|
external_id=t.name + "-" + title,
|
|
type=t,
|
|
title=title,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
|
|
|
|
class TestFakeRelServiceCRUD:
|
|
@pytest.mark.asyncio
|
|
async def test_create_emits_audit_and_stores(self, rel_service, chain_artifacts_ids, fake_audit_repo):
|
|
a, b, *_ = chain_artifacts_ids
|
|
body = ArtifactRelationshipCreate(
|
|
source_artifact_id=a, target_artifact_id=b, relationship_type=RelationshipType.RELATED_TO,
|
|
)
|
|
r = await rel_service.create(body)
|
|
assert r.relationship_type == RelationshipType.RELATED_TO
|
|
_, ev = fake_audit_repo.events[0]
|
|
assert ev.action == AuditAction.CREATED
|
|
assert ev.resource_type == "artifact_relationship"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_outgoing_incoming_for_artifact(self, seeded_service):
|
|
svc, (a, b, c, d) = seeded_service
|
|
assert len(await svc.outgoing(a)) == 1
|
|
assert len(await svc.incoming(d)) == 1
|
|
assert len(await svc.for_artifact(b)) == 2 # a->b + b->c
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filter_by_relationship_type(self, seeded_service):
|
|
svc, (a, b, c, d) = seeded_service
|
|
assert len(await svc.outgoing(c, relationship_type=RelationshipType.GENERATED_FROM)) == 1
|
|
assert len(await svc.outgoing(c, relationship_type=RelationshipType.IMPLEMENTS)) == 0
|
|
|
|
|
|
class TestContextEngineBFS:
|
|
@pytest.mark.asyncio
|
|
async def test_bfs_depth_2_stops_before_third_node(self, seeded_service, chain_artifacts_ids):
|
|
from app.context_engine import ContextEngine, AssemblyOptions
|
|
a, b, c, d = chain_artifacts_ids
|
|
arts = [
|
|
_mk_artifact(a, ArtifactType.REQUIREMENT, "A"),
|
|
_mk_artifact(b, ArtifactType.TASK, "B"),
|
|
_mk_artifact(c, ArtifactType.PULL_REQUEST, "C"),
|
|
_mk_artifact(d, ArtifactType.BUILD, "D"),
|
|
]
|
|
|
|
class _ArtifactSvc:
|
|
def __init__(self): self._m = {x.id: x for x in arts}
|
|
async def get_many(self, ids):
|
|
return [self._m[i] for i in ids if i in self._m]
|
|
async def list(self, **kw):
|
|
return list(self._m.values())
|
|
|
|
class _RelSvc:
|
|
def __init__(self, svc): self._svc = svc
|
|
async def outgoing(self, aid, *, limit=100):
|
|
return await self._svc.outgoing(aid, limit=limit)
|
|
async def incoming(self, aid, *, limit=100):
|
|
return await self._svc.incoming(aid, limit=limit)
|
|
|
|
svc, *_ = seeded_service
|
|
engine = ContextEngine(relationship_service=_RelSvc(svc), artifact_service=_ArtifactSvc())
|
|
opts = AssemblyOptions(max_depth=2, max_nodes=50)
|
|
asm = await engine.assemble(tenant_id=TENANT_ID, seed_artifact_id=a, project_id=PROJ_ID, options=opts)
|
|
summary = asm.summary()
|
|
visited = {x.id for x in asm.all_artifacts}
|
|
# depth=2 from a: visits a (0), b (1), c (2). NOT d (depth 3).
|
|
assert a in visited and b in visited and c in visited
|
|
assert d not in visited
|
|
assert summary["total_relationships"] >= 2 # a->b + b->c traversed
|
|
assert summary["total_artifacts"] == 3
|