root/Jyotisha
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

test(rectification): cover upstream acceptance

Browse Source
This commit is contained in:
Jesse_Chen
2026-09-04 03:06:43 +08:00
parent ae79911859
commit 4b711b72cb
2 changed files with 169 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_active_rectification_selector.py
+144 -1
View File
@@ -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"
tests/test_candidate_time_sensitivity_scan.py
+25
View File
@@ -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