File size: 2,874 Bytes
df98fca | 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 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | """Tests for the biological rule engine."""
from models import ActionType, ExperimentAction
from server.rules.engine import RuleEngine, Severity
from server.simulator.latent_state import (
ExperimentProgress,
FullLatentState,
ResourceState,
)
def _state(**progress_flags) -> FullLatentState:
return FullLatentState(
progress=ExperimentProgress(**progress_flags),
resources=ResourceState(budget_total=100_000, time_limit_days=180),
)
class TestPrerequisites:
def test_sequence_without_library_blocked(self):
engine = RuleEngine()
violations = engine.check(
ExperimentAction(action_type=ActionType.SEQUENCE_CELLS),
_state(samples_collected=True),
)
hard = engine.hard_violations(violations)
assert any("library" in m.lower() for m in hard)
def test_sequence_with_library_allowed(self):
engine = RuleEngine()
violations = engine.check(
ExperimentAction(action_type=ActionType.SEQUENCE_CELLS),
_state(samples_collected=True, library_prepared=True),
)
hard = engine.hard_violations(violations)
assert not hard
def test_de_without_normalization_blocked(self):
engine = RuleEngine()
violations = engine.check(
ExperimentAction(action_type=ActionType.DIFFERENTIAL_EXPRESSION),
_state(cells_sequenced=True, qc_performed=True, data_filtered=True),
)
hard = engine.hard_violations(violations)
assert any("normalis" in m.lower() or "normaliz" in m.lower() for m in hard)
def test_validate_marker_without_discovery_blocked(self):
engine = RuleEngine()
violations = engine.check(
ExperimentAction(action_type=ActionType.VALIDATE_MARKER),
_state(de_performed=True),
)
hard = engine.hard_violations(violations)
assert any("marker" in m.lower() for m in hard)
class TestRedundancy:
def test_double_qc_is_soft(self):
engine = RuleEngine()
violations = engine.check(
ExperimentAction(action_type=ActionType.RUN_QC),
_state(cells_sequenced=True, qc_performed=True),
)
hard = engine.hard_violations(violations)
soft = engine.soft_violations(violations)
assert not hard
assert any("redundant" in m.lower() for m in soft)
class TestResourceConstraints:
def test_exhausted_budget_blocked(self):
s = _state()
s.resources.budget_used = 100_000
engine = RuleEngine()
violations = engine.check(
ExperimentAction(action_type=ActionType.COLLECT_SAMPLE), s,
)
hard = engine.hard_violations(violations)
assert any("budget" in m.lower() for m in hard)
|