(scenario, db_path)
| 535 | |
| 536 | |
| 537 | def evaluate_assertions(scenario, db_path): |
| 538 | outcomes = [] |
| 539 | assertions = [ |
| 540 | assertion |
| 541 | for assertion in scenario["assertions"] |
| 542 | if assertion_applies_to_real_model(assertion) |
| 543 | ] |
| 544 | if not assertions: |
| 545 | return outcomes |
| 546 | |
| 547 | try: |
| 548 | db = sqlite3.connect(f"{db_path.resolve().as_uri()}?mode=ro", uri=True) |
| 549 | except sqlite3.Error as exc: |
| 550 | return [ |
| 551 | failed_assertion( |
| 552 | assertion, |
| 553 | f"failed to open TraceDecay store: {exc}", |
| 554 | error_type="sqlite", |
| 555 | ) |
| 556 | for assertion in assertions |
| 557 | ] |
| 558 | |
| 559 | for assertion in scenario["assertions"]: |
| 560 | if not assertion_applies_to_real_model(assertion): |
| 561 | continue |
| 562 | if assertion["kind"] != "sql": |
| 563 | outcomes.append( |
| 564 | failed_assertion( |
| 565 | assertion, |
| 566 | f"unsupported assertion kind for real-model eval: {assertion['kind']}", |
| 567 | ) |
| 568 | ) |
| 569 | continue |
| 570 | expected = assertion["value"] |
| 571 | op = assertion["op"] |
| 572 | try: |
| 573 | row = db.execute(assertion["sql"]).fetchone() |
| 574 | actual = row[0] if row is not None else None |
| 575 | except sqlite3.Error as exc: |
| 576 | outcomes.append( |
| 577 | failed_assertion( |
| 578 | assertion, |
| 579 | str(exc), |
| 580 | error_type="sqlite", |
| 581 | actual=None, |
| 582 | op=op, |
| 583 | expected=expected, |
| 584 | ) |
| 585 | ) |
| 586 | continue |
| 587 | try: |
| 588 | passed = compare_assertion_value(actual, expected, op) |
| 589 | except TypeError as exc: |
| 590 | outcomes.append( |
| 591 | failed_assertion( |
| 592 | assertion, |
| 593 | f"could not compare assertion values: {exc}", |
| 594 | actual=actual, |
no test coverage detected