Spaces:
Running
Running
File size: 1,352 Bytes
d6ee0a6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 | import pytest
pytest.importorskip("langgraph")
from ragforge.pipeline import RAGEngine
from ragforge.schemas import PipelineConfig
class _SQLStub:
tables = {}
class _EmptyWorkspace:
chunks = []
source_profiles = {}
sql = _SQLStub()
history = []
def manifest(self, *args, **kwargs):
return "No unstructured documents are indexed. Structured tables: none."
def test_document_route_preflight_abstains_before_retrieval():
engine = object.__new__(RAGEngine)
engine.workspace = _EmptyWorkspace()
state = {
"query": "What are the documents about?",
"config": PipelineConfig(mode="Documents", profile="Fast"),
"trace": {"nodes": []},
}
out = engine._route(state)
assert out["route"] == "documents"
assert out["abstain_reason"] == "workspace_empty_documents"
def test_abstain_is_terminal_operational_response():
engine = object.__new__(RAGEngine)
engine.workspace = _EmptyWorkspace()
state = {
"abstain_reason": "workspace_empty_documents",
"doc_hits": [],
"web_hits": [],
"trace": {"nodes": []},
}
out = engine._abstain(state)
assert "No document corpus is indexed" in out["answer"]
assert out["sources"] == []
assert out["confidence"] >= 0.9
assert out["trace"]["nodes"][-1]["node"] == "abstain"
|