Jesse_Chen
629 lines
30 KiB
Python
629 lines
30 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import date
|
|
from unittest.mock import patch
|
|
from uuid import UUID
|
|
|
|
from scripts.active_rectification_event_engine import _candidate_datetimes
|
|
from scripts.jyotish_api_server import (
|
|
API_COMMAND_MAP,
|
|
TECHNIQUE_EXAMPLE_ENDPOINTS,
|
|
BadRequest,
|
|
JyotishAPIHandler,
|
|
)
|
|
from scripts.rectification.api_service import diagnostics, score_candidates
|
|
from scripts.rectification.contracts import normalize_rectification_request
|
|
from scripts.rectification.scoring_service import (
|
|
build_event_contribution_matrix,
|
|
calculation_spec,
|
|
sample_event_dates,
|
|
score_from_matrix,
|
|
sha256,
|
|
)
|
|
|
|
EVENT_ID = "00000000-0000-4000-8000-000000000001"
|
|
|
|
|
|
def event(index: int, domain: str, event_kind: str, *, precision: str = "day"):
|
|
return {
|
|
"id": f"00000000-0000-4000-8000-{index:012d}",
|
|
"domain": domain,
|
|
"event_kind": event_kind,
|
|
"date_start": "2016-09-15",
|
|
"date_end": "2016-09-15",
|
|
"precision": precision,
|
|
"summary": event_kind,
|
|
}
|
|
|
|
|
|
def request(
|
|
*,
|
|
precision: str = "month",
|
|
event_kind: str = "education_milestone",
|
|
domain: str = "education",
|
|
start_time: str = "05:13",
|
|
end_time: str = "05:15",
|
|
date_start: str = "2016-09-01",
|
|
date_end: str = "2016-09-30",
|
|
):
|
|
return {
|
|
"birth_date": "1997-08-08",
|
|
"start_time": start_time,
|
|
"end_time": end_time,
|
|
"lat": 36.419,
|
|
"lon": 114.213,
|
|
"tz": 8,
|
|
"events": [{
|
|
"id": EVENT_ID,
|
|
"domain": domain,
|
|
"event_kind": event_kind,
|
|
"date_start": date_start,
|
|
"date_end": date_end,
|
|
"precision": precision,
|
|
"summary": "大学入学",
|
|
}],
|
|
}
|
|
|
|
|
|
class RectificationV5ServicesTest(unittest.TestCase):
|
|
def test_event_contract_v2_accepts_native_kinds_and_retains_background_events(self):
|
|
supported = {
|
|
"education": ("education_start", "education_completion", "education_interruption", "education_change"),
|
|
"career": ("career_entry", "career_change", "promotion", "career_pressure", "career_exit", "business_start"),
|
|
"relationship": ("relationship_start", "relationship_commitment", "relationship_separation", "relationship_end"),
|
|
"relocation": ("relocation", "foreign_move", "return", "home_change"),
|
|
"finance": ("finance_gain", "finance_loss", "income_change", "asset_change"),
|
|
"health": ("self_health_event", "pressure_period"),
|
|
"family": ("family_event",),
|
|
"other": ("other",),
|
|
}
|
|
|
|
for domain, kinds in supported.items():
|
|
for event_kind in kinds:
|
|
normalized = normalize_rectification_request(
|
|
request(domain=domain, event_kind=event_kind),
|
|
today=date(2026, 7, 28),
|
|
)
|
|
self.assertEqual(normalized["events"][0]["domain"], domain)
|
|
self.assertEqual(normalized["events"][0]["event_kind"], event_kind)
|
|
|
|
def test_event_contract_v2_preserves_server_provenance_and_enforces_self_scoring(self):
|
|
body = request(domain="career", event_kind="career_entry")
|
|
body["events"][0].update({
|
|
"source_turn_id": "33333333-3333-4333-8333-333333333333",
|
|
"subject": "self",
|
|
"date_source": "user_quote",
|
|
"date_reliability": "month_exact",
|
|
"date_corroboration": "劳动合同",
|
|
"date_conflict_status": "none",
|
|
})
|
|
normalized = normalize_rectification_request(body, today=date(2026, 7, 28))
|
|
event_value = normalized["events"][0]
|
|
self.assertEqual(event_value["source_turn_id"], "33333333-3333-4333-8333-333333333333")
|
|
self.assertEqual(event_value["subject"], "self")
|
|
self.assertEqual(event_value["date_source"], "user_quote")
|
|
self.assertEqual(event_value["date_reliability"], "month_exact")
|
|
self.assertEqual(event_value["date_corroboration"], "劳动合同")
|
|
self.assertEqual(event_value["date_conflict_status"], "none")
|
|
|
|
body["events"][0]["subject"] = "family"
|
|
with self.assertRaisesRegex(ValueError, "subject must be self for scoreable events"):
|
|
normalize_rectification_request(body, today=date(2026, 7, 28))
|
|
|
|
def test_background_only_events_are_retained_without_invoking_the_scoring_engine(self):
|
|
body = request()
|
|
body["events"] = [
|
|
event(1, "family", "family_event"),
|
|
event(2, "other", "other", precision="year"),
|
|
]
|
|
normalized = normalize_rectification_request(body, today=date(2026, 7, 28))
|
|
|
|
def fail_if_called(_request):
|
|
self.fail("background events must not reach the scoring engine")
|
|
|
|
built = build_event_contribution_matrix(normalized, row_provider=fail_if_called)
|
|
|
|
self.assertEqual(built["candidate_times"], [])
|
|
self.assertEqual(built["matrix"], {})
|
|
self.assertEqual(built["date_sensitivity"], [])
|
|
|
|
def test_declared_date_precision_changes_real_contribution_weight(self):
|
|
def rows(value):
|
|
event = value["events"][0]
|
|
return [{
|
|
"time": "05:13",
|
|
"score": 10,
|
|
"evidence": [{
|
|
"event_id": event["id"],
|
|
"domain": event["domain"],
|
|
"candidate_time": "05:13",
|
|
"rule_ids": ["vim_md_domain_house"],
|
|
"points": 10,
|
|
}],
|
|
"missing_layers": [],
|
|
}]
|
|
|
|
cases = {
|
|
"day": request(precision="day", event_kind="education_start", date_start="2016-09-15", date_end="2016-09-15"),
|
|
"month": request(precision="month", event_kind="education_start"),
|
|
"year": request(
|
|
precision="year",
|
|
event_kind="education_start",
|
|
date_start="2016-01-01",
|
|
date_end="2016-12-31",
|
|
),
|
|
}
|
|
points = {}
|
|
for precision, body in cases.items():
|
|
normalized = normalize_rectification_request(body, today=date(2026, 7, 28))
|
|
built = build_event_contribution_matrix(normalized, row_provider=rows)
|
|
points[precision] = built["matrix"][EVENT_ID]["05:13"]["points"]
|
|
|
|
self.assertEqual(points, {"day": 10, "month": 8, "year": 5})
|
|
|
|
def test_calculation_spec_hash_matches_typescript_for_integral_timezone(self):
|
|
normalized = normalize_rectification_request(request(), today=date(2026, 7, 28))
|
|
self.assertEqual(
|
|
sha256(calculation_spec(normalized)),
|
|
"f05fe0f56ef9ba2b18ec3c6c54f1649f06f1ae5a926491a5c5f676d718d92865",
|
|
)
|
|
|
|
def test_cross_midnight_range_preserves_next_day_datetimes_and_typescript_hash(self):
|
|
normalized = normalize_rectification_request(
|
|
request(start_time="23:00", end_time="03:59"),
|
|
today=date(2026, 7, 28),
|
|
)
|
|
|
|
candidates = _candidate_datetimes(normalized)
|
|
|
|
self.assertEqual(len(candidates), 300)
|
|
self.assertEqual(candidates[0].isoformat(), "1997-08-08T23:00:00")
|
|
self.assertEqual(candidates[60].isoformat(), "1997-08-09T00:00:00")
|
|
self.assertEqual(candidates[-1].isoformat(), "1997-08-09T03:59:00")
|
|
self.assertEqual(
|
|
sha256(calculation_spec(normalized)),
|
|
"b0d5c5ec7f56edbfa2b2e1041b4aa3b648c6cb0681f3f502c7b7910c2894b205",
|
|
)
|
|
|
|
def test_candidate_range_boundaries_stay_bounded_and_equal_is_one_minute(self):
|
|
full_day = normalize_rectification_request(
|
|
request(start_time="00:00", end_time="23:59"),
|
|
today=date(2026, 7, 28),
|
|
)
|
|
equal = normalize_rectification_request(
|
|
request(start_time="05:13", end_time="05:13"),
|
|
today=date(2026, 7, 28),
|
|
)
|
|
|
|
self.assertEqual(len(_candidate_datetimes(full_day)), 1_440)
|
|
self.assertEqual(len(_candidate_datetimes(equal)), 1)
|
|
with self.assertRaisesRegex(ValueError, "end_time must be HH:MM"):
|
|
normalize_rectification_request(
|
|
request(start_time="00:00", end_time="24:00"),
|
|
today=date(2026, 7, 28),
|
|
)
|
|
|
|
def test_daytime_candidate_range_remains_on_birth_date(self):
|
|
normalized = normalize_rectification_request(request(), today=date(2026, 7, 28))
|
|
|
|
candidates = _candidate_datetimes(normalized)
|
|
|
|
self.assertEqual([value.isoformat() for value in candidates], [
|
|
"1997-08-08T05:13:00",
|
|
"1997-08-08T05:14:00",
|
|
"1997-08-08T05:15:00",
|
|
])
|
|
|
|
def test_shared_validator_retains_background_events_and_rejects_non_self_health(self):
|
|
background = normalize_rectification_request(
|
|
request(domain="family", event_kind="family_event"), today=date(2026, 7, 28)
|
|
)
|
|
self.assertEqual(background["events"][0]["event_kind"], "family_event")
|
|
with self.assertRaisesRegex(ValueError, "event_kind does not match domain"):
|
|
normalize_rectification_request(
|
|
request(domain="family", event_kind="family_bereavement"), today=date(2026, 7, 28)
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "event_kind does not match domain"):
|
|
normalize_rectification_request(request(domain="health_pressure", event_kind="family_health_event"), today=date(2026, 7, 28))
|
|
normalized = normalize_rectification_request(request(domain="health_pressure", event_kind="self_health_event"), today=date(2026, 7, 28))
|
|
self.assertEqual(normalized["events"][0]["event_kind"], "self_health_event")
|
|
|
|
def test_relationship_end_is_native_and_uses_pressure_semantics(self):
|
|
normalized = normalize_rectification_request(
|
|
request(domain="relationship", event_kind="relationship_end", precision="day"),
|
|
today=date(2026, 7, 28),
|
|
)
|
|
|
|
def rows(value):
|
|
event_value = value["events"][0]
|
|
return [{
|
|
"time": "05:13", "score": 10,
|
|
"evidence": [{
|
|
"event_id": EVENT_ID, "domain": "relationship", "candidate_time": "05:13",
|
|
"rule_ids": ["controlled_transit_saturn_domain_house", f"event_kind:{event_value['event_kind']}"],
|
|
"points": 10,
|
|
}],
|
|
"missing_layers": [],
|
|
}]
|
|
|
|
built = build_event_contribution_matrix(normalized, row_provider=rows)
|
|
contribution = built["matrix"][EVENT_ID]["05:13"]
|
|
self.assertIn("event_kind_profile:relationship_end:pressure", contribution["rule_ids"])
|
|
self.assertGreater(contribution["points"], 10)
|
|
|
|
def test_relationship_start_and_change_use_distinct_rule_conditioned_profiles(self):
|
|
def relationship_rows(_value):
|
|
return [
|
|
{
|
|
"time": "05:13", "score": 10,
|
|
"evidence": [{
|
|
"event_id": EVENT_ID, "domain": "relationship", "candidate_time": "05:13",
|
|
"rule_ids": ["vim_md_domain_house", "vim_ad_functional_benefic_auxiliary"],
|
|
"points": 10,
|
|
}],
|
|
"missing_layers": [],
|
|
},
|
|
{
|
|
"time": "05:14", "score": 9,
|
|
"evidence": [{
|
|
"event_id": EVENT_ID, "domain": "relationship", "candidate_time": "05:14",
|
|
"rule_ids": ["controlled_transit_saturn_domain_house", "vim_ad_functional_malefic_auxiliary"],
|
|
"points": 9,
|
|
}],
|
|
"missing_layers": [],
|
|
},
|
|
]
|
|
|
|
start = normalize_rectification_request(
|
|
request(domain="relationship", event_kind="relationship_start"), today=date(2026, 7, 28)
|
|
)
|
|
change = normalize_rectification_request(
|
|
request(domain="relationship", event_kind="relationship_change"), today=date(2026, 7, 28)
|
|
)
|
|
|
|
start_matrix = build_event_contribution_matrix(start, row_provider=relationship_rows)
|
|
change_matrix = build_event_contribution_matrix(change, row_provider=relationship_rows)
|
|
|
|
self.assertGreater(start_matrix["matrix"][EVENT_ID]["05:13"]["points"], start_matrix["matrix"][EVENT_ID]["05:14"]["points"] )
|
|
self.assertGreater(change_matrix["matrix"][EVENT_ID]["05:14"]["points"], change_matrix["matrix"][EVENT_ID]["05:13"]["points"] )
|
|
self.assertEqual(change_matrix["date_sensitivity"][0]["sample_winners"], ["05:14", "05:14", "05:14"])
|
|
self.assertNotIn("event_kind_profile", change_matrix["matrix"][EVENT_ID]["05:14"]["technique_layers"])
|
|
|
|
def test_relationship_kind_profile_does_not_create_points_without_activation(self):
|
|
normalized = normalize_rectification_request(
|
|
request(domain="relationship", event_kind="relationship_change", precision="day"),
|
|
today=date(2026, 7, 28),
|
|
)
|
|
|
|
def rows(_value):
|
|
return [{
|
|
"time": "05:13", "score": 0,
|
|
"evidence": [{
|
|
"event_id": EVENT_ID, "domain": "relationship", "candidate_time": "05:13",
|
|
"rule_ids": ["no_domain_activation", "event_kind:relationship_change"],
|
|
"points": 0,
|
|
}],
|
|
"missing_layers": [],
|
|
}]
|
|
|
|
built = build_event_contribution_matrix(normalized, row_provider=rows)
|
|
self.assertEqual(built["matrix"][EVENT_ID]["05:13"]["points"], 0)
|
|
|
|
def test_date_sampling_preserves_declared_range_and_uses_bounded_samples(self):
|
|
base = request()["events"][0]
|
|
self.assertEqual(sample_event_dates({**base, "precision": "month"}), ["2016-09-01", "2016-09-15", "2016-09-30"])
|
|
year = {**base, "precision": "year", "date_start": "2016-01-01", "date_end": "2016-12-31"}
|
|
self.assertEqual(len(sample_event_dates(year)), 12)
|
|
ranged = {**base, "precision": "range", "date_start": "2015-01-01", "date_end": "2016-12-31"}
|
|
self.assertLessEqual(len(sample_event_dates(ranged)), 12)
|
|
|
|
def test_contribution_matrix_and_leave_out_diagnostics_use_matrix_math(self):
|
|
normalized = normalize_rectification_request(request(), today=date(2026, 7, 28))
|
|
|
|
def rows(value):
|
|
sampled = value["events"][0]["date"]
|
|
shift = {"2016-09-01": 0, "2016-09-15": 1, "2016-09-30": 2}[sampled]
|
|
return [{
|
|
"time": candidate,
|
|
"score": points + shift,
|
|
"evidence": [{
|
|
"event_id": EVENT_ID,
|
|
"domain": "education",
|
|
"candidate_time": candidate,
|
|
"rule_ids": ["D24:test"],
|
|
"points": points + shift,
|
|
}],
|
|
"missing_layers": ["KP_cusps"],
|
|
} for candidate, points in [("05:13", 9), ("05:14", 10), ("05:15", 8)]]
|
|
|
|
built = build_event_contribution_matrix(normalized, row_provider=rows)
|
|
scored = score_from_matrix(normalized, built)
|
|
self.assertEqual(built["matrix"][EVENT_ID]["05:14"]["points"], 8.8)
|
|
self.assertEqual(scored[1]["score"], 8.8)
|
|
self.assertEqual(built["missing_layers"], ["KP_cusps"])
|
|
|
|
def test_formal_score_and_diagnostics_endpoints_share_the_service_bundle(self):
|
|
normalized = normalize_rectification_request(request(), today=date(2026, 7, 28))
|
|
built = {
|
|
"candidate_times": ["05:13", "05:14"],
|
|
"matrix": {EVENT_ID: {
|
|
"05:13": {"points": 10, "rule_ids": ["D24:a"], "technique_layers": ["D24"]},
|
|
"05:14": {"points": 8, "rule_ids": ["D24:b"], "technique_layers": ["D24"]},
|
|
}},
|
|
"date_sensitivity": [{
|
|
"event_id": EVENT_ID,
|
|
"declared_date_range": {"start": "2016-09-01", "end": "2016-09-30", "precision": "month"},
|
|
"sample_dates": ["2016-09-01", "2016-09-15", "2016-09-30"],
|
|
"winner_retention_rate": 1,
|
|
"score_variance": 1,
|
|
"sample_winners": ["05:13", "05:13", "05:13"],
|
|
}],
|
|
"missing_layers": ["KP_cusps"],
|
|
"static_contexts": [{"feature": {"time": "05:13"}}, {"feature": {"time": "05:14"}}],
|
|
}
|
|
feature = {
|
|
"calculation_spec_hash": "0" * 64,
|
|
"algorithm_version": "rectification-v5-matrix-scoring-2",
|
|
"candidate_count": 2,
|
|
"feature_hash": "1" * 64,
|
|
"features": [{"time": "05:13"}, {"time": "05:14"}],
|
|
}
|
|
with patch("scripts.rectification.api_service.build_event_contribution_matrix", return_value=built), patch(
|
|
"scripts.rectification.api_service.build_candidate_feature_snapshot", return_value=feature
|
|
):
|
|
scored = score_candidates(normalized)
|
|
diagnostic_result = diagnostics(normalized)
|
|
self.assertFalse(scored["can_confirm_exact_minute"])
|
|
self.assertIn("event_contribution_matrix", scored)
|
|
self.assertEqual(diagnostic_result["diagnostics"]["leave_one_event_out_retention_rate"], 1)
|
|
self.assertFalse(diagnostic_result["can_confirm_exact_minute"])
|
|
|
|
def test_score_endpoint_returns_candidate_decision_receipt_v2_and_real_execution_ledger(self):
|
|
body = request()
|
|
body["events"] = [
|
|
event(1, "education", "education_start"),
|
|
event(2, "career", "promotion", precision="month"),
|
|
event(3, "finance", "finance_gain"),
|
|
event(4, "family", "family_event", precision="year"),
|
|
]
|
|
normalized = normalize_rectification_request(body, today=date(2026, 7, 28))
|
|
scored_ids = [item["id"] for item in normalized["events"][:3]]
|
|
built = {
|
|
"candidate_times": ["05:13", "05:14", "05:15"],
|
|
"matrix": {
|
|
event_id: {
|
|
"05:13": {"points": 4, "rule_ids": ["vim_md_domain_house"], "technique_layers": ["vim_md_domain_house"]},
|
|
"05:14": {"points": 2, "rule_ids": ["vim_md_domain_house"], "technique_layers": ["vim_md_domain_house"]},
|
|
"05:15": {"points": 1, "rule_ids": ["vim_md_domain_house"], "technique_layers": ["vim_md_domain_house"]},
|
|
}
|
|
for event_id in scored_ids
|
|
},
|
|
"date_sensitivity": [
|
|
{
|
|
"event_id": event_id,
|
|
"declared_date_range": {"start": "2016-09-15", "end": "2016-09-15", "precision": "day"},
|
|
"sample_dates": ["2016-09-15"],
|
|
"winner_retention_rate": 1,
|
|
"score_variance": 0,
|
|
"sample_winners": ["05:13"],
|
|
}
|
|
for event_id in scored_ids
|
|
],
|
|
"missing_layers": [],
|
|
"static_contexts": [],
|
|
}
|
|
diagnostic_values = {
|
|
"primary_cluster_retention_rate": 1,
|
|
"leave_one_event_out_retention_rate": .9,
|
|
"leave_one_domain_out_retention_rate": .9,
|
|
"date_sensitivity_retention_rate": .9,
|
|
"neighbor_support_minutes": 1,
|
|
"primary_secondary_margin_percent": 30,
|
|
"event_date_sensitivity": built["date_sensitivity"],
|
|
}
|
|
feature = {
|
|
"calculation_spec_hash": "0" * 64,
|
|
"algorithm_version": "rectification-v5-matrix-scoring-2",
|
|
"candidate_count": 3,
|
|
"feature_hash": "1" * 64,
|
|
"features": [],
|
|
}
|
|
with patch("scripts.rectification.api_service.build_event_contribution_matrix", return_value=built), patch(
|
|
"scripts.rectification.api_service.build_candidate_feature_snapshot", return_value=feature
|
|
), patch("scripts.rectification.api_service.run_diagnostics", return_value=diagnostic_values):
|
|
first = score_candidates(normalized)
|
|
second = score_candidates(normalized)
|
|
|
|
self.assertEqual(first["event_contract_version"], "rectification-event-contract-v2")
|
|
self.assertEqual(first["candidate_decisions"], second["candidate_decisions"])
|
|
self.assertEqual(sum(item["relative_support"] for item in first["candidate_decisions"]), 100)
|
|
for rank, candidate in enumerate(first["candidate_decisions"], start=1):
|
|
self.assertEqual(set(candidate), {"candidate_id", "rank", "time", "relative_support", "tied_minute_count"})
|
|
self.assertEqual(candidate["rank"], rank)
|
|
UUID(candidate["candidate_id"])
|
|
|
|
receipt = first["candidate_decision_receipt"]
|
|
self.assertEqual(receipt["receipt_version"], "candidate-decision-receipt-v2")
|
|
self.assertTrue(receipt["selection_allowed"])
|
|
self.assertTrue(receipt["acceptance_allowed"])
|
|
self.assertFalse(receipt["confirmation_allowed"])
|
|
self.assertTrue(receipt["gates"]["domain_diversity"]["passed"])
|
|
self.assertTrue(receipt["gates"]["date_quality"]["passed"])
|
|
self.assertTrue(receipt["gates"]["exact_confirmation"]["fail_closed"])
|
|
|
|
self.assertEqual(first["decision_receipt"], receipt)
|
|
self.assertEqual(first["decision_policy_version"], "rectification-candidate-policy-v2")
|
|
self.assertTrue(receipt["display_allowed"])
|
|
self.assertTrue(receipt["accept_allowed"])
|
|
self.assertFalse(receipt["confirm_allowed"])
|
|
self.assertIsNotNone(receipt["representative_candidate_id"])
|
|
self.assertEqual(receipt["representative_time"], "05:13")
|
|
|
|
entries = first["execution_ledger"]
|
|
background = next(item for item in entries if item.get("event_id") == normalized["events"][3]["id"])
|
|
self.assertEqual(background["status"], "retained_not_scored")
|
|
executed = next(item for item in entries if item.get("event_id") == normalized["events"][0]["id"])
|
|
self.assertEqual(executed["technique_layers"], ["vim_md_domain_house"])
|
|
|
|
def test_diagnostics_endpoint_mirrors_candidate_decision_policy_fields(self):
|
|
normalized = normalize_rectification_request(request(), today=date(2026, 7, 28))
|
|
scored = {
|
|
"result_id": "00000000-0000-4000-8000-000000000099",
|
|
"algorithm_version": "rectification-v5-matrix-scoring-2",
|
|
"event_contract_version": "rectification-event-contract-v2",
|
|
"decision_policy_version": "rectification-candidate-policy-v2",
|
|
"calculation_spec_hash": "0" * 64,
|
|
"candidate_decisions": [],
|
|
"candidate_decision_receipt": {},
|
|
"decision_receipt": {},
|
|
"execution_ledger_version": "rectification-execution-ledger-v2",
|
|
"execution_ledger": [],
|
|
"diagnostics": {},
|
|
"missing_layers": [],
|
|
"display_allowed": False,
|
|
"selection_allowed": False,
|
|
"acceptance_allowed": False,
|
|
"confirmation_allowed": False,
|
|
"representative_candidate_id": None,
|
|
"representative_time": None,
|
|
"overall_confidence": "low",
|
|
"margin_percent": 0.0,
|
|
}
|
|
with patch("scripts.rectification.api_service.score_candidates", return_value=scored):
|
|
result = diagnostics(normalized)
|
|
|
|
for field in (
|
|
"display_allowed",
|
|
"selection_allowed",
|
|
"acceptance_allowed",
|
|
"confirmation_allowed",
|
|
"representative_candidate_id",
|
|
"representative_time",
|
|
"overall_confidence",
|
|
"margin_percent",
|
|
):
|
|
self.assertEqual(result[field], scored[field])
|
|
self.assertFalse(result["can_confirm_exact_minute"])
|
|
|
|
def test_single_domain_or_quantized_top_tie_blocks_candidate_acceptance(self):
|
|
body = request()
|
|
body["events"] = [
|
|
event(1, "career", "career_entry"),
|
|
event(2, "career", "promotion"),
|
|
event(3, "career", "career_change"),
|
|
]
|
|
normalized = normalize_rectification_request(body, today=date(2026, 7, 28))
|
|
built = {
|
|
"candidate_times": ["05:13", "05:14"],
|
|
"matrix": {
|
|
item["id"]: {
|
|
"05:13": {"points": 3.333346, "rule_ids": ["D10:test"], "technique_layers": ["D10"]},
|
|
"05:14": {"points": 3.333333, "rule_ids": ["D10:test"], "technique_layers": ["D10"]},
|
|
}
|
|
for item in normalized["events"]
|
|
},
|
|
"date_sensitivity": [],
|
|
"missing_layers": [],
|
|
"static_contexts": [],
|
|
}
|
|
diagnostic_values = {
|
|
"primary_cluster_retention_rate": 1,
|
|
"leave_one_event_out_retention_rate": 1,
|
|
"leave_one_domain_out_retention_rate": 1,
|
|
"date_sensitivity_retention_rate": 1,
|
|
"neighbor_support_minutes": 1,
|
|
"primary_secondary_margin_percent": 50,
|
|
}
|
|
with patch("scripts.rectification.api_service.build_event_contribution_matrix", return_value=built), patch(
|
|
"scripts.rectification.api_service.build_candidate_feature_snapshot", return_value={}
|
|
), patch("scripts.rectification.api_service.run_diagnostics", return_value=diagnostic_values):
|
|
result = score_candidates(normalized)
|
|
|
|
receipt = result["candidate_decision_receipt"]
|
|
self.assertFalse(receipt["acceptance_allowed"])
|
|
self.assertIn("insufficient_domain_diversity", receipt["reasons"])
|
|
self.assertIn("tied_top_score", receipt["reasons"])
|
|
self.assertEqual(result["candidate_decisions"][0]["tied_minute_count"], 2)
|
|
self.assertEqual(receipt["tie_policy"]["score_quantum"], .0001)
|
|
self.assertEqual(receipt["tie_policy"]["absolute_tolerance"], .0001)
|
|
|
|
def test_low_date_quality_blocks_acceptance_even_with_multiple_domains_and_a_unique_candidate(self):
|
|
body = request()
|
|
body["events"] = [
|
|
event(1, "education", "education_start", precision="year"),
|
|
event(2, "career", "promotion", precision="year"),
|
|
event(3, "finance", "finance_gain", precision="year"),
|
|
]
|
|
normalized = normalize_rectification_request(body, today=date(2026, 7, 28))
|
|
built = {
|
|
"candidate_times": ["05:13", "05:14"],
|
|
"matrix": {
|
|
item["id"]: {
|
|
"05:13": {"points": 4, "rule_ids": ["test"], "technique_layers": ["test"]},
|
|
"05:14": {"points": 1, "rule_ids": ["test"], "technique_layers": ["test"]},
|
|
}
|
|
for item in normalized["events"]
|
|
},
|
|
"date_sensitivity": [],
|
|
"missing_layers": [],
|
|
"static_contexts": [],
|
|
}
|
|
diagnostic_values = {
|
|
"leave_one_event_out_retention_rate": 1,
|
|
"leave_one_domain_out_retention_rate": 1,
|
|
"date_sensitivity_retention_rate": 1,
|
|
"primary_secondary_margin_percent": 50,
|
|
}
|
|
with patch("scripts.rectification.api_service.build_event_contribution_matrix", return_value=built), patch(
|
|
"scripts.rectification.api_service.build_candidate_feature_snapshot", return_value={}
|
|
), patch("scripts.rectification.api_service.run_diagnostics", return_value=diagnostic_values):
|
|
result = score_candidates(normalized)
|
|
|
|
receipt = result["candidate_decision_receipt"]
|
|
self.assertTrue(result["candidate_decisions"])
|
|
self.assertFalse(receipt["acceptance_allowed"])
|
|
self.assertFalse(receipt["gates"]["date_quality"]["passed"])
|
|
self.assertIn("low_date_quality", receipt["reasons"])
|
|
self.assertFalse(result["can_confirm_exact_minute"])
|
|
|
|
def test_http_registry_exposes_all_v5_endpoints(self):
|
|
expected = {
|
|
"rectification-v5-candidate-features": "/api/rectification/v5/candidate-features",
|
|
"rectification-v5-score": "/api/rectification/v5/score",
|
|
"rectification-v5-diagnostics": "/api/rectification/v5/diagnostics",
|
|
}
|
|
for command, endpoint in expected.items():
|
|
self.assertEqual(API_COMMAND_MAP[command], endpoint)
|
|
self.assertIn(endpoint, TECHNIQUE_EXAMPLE_ENDPOINTS)
|
|
|
|
def test_http_handler_enforces_subject_and_event_kind_boundaries(self):
|
|
handler = object.__new__(JyotishAPIHandler)
|
|
retained = handler._rectification_v5_request(request(domain="family", event_kind="family_event"))
|
|
self.assertEqual(retained["events"][0]["event_kind"], "family_event")
|
|
with self.assertRaisesRegex(BadRequest, "event_kind does not match domain"):
|
|
handler._rectification_v5_request(request(domain="family", event_kind="family_bereavement"))
|
|
with self.assertRaisesRegex(BadRequest, "event_kind does not match domain"):
|
|
handler._rectification_v5_request(request(domain="health_pressure", event_kind="family_health_event"))
|
|
normalized = handler._rectification_v5_request(
|
|
request(domain="health_pressure", event_kind="self_health_event")
|
|
)
|
|
self.assertEqual(normalized["events"][0]["event_kind"], "self_health_event")
|
|
|
|
def test_v4_compatibility_and_v5_score_handlers_share_the_v5_service(self):
|
|
handler = object.__new__(JyotishAPIHandler)
|
|
result = {"result_id": "00000000-0000-4000-8000-000000000099", "can_confirm_exact_minute": False}
|
|
with patch("scripts.rectification.api_service.score_candidates", return_value=result) as scorer:
|
|
v5 = handler._compute_rectification_v5_score(request())
|
|
v4 = handler._compute_active_rectification_events_v4(request())
|
|
self.assertEqual(scorer.call_count, 2)
|
|
self.assertEqual(v5["endpoint"], "rectification_v5_score")
|
|
self.assertEqual(v4["endpoint"], "active_rectification_events_v4")
|
|
self.assertEqual(v5["result_id"], v4["result_id"])
|
|
self.assertFalse(v5["can_confirm_exact_minute"])
|
|
self.assertFalse(v4["can_confirm_exact_minute"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|