57 lines
2.3 KiB
Python
57 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from kab_ingestion.models import IngestionRequest, SourceDocument
|
|
from app.workflows import create_ingestion_workflow
|
|
from app.states.ingestion_state import IngestionState
|
|
|
|
|
|
class FakeSCM:
|
|
def __init__(self, documents: list[SourceDocument]):
|
|
self.documents = documents
|
|
|
|
def fetch(self, request: IngestionRequest) -> list[SourceDocument]:
|
|
return self.documents
|
|
|
|
|
|
def test_4_phase_workflow_success_path():
|
|
doc = SourceDocument("doc-100", "Architecture Guide", "Detailed content body", "github", metadata={"governance_approved": True})
|
|
scm = FakeSCM([doc])
|
|
workflow = create_ingestion_workflow(scm=scm, publish_enabled=False, require_governance_approval=True)
|
|
|
|
request = IngestionRequest(source="github", locator="owner/repo/contents/guide.md")
|
|
initial_state: IngestionState = {"request": request, "tenant_id": "tenant-az"}
|
|
|
|
final_state = workflow.invoke(initial_state)
|
|
|
|
assert final_state.get("current_step") == "deployment"
|
|
assert not final_state.get("has_errors")
|
|
assert final_state.get("existing_resources", {}).get("connector_available") is True
|
|
assert final_state.get("validation_errors") == []
|
|
assert len(final_state.get("fetched_documents", [])) == 1
|
|
assert len(final_state.get("published_results", [])) == 1
|
|
assert final_state.get("published_results", [])[0].status == "validated"
|
|
|
|
|
|
def test_validation_phase_fails_fast_on_invalid_source():
|
|
workflow = create_ingestion_workflow(allowed_sources={"github"})
|
|
request = IngestionRequest(source="unsupported_source", locator="some/locator")
|
|
initial_state: IngestionState = {"request": request}
|
|
|
|
final_state = workflow.invoke(initial_state)
|
|
|
|
assert final_state.get("current_step") == "validation"
|
|
assert final_state.get("has_errors") is True
|
|
assert "unsupported source 'unsupported_source'" in final_state.get("validation_errors", [])
|
|
assert final_state.get("next_step") == "END"
|
|
|
|
|
|
def test_validation_phase_fails_fast_on_missing_locator():
|
|
workflow = create_ingestion_workflow(allowed_sources={"github"})
|
|
request = IngestionRequest(source="github", locator="")
|
|
initial_state: IngestionState = {"request": request}
|
|
|
|
final_state = workflow.invoke(initial_state)
|
|
|
|
assert final_state.get("has_errors") is True
|
|
assert "missing request locator" in final_state.get("validation_errors", [])
|