File size: 1,926 Bytes
88e3f4a | 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 | from omniff.validators.pipeline import ValidationPipeline
from omniff.validators.text_validator import TextValidator, ValidationResult
def _always_pass(output):
return ValidationResult(True, 1.0, "pass")
def _always_fail(output):
return ValidationResult(False, 0.0, "fail", "forced failure")
def test_empty_pipeline():
pipe = ValidationPipeline()
result = pipe.run({"text": "hello"})
assert result.passed
assert result.results == []
def test_single_pass_success():
pipe = ValidationPipeline()
pipe.add_pass("length", TextValidator(min_length=1).validate)
result = pipe.run({"text": "hello world"})
assert result.passed
assert len(result.results) == 1
def test_single_pass_failure():
pipe = ValidationPipeline()
pipe.add_pass("length", TextValidator(min_length=100).validate, required=True)
result = pipe.run({"text": "short"})
assert not result.passed
assert result.failed_pass == "length"
def test_multi_pass_stops_on_required_failure():
pipe = ValidationPipeline()
pipe.add_pass("first", _always_fail, required=True)
pipe.add_pass("second", _always_pass, required=True)
result = pipe.run({})
assert not result.passed
assert len(result.results) == 1
assert result.failed_pass == "first"
def test_optional_failure_continues():
pipe = ValidationPipeline()
pipe.add_pass("optional", _always_fail, required=False)
pipe.add_pass("required", _always_pass, required=True)
result = pipe.run({})
assert result.passed
assert len(result.results) == 2
def test_pipeline_score():
pipe = ValidationPipeline()
pipe.add_pass("a", _always_pass)
pipe.add_pass("b", _always_pass)
result = pipe.run({})
assert result.score == 1.0
def test_pass_count():
pipe = ValidationPipeline()
pipe.add_pass("a", _always_pass)
pipe.add_pass("b", _always_fail)
assert pipe.pass_count == 2
|