75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""Tests for the multi-pass pipeline."""
|
|
|
|
from physcom.engine.constraint_resolver import ConstraintResolver
|
|
from physcom.engine.scorer import Scorer
|
|
from physcom.engine.pipeline import Pipeline
|
|
from physcom.llm.providers.mock import MockLLMProvider
|
|
|
|
|
|
def test_pass1_filters_impossible_combos(seeded_repo):
|
|
"""Pass 1 should block known-impossible combinations (e.g., solar sail + walking)."""
|
|
domain = seeded_repo.get_domain("urban_commuting")
|
|
resolver = ConstraintResolver()
|
|
scorer = Scorer(domain)
|
|
pipeline = Pipeline(seeded_repo, resolver, scorer)
|
|
|
|
result = pipeline.run(domain, ["platform", "actuator", "energy_storage"], passes=[1])
|
|
|
|
from physcom.seed.transport_example import PLATFORMS, ACTUATORS, ENERGY_STORAGES
|
|
expected = len(PLATFORMS) * len(ACTUATORS) * len(ENERGY_STORAGES)
|
|
assert result.total_generated == expected
|
|
assert result.pass1_failed > 0
|
|
assert result.pass1_valid + result.pass1_conditional + result.pass1_failed == expected
|
|
|
|
|
|
def test_pass123_produces_scored_results(seeded_repo):
|
|
"""Passes 1-3 should produce a scored shortlist."""
|
|
domain = seeded_repo.get_domain("urban_commuting")
|
|
resolver = ConstraintResolver()
|
|
scorer = Scorer(domain)
|
|
pipeline = Pipeline(seeded_repo, resolver, scorer)
|
|
|
|
result = pipeline.run(
|
|
domain, ["platform", "actuator", "energy_storage"],
|
|
score_threshold=0.01, passes=[1, 2, 3, 5],
|
|
)
|
|
|
|
assert result.pass2_estimated > 0
|
|
assert result.pass3_above_threshold > 0
|
|
|
|
|
|
def test_pass4_with_mock_llm(seeded_repo):
|
|
"""Pass 4 should annotate with LLM review."""
|
|
domain = seeded_repo.get_domain("urban_commuting")
|
|
resolver = ConstraintResolver()
|
|
scorer = Scorer(domain)
|
|
mock_llm = MockLLMProvider(default_estimates={
|
|
"speed": 50.0, "cost_efficiency": 0.5, "safety": 0.6,
|
|
"availability": 0.7, "range_fuel": 200.0,
|
|
})
|
|
pipeline = Pipeline(seeded_repo, resolver, scorer, llm=mock_llm)
|
|
|
|
result = pipeline.run(
|
|
domain, ["platform", "actuator", "energy_storage"],
|
|
score_threshold=0.01, passes=[1, 2, 3, 4, 5],
|
|
)
|
|
|
|
assert result.pass4_reviewed > 0
|
|
|
|
|
|
def test_blocked_combos_not_scored(seeded_repo):
|
|
"""Blocked combinations should not make it to scoring."""
|
|
domain = seeded_repo.get_domain("urban_commuting")
|
|
resolver = ConstraintResolver()
|
|
scorer = Scorer(domain)
|
|
pipeline = Pipeline(seeded_repo, resolver, scorer)
|
|
|
|
result = pipeline.run(
|
|
domain, ["platform", "actuator", "energy_storage"],
|
|
score_threshold=0.0, passes=[1, 2, 3, 5],
|
|
)
|
|
|
|
# Estimated count should be less than total (blocked ones filtered)
|
|
assert result.pass2_estimated < result.total_generated
|
|
assert result.pass2_estimated == result.pass1_valid + result.pass1_conditional
|