tegwick
fc8a7dbb49
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0cffe-206f-7323-89d3-b20ca4cf4bf9
133 lines
4.9 KiB
Python
133 lines
4.9 KiB
Python
"""Integration checks use real bench workflows and deterministic responses."""
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
from infospace_bench import register_artifact
|
|
from infospace_bench.workflow import FixtureAssistedGenerationAdapter
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location("research_bench", REPO / "scripts/research_bench.py")
|
|
bench = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(bench)
|
|
FIXTURES = REPO / "research/bench/fixtures"
|
|
|
|
|
|
def initialize(tmp_path):
|
|
return bench.initialize(FIXTURES / "source.md", tmp_path, "synthetic-probe")
|
|
|
|
|
|
def adapter():
|
|
return FixtureAssistedGenerationAdapter.from_file(FIXTURES / "responses.yaml")
|
|
|
|
|
|
def test_pipeline_carries_source_and_previous_outputs_into_review(tmp_path):
|
|
root = initialize(tmp_path)
|
|
delegate = adapter()
|
|
prompts = {}
|
|
|
|
class Capture:
|
|
def generate(self, request):
|
|
prompts[request.stage_id] = request.prompt
|
|
return delegate.generate(request)
|
|
|
|
result = bench.execute(root, Capture())
|
|
assert result["provider_calls_attempted"] == 4
|
|
assert result["status"] == "completed"
|
|
assert "server can reject submission" in prompts["critique"]
|
|
assert "# Synthetic claims" in prompts["critique"]
|
|
assert "# Local Draft State" in prompts["critique"]
|
|
assert "only P01 exists" in prompts["critique"]
|
|
assert "{{" not in prompts["critique"]
|
|
assert (root / "output/evaluations/source-review.md").exists()
|
|
index = (root / "artifacts/index.yaml").read_text()
|
|
assert "provider: fixture" in index
|
|
assert result["review_status"] == "unreviewed"
|
|
assert result["evidence_kind"] == "synthetic"
|
|
assert not list((root / "artifacts/entities").glob("*.md"))
|
|
assert not (root / "patterns").exists()
|
|
with pytest.raises(ValueError, match="already attempted"):
|
|
bench.execute(root, Capture())
|
|
|
|
|
|
def test_mutated_source_and_multiple_sources_are_rejected(tmp_path):
|
|
root = initialize(tmp_path)
|
|
source = root / "artifacts/sources/source.md"
|
|
original = source.read_bytes()
|
|
source.write_text("changed")
|
|
with pytest.raises(ValueError, match="Source changed"):
|
|
bench.plan(root)
|
|
source.write_bytes(original)
|
|
register_artifact(root, artifact_id="source/second.md", path=source,
|
|
kind="source", title="Second")
|
|
with pytest.raises(ValueError, match="Exactly one"):
|
|
bench.plan(root)
|
|
|
|
|
|
def test_call_cap_prevents_provider_use(tmp_path):
|
|
root = initialize(tmp_path)
|
|
|
|
class NeverCall:
|
|
def generate(self, request):
|
|
pytest.fail("Provider must not be called")
|
|
|
|
with pytest.raises(ValueError, match="exceed"):
|
|
bench.execute(root, NeverCall(), max_calls=3)
|
|
assert not (root / "reports/research-run.json").exists()
|
|
|
|
|
|
def test_failure_is_recorded_and_cannot_be_silently_retried(tmp_path):
|
|
root = initialize(tmp_path)
|
|
|
|
class Fail:
|
|
def generate(self, request):
|
|
raise RuntimeError("fixture failure")
|
|
|
|
with pytest.raises(RuntimeError):
|
|
bench.execute(root, Fail())
|
|
receipt = json.loads((root / "reports/research-run.json").read_text())
|
|
assert receipt["status"] == "failed"
|
|
assert receipt["provider_calls_attempted"] == 1
|
|
with pytest.raises(ValueError, match="already attempted"):
|
|
bench.execute(root, adapter())
|
|
|
|
|
|
def test_unidentifiable_source_is_rejected_before_creating_workspace(tmp_path):
|
|
source = tmp_path / "unidentified.md"
|
|
source.write_text("# No bibliography")
|
|
with pytest.raises(ValueError, match="frontmatter"):
|
|
bench.initialize(source, tmp_path, "invalid")
|
|
assert not (tmp_path / "infospaces/invalid").exists()
|
|
|
|
|
|
def test_negative_extraction_is_a_valid_result(tmp_path):
|
|
from infospace_bench.workflow import AssistedGenerationResult
|
|
|
|
root = initialize(tmp_path)
|
|
|
|
class Negative:
|
|
def generate(self, request):
|
|
return AssistedGenerationResult(
|
|
markdown="# No supported candidates\n\nInsufficient synthetic evidence.\n",
|
|
provider="fixture",
|
|
)
|
|
|
|
assert bench.execute(root, Negative())["status"] == "completed"
|
|
assert not list((root / "artifacts/entities").glob("*.md"))
|
|
|
|
|
|
def test_cli_refuses_fixture_output_on_real_source(tmp_path):
|
|
source = tmp_path / "source.md"
|
|
source.write_text((FIXTURES / "source.md").read_text().replace(
|
|
"evidence_kind: synthetic", "evidence_kind: primary"))
|
|
root = bench.initialize(source, tmp_path, "primary-probe")
|
|
result = subprocess.run([
|
|
sys.executable, str(REPO / "scripts/research_bench.py"), "run", str(root),
|
|
"--fixture-responses", str(FIXTURES / "responses.yaml"),
|
|
], capture_output=True, text=True)
|
|
assert result.returncode != 0
|
|
assert "only run against synthetic" in result.stderr
|
|
assert not (root / "reports/research-run.json").exists()
|