Jesse_Chen
350 lines
13 KiB
Python
350 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date, datetime
|
|
|
|
import pytest
|
|
|
|
from scripts import dynamic_rectification
|
|
from scripts import dynamic_rectification_opportunities
|
|
|
|
|
|
def _base_request() -> dict:
|
|
return {
|
|
"case_id": "case-1",
|
|
"birth_date": "1990-01-01",
|
|
"as_of_date": "2026-07-18",
|
|
"start_time": "05:30",
|
|
"end_time": "05:33",
|
|
"lat": 31.23,
|
|
"lon": 121.47,
|
|
"tz": 8.0,
|
|
"evidence": [],
|
|
"dismissed_opportunity_ids": [],
|
|
"question_fingerprints": [],
|
|
"partition_fingerprints": [],
|
|
"recent_ranges": [],
|
|
"events": [],
|
|
}
|
|
|
|
|
|
def _fake_rows(_request: dict) -> list[dict]:
|
|
return [
|
|
{
|
|
"window_group": "periods-3",
|
|
"dimension_code": "career",
|
|
"window_start": "2014-01-01",
|
|
"window_end": "2017-12-31",
|
|
"activations": {"05:30": 5.0, "05:31": 1.0, "05:32": 0.0, "05:33": 0.0},
|
|
"missing_layers": [],
|
|
"fact_selection_priority": 0.0,
|
|
"fact_priority_version": "birth-time-question-fact-priority-v1",
|
|
"event_fact_selection_priority": 0.0,
|
|
"event_fact_priority_version": "birth-time-question-event-fact-priority-v1",
|
|
},
|
|
{
|
|
"window_group": "periods-3",
|
|
"dimension_code": "career",
|
|
"window_start": "2018-01-01",
|
|
"window_end": "2021-12-31",
|
|
"activations": {"05:30": 0.0, "05:31": 5.0, "05:32": 4.0, "05:33": 0.0},
|
|
"missing_layers": [],
|
|
"fact_selection_priority": 0.0,
|
|
"fact_priority_version": "birth-time-question-fact-priority-v1",
|
|
"event_fact_selection_priority": 0.0,
|
|
"event_fact_priority_version": "birth-time-question-event-fact-priority-v1",
|
|
},
|
|
{
|
|
"window_group": "periods-3",
|
|
"dimension_code": "career",
|
|
"window_start": "2022-01-01",
|
|
"window_end": "2026-07-18",
|
|
"activations": {"05:30": 0.0, "05:31": 0.0, "05:32": 1.0, "05:33": 5.0},
|
|
"missing_layers": [],
|
|
"fact_selection_priority": 0.0,
|
|
"fact_priority_version": "birth-time-question-fact-priority-v1",
|
|
"event_fact_selection_priority": 0.0,
|
|
"event_fact_priority_version": "birth-time-question-event-fact-priority-v1",
|
|
},
|
|
]
|
|
|
|
|
|
def _fake_model() -> dict:
|
|
return {
|
|
"version": "birth-time-choice-scoring-v2",
|
|
"opportunity_model_version": "birth-time-opportunity-model-v4",
|
|
"historical_event_fingerprint": dynamic_rectification_opportunities.historical_event_fingerprint(_base_request()),
|
|
"birth_date": "1990-01-01",
|
|
"as_of_date": "2026-07-18",
|
|
"range": {"start_time": "05:30", "end_time": "05:33"},
|
|
"location": {"lat": 31.23, "lon": 121.47, "tz": 8.0},
|
|
"candidate_times": ["05:30", "05:31", "05:32", "05:33"],
|
|
"windows": _fake_rows({}),
|
|
}
|
|
|
|
|
|
def test_packet_contains_only_candidate_backed_high_gain_opportunities(monkeypatch) -> None:
|
|
monkeypatch.setattr(dynamic_rectification, "_candidate_window_rows", _fake_rows)
|
|
|
|
packet = dynamic_rectification.build_difference_packet(_base_request())
|
|
|
|
assert packet["scoring_version"] == "birth-time-choice-scoring-v2"
|
|
assert packet["current_range"] == {"start_time": "05:30", "end_time": "05:33"}
|
|
assert packet["opportunities"]
|
|
for opportunity in packet["opportunities"]:
|
|
assert opportunity["estimated_information_gain"] >= 0.15
|
|
assert 2 <= len(opportunity["partitions"]) <= 4
|
|
assert len({item["partition_id"] for item in opportunity["partitions"]}) == len(
|
|
opportunity["partitions"]
|
|
)
|
|
for partition in opportunity["partitions"]:
|
|
assert set(partition["candidate_scores"]) == {
|
|
"05:30", "05:31", "05:32", "05:33",
|
|
}
|
|
|
|
|
|
def test_period_range_offers_multiple_distinct_evidence_domains() -> None:
|
|
packet = dynamic_rectification.build_difference_packet({
|
|
**_base_request(),
|
|
"birth_date": "1997-08-09",
|
|
"as_of_date": "2026-07-19",
|
|
"start_time": "04:00",
|
|
"end_time": "07:59",
|
|
"lat": 36.6,
|
|
"lon": 114.5,
|
|
})
|
|
|
|
dimensions = {item["dimension_code"] for item in packet["opportunities"]}
|
|
assert len(dimensions) >= 4
|
|
|
|
|
|
def test_real_historical_event_changes_next_question_order_not_public_gain() -> None:
|
|
request = {
|
|
**_base_request(),
|
|
"birth_date": "1993-04-17",
|
|
"as_of_date": "2026-07-21",
|
|
"start_time": "14:20",
|
|
"end_time": "14:40",
|
|
"lat": 36.683333,
|
|
"lon": 114.35,
|
|
"tz": 8.0,
|
|
}
|
|
without_event = dynamic_rectification.build_difference_packet(request)
|
|
with_event = dynamic_rectification.build_difference_packet({
|
|
**request,
|
|
"events": [{
|
|
"id": "11111111-1111-4111-8111-111111111111",
|
|
"domain": "education",
|
|
"date": "2018",
|
|
"precision": "year",
|
|
}],
|
|
})
|
|
|
|
assert without_event["opportunities"][0]["dimension_code"] == "career"
|
|
assert with_event["opportunities"][0]["dimension_code"] == "education"
|
|
gains_without = {
|
|
item["opportunity_id"]: item["estimated_information_gain"]
|
|
for item in without_event["opportunities"]
|
|
}
|
|
gains_with = {
|
|
item["opportunity_id"]: item["estimated_information_gain"]
|
|
for item in with_event["opportunities"]
|
|
}
|
|
assert gains_with == gains_without
|
|
|
|
|
|
def test_packet_excludes_used_opportunity_and_partition_fingerprints(monkeypatch) -> None:
|
|
monkeypatch.setattr(dynamic_rectification, "_candidate_window_rows", _fake_rows)
|
|
first = dynamic_rectification.build_difference_packet(_base_request())
|
|
used = first["opportunities"][0]
|
|
request = _base_request()
|
|
request["dismissed_opportunity_ids"] = [used["opportunity_id"]]
|
|
request["partition_fingerprints"] = [used["candidate_partition_fingerprint"]]
|
|
|
|
second = dynamic_rectification.build_difference_packet(request)
|
|
|
|
assert all(item["opportunity_id"] != used["opportunity_id"] for item in second["opportunities"])
|
|
assert all(
|
|
item["candidate_partition_fingerprint"] != used["candidate_partition_fingerprint"]
|
|
for item in second["opportunities"]
|
|
)
|
|
|
|
|
|
def test_packet_reuses_the_persisted_candidate_model(monkeypatch) -> None:
|
|
calls: list[dict] = []
|
|
monkeypatch.setattr(
|
|
dynamic_rectification,
|
|
"_compute_candidate_model",
|
|
lambda request: calls.append(request) or _fake_model(),
|
|
)
|
|
first = dynamic_rectification.build_difference_packet(_base_request())
|
|
|
|
second = dynamic_rectification.build_difference_packet(
|
|
{**_base_request(), "candidate_model": first["candidate_model"]}
|
|
)
|
|
|
|
assert len(calls) == 1
|
|
assert second["candidate_model"] == first["candidate_model"]
|
|
|
|
|
|
def test_legacy_v2_candidate_model_is_recomputed_instead_of_blocking_resume(monkeypatch) -> None:
|
|
calls: list[dict] = []
|
|
replacement = _fake_model()
|
|
legacy = {**replacement, "opportunity_model_version": "birth-time-opportunity-model-v2"}
|
|
monkeypatch.setattr(
|
|
dynamic_rectification,
|
|
"_compute_candidate_model",
|
|
lambda request: calls.append(request) or replacement,
|
|
)
|
|
|
|
packet = dynamic_rectification.build_difference_packet({
|
|
**_base_request(), "candidate_model": legacy,
|
|
})
|
|
|
|
assert len(calls) == 1
|
|
assert packet["candidate_model"]["opportunity_model_version"] == "birth-time-opportunity-model-v4"
|
|
|
|
|
|
def test_new_or_corrected_historical_event_recomputes_the_private_candidate_model(monkeypatch) -> None:
|
|
changed_request = {
|
|
**_base_request(),
|
|
"events": [{
|
|
"id": "11111111-1111-4111-8111-111111111111",
|
|
"domain": "career",
|
|
"date": "2020",
|
|
"precision": "year",
|
|
}],
|
|
}
|
|
replacement = {
|
|
**_fake_model(),
|
|
"historical_event_fingerprint": dynamic_rectification_opportunities.historical_event_fingerprint(
|
|
changed_request,
|
|
),
|
|
}
|
|
calls: list[dict] = []
|
|
monkeypatch.setattr(
|
|
dynamic_rectification,
|
|
"_compute_candidate_model",
|
|
lambda request: calls.append(request) or replacement,
|
|
)
|
|
|
|
packet = dynamic_rectification.build_difference_packet({
|
|
**changed_request,
|
|
"candidate_model": _fake_model(),
|
|
})
|
|
|
|
assert len(calls) == 1
|
|
assert packet["candidate_model"]["historical_event_fingerprint"] == replacement[
|
|
"historical_event_fingerprint"
|
|
]
|
|
|
|
|
|
def test_candidate_model_rejects_wrong_range_and_non_finite_activation() -> None:
|
|
model = _fake_model()
|
|
model["range"] = {"start_time": "05:31", "end_time": "05:33"}
|
|
with pytest.raises(ValueError, match="candidate model"):
|
|
dynamic_rectification.build_difference_packet({**_base_request(), "candidate_model": model})
|
|
|
|
model = _fake_model()
|
|
model["windows"][0]["activations"]["05:30"] = float("nan")
|
|
with pytest.raises(ValueError, match="candidate model"):
|
|
dynamic_rectification.build_difference_packet({**_base_request(), "candidate_model": model})
|
|
|
|
|
|
def test_candidate_model_rejects_out_of_bounds_windows_and_boolean_activations() -> None:
|
|
model = _fake_model()
|
|
model["windows"][0]["window_end"] = "2027-01-01"
|
|
with pytest.raises(ValueError, match="candidate model"):
|
|
dynamic_rectification.build_difference_packet({**_base_request(), "candidate_model": model})
|
|
|
|
model = _fake_model()
|
|
model["windows"][0]["activations"]["05:30"] = True
|
|
with pytest.raises(ValueError, match="candidate model"):
|
|
dynamic_rectification.build_difference_packet({**_base_request(), "candidate_model": model})
|
|
|
|
|
|
@pytest.mark.parametrize(("field", "changed"), [("lat", 30.0), ("lon", 120.0), ("tz", 7.0)])
|
|
def test_candidate_model_reuse_rejects_location_or_timezone_change(field, changed) -> None:
|
|
with pytest.raises(ValueError, match="candidate model"):
|
|
dynamic_rectification.build_difference_packet({
|
|
**_base_request(), field: changed, "candidate_model": _fake_model(),
|
|
})
|
|
|
|
|
|
def test_existing_evidence_summary_must_be_effective_partition_evidence(monkeypatch) -> None:
|
|
monkeypatch.setattr(dynamic_rectification, "_candidate_window_rows", _fake_rows)
|
|
|
|
with pytest.raises(ValueError, match="partition evidence"):
|
|
dynamic_rectification.build_difference_packet(
|
|
{**_base_request(), "evidence": [{"kind": "unmatched", "note": "free text"}]}
|
|
)
|
|
|
|
|
|
def test_candidate_charts_are_computed_once_and_missing_layers_are_dimension_scoped(
|
|
monkeypatch,
|
|
) -> None:
|
|
from scripts import active_rectification_event_engine
|
|
|
|
candidates = [datetime(1990, 1, 1, 5, 30), datetime(1990, 1, 1, 5, 31)]
|
|
calls: list[datetime] = []
|
|
monkeypatch.setattr(active_rectification_event_engine, "_candidate_datetimes", lambda _: candidates)
|
|
|
|
def fake_candidate_row(request: dict, candidate: datetime) -> dict:
|
|
calls.append(candidate)
|
|
return {
|
|
"time": candidate.strftime("%H:%M"),
|
|
"score": 0.0,
|
|
"evidence": [
|
|
{
|
|
"event_id": event["id"],
|
|
"domain": event["domain"],
|
|
"candidate_time": candidate.strftime("%H:%M"),
|
|
"rule_ids": ["fixture"],
|
|
"points": 1.0,
|
|
}
|
|
for event in request["events"] if event["domain"] != "career"
|
|
],
|
|
"missing_layers": ["D10"],
|
|
}
|
|
|
|
monkeypatch.setattr(active_rectification_event_engine, "_candidate_row", fake_candidate_row)
|
|
monkeypatch.setattr(
|
|
"scripts.dynamic_rectification_opportunities.build_domain_fact_priorities",
|
|
lambda _: {
|
|
dimension: {"selection_priority": 0.0}
|
|
for dimension in ["education", "relocation", "relationship", "career", "health_pressure"]
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
"scripts.dynamic_rectification_opportunities.build_historical_event_priorities",
|
|
lambda _: {
|
|
dimension: {"selection_priority": 0.0}
|
|
for dimension in ["education", "relocation", "relationship", "career", "health_pressure"]
|
|
},
|
|
)
|
|
rows = dynamic_rectification._candidate_window_rows(_base_request())
|
|
|
|
assert calls == candidates
|
|
assert {tuple(row["missing_layers"]) for row in rows if row["dimension_code"] == "career"} == {("D10",)}
|
|
assert {tuple(row["missing_layers"]) for row in rows if row["dimension_code"] != "career"} == {()}
|
|
assert {row["fact_priority_version"] for row in rows} == {"birth-time-question-fact-priority-v1"}
|
|
assert {row["event_fact_priority_version"] for row in rows} == {
|
|
"birth-time-question-event-fact-priority-v1",
|
|
}
|
|
|
|
|
|
def test_window_edges_and_under_age_cases_do_not_create_invalid_calculation(monkeypatch) -> None:
|
|
assert dynamic_rectification._experience_windows("2000-01-01", "2012-01-01") == [
|
|
(date(2012, 1, 1), date(2012, 1, 1)),
|
|
]
|
|
|
|
from scripts import active_rectification_event_engine
|
|
|
|
monkeypatch.setattr(
|
|
active_rectification_event_engine,
|
|
"_candidate_row",
|
|
lambda *_: pytest.fail("candidate chart should not be computed"),
|
|
)
|
|
assert dynamic_rectification._candidate_window_rows({
|
|
**_base_request(), "birth_date": "2020-01-01",
|
|
}) == []
|