test_assist.mojo (5893B)
1 from std.collections import List 2 from std.testing import TestSuite, assert_equal, assert_raises, assert_true 3 4 from hyf_assist.questions import Question 5 from hyf_assist.evaluator import ( 6 DisabledSemanticEvaluator, 7 SemanticEvaluatorRequest, 8 SemanticEvaluatorResponse, 9 TypedAnswer, 10 normalize_score, 11 typed_choice, 12 typed_noul, 13 typed_score, 14 ) 15 16 17 def test_evaluator_boundary_is_typed_and_scriptable() raises: 18 var answers = List[TypedAnswer]() 19 answers.append(typed_choice("supply_status", "offered", 1.0)) 20 var response = SemanticEvaluatorResponse( 21 model="test-model", answers=answers^ 22 ) 23 assert_equal(response.model, "test-model") 24 assert_equal(len(response.answers), 1) 25 assert_equal(response.answers[0].kind, "choice") 26 assert_equal(response.answers[0].choice, "offered") 27 28 var disabled = DisabledSemanticEvaluator() 29 with assert_raises(): 30 _ = disabled.evaluate( 31 SemanticEvaluatorRequest(state="s", question_bundle="q") 32 ) 33 34 35 def test_typed_answers_and_score_normalization() raises: 36 assert_equal(typed_noul("q", 0.5).kind, "noul") 37 assert_equal(typed_score("s", 2, 3, 1.0).score, 2) 38 assert_equal(normalize_score(typed_score("s", 2, 3, 1.0), 3), 1.0) 39 assert_equal(normalize_score(typed_score("s", 1, 3, 1.0), 3), 0.5) 40 with assert_raises(): 41 _ = typed_choice("q", "x", 1.5) 42 with assert_raises(): 43 _ = typed_noul("q", -0.1) 44 with assert_raises(): 45 _ = typed_score("s", 5, 3, 1.0) 46 47 48 def main() raises: 49 TestSuite.discover_tests[__functions_in_module()]().run() 50 51 52 from hyf_runtime.clock import ( 53 Clock, 54 advance, 55 clock_monotonic_ns, 56 clock_wall_epoch_seconds, 57 fixed_clock, 58 monotonic_elapsed_ns, 59 system_monotonic_ns, 60 system_wall_epoch_seconds, 61 ) 62 63 64 def test_clock_ports_are_injectable_and_monotonic() raises: 65 var start = fixed_clock(1789000000, 1000) 66 assert_equal(clock_wall_epoch_seconds(start), 1789000000) 67 var end = advance(start, 5000) 68 assert_equal(clock_monotonic_ns(end), 6000) 69 assert_equal(monotonic_elapsed_ns(start, end), 5000) 70 assert_true(system_wall_epoch_seconds() > 1600000000) 71 assert_true(system_monotonic_ns() > 0) 72 73 74 from hyf_assist.scripted import scripted_evaluate, scripted_evaluator 75 76 77 def test_scripted_evaluator_is_strict_and_bounded() raises: 78 var answers = List[TypedAnswer]() 79 answers.append(typed_choice("supply_status", "offered", 1.0)) 80 var evaluator = scripted_evaluator("jev-1.13.0", answers, 1) 81 var response = scripted_evaluate( 82 evaluator, SemanticEvaluatorRequest(state="s", question_bundle="qb1") 83 ) 84 assert_equal(response.answers[0].choice, "offered") 85 assert_equal(evaluator.call_count, 1) 86 with assert_raises(): 87 _ = scripted_evaluate( 88 evaluator, 89 SemanticEvaluatorRequest(state="s", question_bundle="qb1"), 90 ) 91 with assert_raises(): 92 _ = scripted_evaluate( 93 evaluator, SemanticEvaluatorRequest(state="", question_bundle="qb1") 94 ) 95 96 97 from hyf_assist.questions import ( 98 bundle_pins_configuration, 99 choice_question, 100 noul_question, 101 question_bundle, 102 score_question, 103 ) 104 105 106 def test_question_bundles_are_versioned_and_validated() raises: 107 var choices = List[String]() 108 choices.append("offered") 109 choices.append("forecast") 110 choices.append("unavailable") 111 choices.append("unclear") 112 var questions = List[Question]() 113 questions.append(choice_question("supply_status", "status?", choices)) 114 questions.append(noul_question("seconds_ok", "seconds permitted?")) 115 var rubric = List[String]() 116 rubric.append("unsuitable") 117 rubric.append("limited") 118 rubric.append("suitable") 119 questions.append(score_question("culinary_fit", "fit?", rubric)) 120 var bundle = question_bundle("qb1", "1", "jev-1.13.0", questions) 121 assert_true(bundle_pins_configuration(bundle)) 122 assert_equal(len(bundle.questions), 3) 123 var too_few = List[String]() 124 too_few.append("only") 125 with assert_raises(): 126 _ = choice_question("x", "y", too_few) 127 var one = List[String]() 128 one.append("only") 129 with assert_raises(): 130 _ = score_question("s", "i", one) 131 132 133 from hyf_assist.question_plan import ( 134 QuestionStage, 135 independent_questions_share_one_state, 136 question_stage, 137 stages_are_acyclic, 138 ) 139 140 141 def test_question_stage_planning_dependencies() raises: 142 var choices = List[String]() 143 choices.append("offered") 144 choices.append("unclear") 145 var base_questions = List[Question]() 146 base_questions.append(choice_question("supply_status", "status?", choices)) 147 var follow_questions = List[Question]() 148 follow_questions.append(noul_question("seconds_ok", "seconds?")) 149 var stages = List[QuestionStage]() 150 stages.append(question_stage("stage-1", List[String](), base_questions)) 151 var deps = List[String]() 152 deps.append("stage-1") 153 stages.append(question_stage("stage-2", deps, follow_questions)) 154 assert_true(stages_are_acyclic(stages)) 155 assert_true(independent_questions_share_one_state()) 156 157 var cyclic = List[QuestionStage]() 158 cyclic.append(question_stage("a", List[String](), base_questions)) 159 var dep_a = List[String]() 160 dep_a.append("b") 161 cyclic.append(question_stage("b", dep_a, follow_questions)) 162 var dep_b = List[String]() 163 dep_b.append("a") 164 # replace first stage's deps to create a cycle: a depends on b, b depends on a 165 var cyclic2 = List[QuestionStage]() 166 var a_deps = List[String]() 167 a_deps.append("b") 168 cyclic2.append(question_stage("a", a_deps, base_questions)) 169 var b_deps = List[String]() 170 b_deps.append("a") 171 cyclic2.append(question_stage("b", b_deps, follow_questions)) 172 assert_true(not stages_are_acyclic(cyclic2)) 173 with assert_raises(): 174 _ = question_stage("empty", List[String](), List[Question]())