From 4b711b72cb803d87e5bc91af9cedbad8a00c8d3d Mon Sep 17 00:00:00 2001 From: Jesse_Chen Date: Fri, 4 Sep 2026 03:06:43 +0800 Subject: [PATCH] test(rectification): cover upstream acceptance --- tests/test_active_rectification_selector.py | 145 +++++++++++++++++- tests/test_candidate_time_sensitivity_scan.py | 25 +++ 2 files changed, 169 insertions(+), 1 deletion(-) diff --git a/tests/test_active_rectification_selector.py b/tests/test_active_rectification_selector.py index f20db8da..050ac1cb 100644 --- a/tests/test_active_rectification_selector.py +++ b/tests/test_active_rectification_selector.py @@ -1,7 +1,11 @@ from __future__ import annotations -from scripts.active_rectification_questions import build_questionnaire +import hashlib +import json + +from scripts.active_rectification_questions import build_questionnaire, score_answers from scripts.active_rectification_selector import select_next_questions +from scripts.rectification.scoring_service import score_from_matrix def test_selector_asks_one_question_and_ranks_by_separation() -> None: @@ -117,6 +121,22 @@ def _minute_question(question_id: str, domain: str, layer: str) -> dict[str, obj } +def _minute_window(*changing_layers: str) -> dict[str, object]: + return { + "candidate_count": 3, + "transitions": [{"between": ["10:19", "10:20"]}], + "rows": [ + { + "divisional_ascendants": { + layer: {"sign": sign if layer in changing_layers else "Leo"} + for layer in ("D9", "D10", "D24") + } + } + for sign in ("Aries", "Taurus", "Gemini") + ], + } + + def test_selector_changes_with_remaining_candidate_window() -> None: questions = [ _minute_question("relationship_split", "relationship", "D9"), @@ -154,3 +174,126 @@ def test_selector_changes_with_remaining_candidate_window() -> None: assert second["selected_questions"][0]["id"] == "career_split" assert first["selected_questions"][0]["minute_relevance"] > 0 assert second["selected_questions"][0]["minute_relevance"] > 0 + + +def test_fictional_adaptive_interview_records_each_selected_question_and_minute_relevance() -> None: + questions = [ + {**_minute_question("career_split", "career", "D10"), "round": 1}, + {**_minute_question("education_split", "education", "D24"), "round": 2}, + {**_minute_question("relationship_split", "relationship", "D9"), "round": 3}, + ] + questionnaire = { + "question_bank": questions, + "candidate_scan": {"candidate_count": 3, "minute_scan": _minute_window("D9", "D10", "D24")}, + } + answers: dict[str, str] = {} + transcript = [] + + for round_number, answer in enumerate(("A", "B", "C"), start=1): + selection = select_next_questions(questionnaire, answers, limit=1) + selected = selection["selected_questions"][0] + transcript.append({ + "round": round_number, + "question": selected["id"], + "minute_relevance": selected["minute_relevance"], + }) + answers[selected["id"]] = answer + + assert transcript == [ + {"round": 1, "question": "career_split", "minute_relevance": 4.35}, + {"round": 2, "question": "education_split", "minute_relevance": 4.35}, + {"round": 3, "question": "relationship_split", "minute_relevance": 4.35}, + ] + assert select_next_questions(questionnaire, answers)["stop"] is True + + +def test_selector_changes_do_not_change_legacy_or_v5_score_bytes() -> None: + questions = [ + {**_minute_question("relationship_split", "relationship", "D9"), "round": 1}, + {**_minute_question("career_split", "career", "D10"), "round": 2}, + ] + answers = {"relationship_split": "A", "career_split": "C"} + questionnaires = [ + { + "questions": questions, + "candidate_scan": {"candidate_count": 3, "minute_scan": _minute_window("D9")}, + }, + { + "questions": questions, + "candidate_scan": {"candidate_count": 3, "minute_scan": _minute_window("D10")}, + }, + ] + selections = [select_next_questions(questionnaire, {}) for questionnaire in questionnaires] + assert [selection["selected_questions"][0]["id"] for selection in selections] == [ + "relationship_split", + "career_split", + ] + + legacy_bytes = [] + for questionnaire in questionnaires: + scored = score_answers(questionnaire, answers) + legacy_contract = { + key: value + for key, value in scored.items() + if key not in {"calculation", "next_round_selection"} + } + legacy_bytes.append(json.dumps( + legacy_contract, + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ).encode()) + assert legacy_bytes[0] == legacy_bytes[1] + assert hashlib.sha256(legacy_bytes[0]).hexdigest() == "bb1a300606458357b2b8d94a487272f15d82c6f0b51847c660ba20153773fb63" + + v5_request = { + "birth_date": "1997-08-08", + "start_time": "05:13", + "end_time": "05:14", + "lat": 36.419, + "lon": 114.213, + "tz": 8.0, + "events": [ + { + "id": "00000000-0000-4000-8000-000000000001", + "domain": "education", + "event_kind": "education_start", + "date_start": "2016-09-01", + "date_end": "2016-09-30", + "precision": "month", + "summary": "fictional enrollment", + "subject": "self", + }, + { + "id": "00000000-0000-4000-8000-000000000002", + "domain": "career", + "event_kind": "career_entry", + "date_start": "2020-07-01", + "date_end": "2020-07-31", + "precision": "month", + "summary": "fictional first role", + "subject": "self", + }, + ], + } + built = { + "candidate_times": ["05:13", "05:14"], + "matrix": { + v5_request["events"][0]["id"]: { + "05:13": {"points": 4.0, "rule_ids": ["D24:fixture"]}, + "05:14": {"points": 1.0, "rule_ids": ["D24:fixture"]}, + }, + v5_request["events"][1]["id"]: { + "05:13": {"points": -1.0, "rule_ids": ["D10:fixture"]}, + "05:14": {"points": 3.0, "rule_ids": ["D10:fixture"]}, + }, + }, + "missing_layers": [], + } + v5_bytes = json.dumps( + score_from_matrix(v5_request, built), + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ).encode() + assert hashlib.sha256(v5_bytes).hexdigest() == "23171c47b9746f4b0a440c9b9ac6d8401d622672cd0d6d780789334b732e9d8f" diff --git a/tests/test_candidate_time_sensitivity_scan.py b/tests/test_candidate_time_sensitivity_scan.py index 3cea8414..fbf9d44b 100644 --- a/tests/test_candidate_time_sensitivity_scan.py +++ b/tests/test_candidate_time_sensitivity_scan.py @@ -30,3 +30,28 @@ def test_scanner_reports_real_divisional_transitions(monkeypatch): assert report["input_contract"]["settings"]["node_mode"] == "mean" assert report["rows"][0]["input_fingerprint"] != report["rows"][1]["input_fingerprint"] assert report["stability_contract"]["minute_confirmation_allowed"] is False + + +def test_scanner_keeps_raman_default_and_applies_explicit_lahiri() -> None: + birth = { + "year": 2000, + "month": 1, + "day": 1, + "hour": 12, + "minute": 1, + "lat": 25.04, + "lon": 121.56, + "tz": 8, + } + + default_report = scanner.scan_candidate_times(birth, uncertainty_minutes=1) + lahiri_report = scanner.scan_candidate_times( + {**birth, "ayanamsa": "lahiri"}, + uncertainty_minutes=1, + ) + default_center = next(row for row in default_report["rows"] if row["offset_minutes"] == 0) + lahiri_center = next(row for row in lahiri_report["rows"] if row["offset_minutes"] == 0) + + assert default_report["input_contract"]["settings"]["ayanamsa"] == "raman" + assert lahiri_report["input_contract"]["settings"]["ayanamsa"] == "lahiri" + assert abs(default_center["d1_longitude"] - lahiri_center["d1_longitude"]) > 1