127 lines
4.1 KiB
Python
127 lines
4.1 KiB
Python
"""Tests for the scripting API (Session, ChannelHandle, TransformProxy)."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from impakt import Session, Template
|
|
|
|
FIXTURE_DATA = Path(__file__).parent / "fixtures" / "sample_mme"
|
|
MME_DATA = Path(__file__).parent / "mme_data"
|
|
|
|
|
|
class TestSession:
|
|
def test_open(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
assert s.test_id == "IMPAKT_SYNTH_001"
|
|
assert len(s) == 26
|
|
|
|
def test_channel_access(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
ch = s.channel("11HEAD0000ACXA")
|
|
assert ch.name == "11HEAD0000ACXA"
|
|
assert ch.peak > 0
|
|
|
|
def test_find(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
channels = s.find("*HEAD*AC*")
|
|
assert len(channels) == 3
|
|
|
|
def test_group(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
group = s.group("HEAD0000AC")
|
|
assert group.x is not None
|
|
|
|
def test_compute_criteria(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
criteria = s.compute_criteria()
|
|
assert len(criteria) > 0
|
|
|
|
def test_evaluate(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
result = s.evaluate("euro_ncap")
|
|
assert result.stars is not None
|
|
assert result.protocol == "Euro NCAP"
|
|
|
|
def test_evaluate_us_ncap(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
result = s.evaluate("us_ncap")
|
|
assert result.stars is not None
|
|
|
|
def test_evaluate_iihs(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
result = s.evaluate("iihs")
|
|
assert result.overall_rating in ("GOOD", "ACCEPTABLE", "MARGINAL", "POOR")
|
|
|
|
def test_evaluate_invalid_protocol(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
with pytest.raises(ValueError, match="Unknown protocol"):
|
|
s.evaluate("invalid")
|
|
|
|
def test_contains(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
assert "11HEAD0000ACXA" in s
|
|
assert "NONEXISTENT" not in s
|
|
|
|
|
|
class TestChannelHandleChaining:
|
|
"""The fluent API must support chaining — each transform returns ChannelHandle."""
|
|
|
|
def test_single_transform(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
ch = s.channel("11HEAD0000ACXA")
|
|
filtered = ch.transform.cfc(600)
|
|
assert type(filtered).__name__ == "ChannelHandle"
|
|
assert filtered.raw.cfc_class == 600
|
|
|
|
def test_double_chain(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
result = s.channel("11HEAD0000ACXA").transform.cfc(600).transform.y_align()
|
|
assert type(result).__name__ == "ChannelHandle"
|
|
assert len(result.raw.transform_history) == 2
|
|
|
|
def test_triple_chain(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
result = (
|
|
s.channel("11HEAD0000ACXA")
|
|
.transform.cfc(1000)
|
|
.transform.y_align()
|
|
.transform.trim(t_start=0.0, t_end=0.1)
|
|
)
|
|
assert type(result).__name__ == "ChannelHandle"
|
|
assert len(result.raw.transform_history) == 3
|
|
|
|
def test_chain_preserves_data(self):
|
|
s = Session.open(FIXTURE_DATA)
|
|
original = s.channel("11HEAD0000ACXA")
|
|
original_peak = original.peak
|
|
filtered = original.transform.cfc(600)
|
|
# Original should be unchanged — peak should be the same
|
|
assert original.peak == original_peak
|
|
# Filtered should have different CFC and lower peak (smoothed)
|
|
assert filtered.raw.cfc_class == 600
|
|
assert filtered.peak <= original_peak
|
|
|
|
|
|
@pytest.mark.skipif(not (MME_DATA / "3239").exists(), reason="Real data not available")
|
|
class TestSessionRealData:
|
|
def test_open_real(self):
|
|
s = Session.open(MME_DATA / "3239")
|
|
assert s.test_id == "3239"
|
|
assert len(s) == 133
|
|
|
|
def test_full_pipeline(self):
|
|
s = Session.open(MME_DATA / "3239")
|
|
# Chain: get channel -> filter -> check
|
|
ch = s.channel("11HEAD0000H3ACXP").transform.cfc(1000)
|
|
assert ch.peak > 100 # Significant head acceleration
|
|
|
|
# Compute criteria
|
|
criteria = s.compute_criteria()
|
|
assert "HIC15" in criteria
|
|
|
|
# Evaluate
|
|
result = s.evaluate("euro_ncap")
|
|
assert result.stars is not None
|
|
assert result.stars >= 0
|