Jyotisha/tests/test_dynamic_rectification_scoring.py

from __future__ import annotations

from types import SimpleNamespace
from uuid import uuid4

import pytest

from scripts import dynamic_rectification
from scripts import jyotish_api_server as api_server


def _score_request() -> dict:
    return {
        "birth_date": "1990-01-01",
        "start_time": "05:30",
        "end_time": "05:33",
        "lat": 31.23,
        "lon": 121.47,
        "tz": 8.0,
        "choice_evidence": [],
    }


def _opportunity_request(**changes) -> dict:
    return {
        "case_id": str(uuid4()),
        "birth_date": "1990-01-01",
        "as_of_date": "2026-07-21",
        "start_time": "05:30",
        "end_time": "05:33",
        "lat": 31.23,
        "lon": 121.47,
        "tz": 8.0,
        "candidate_model": None,
        "evidence": [],
        "dismissed_opportunity_ids": [],
        "question_fingerprints": [],
        "partition_fingerprints": [],
        "recent_ranges": [],
        **changes,
    }


def _decisive_rows() -> list[dict]:
    return [
        {"time": "05:30", "score": 20.0},
        {"time": "05:31", "score": 20.0},
        {"time": "05:32", "score": 20.0},
        {"time": "05:33", "score": 10.0},
    ]


def _evidence(**changes) -> dict:
    return {
        "question_id": str(uuid4()),
        "opportunity_id": "career-window",
        "partition_id": "career-2020-2022",
        "dimension_code": "career",
        "candidate_scores": {"05:30": 0.0, "05:31": 1.0, "05:32": 1.0, "05:33": 0.0},
        "information_gain": 0.5,
        **changes,
    }


def _handler() -> api_server.JyotishAPIHandler:
    handler = api_server.JyotishAPIHandler.__new__(api_server.JyotishAPIHandler)
    handler.headers = {}
    return handler


def test_dynamic_routes_fail_closed_and_compare_wrong_bearers_in_constant_time(
    monkeypatch,
) -> None:
    handler = _handler()
    monkeypatch.delenv("JYOTISH_DYNAMIC_RECTIFICATION_TOKEN", raising=False)
    with pytest.raises(api_server.Forbidden, match="token"):
        handler._compute_dynamic_rectification_score({})

    calls: list[tuple[str, str]] = []
    original = api_server.secrets.compare_digest
    monkeypatch.setattr(
        api_server.secrets,
        "compare_digest",
        lambda supplied, configured: calls.append((supplied, configured))
        or original(supplied, configured),
    )
    monkeypatch.setenv("JYOTISH_DYNAMIC_RECTIFICATION_TOKEN", "server-secret")
    handler.headers = {"Authorization": "Bearer wrong-secret"}
    with pytest.raises(api_server.Forbidden, match="token"):
        handler._compute_dynamic_rectification_opportunities({})
    assert calls == [("wrong-secret", "server-secret")]


def test_unauthenticated_forged_scores_cannot_obtain_an_applicable_result(monkeypatch) -> None:
    monkeypatch.setenv("JYOTISH_DYNAMIC_RECTIFICATION_TOKEN", "server-secret")
    scores = {"05:30": 10_000.0, "05:31": 0.0, "05:32": 0.0, "05:33": 0.0}
    evidence = [
        _evidence(
            question_id=f"question-{index}",
            opportunity_id=f"opportunity-{index}",
            partition_id=f"partition-{index}",
            dimension_code=dimension,
            candidate_scores=scores,
            information_gain=1.0,
        )
        for index, dimension in enumerate(
            ["career", "relationship", "education", "career"], start=1
        )
    ]
    with pytest.raises(api_server.Forbidden, match="token"):
        _handler()._compute_dynamic_rectification_score({
            **_score_request(), "choice_evidence": evidence,
        })


def test_authenticated_dynamic_questionnaire_cannot_open_minute_confirmation(monkeypatch) -> None:
    monkeypatch.setenv("JYOTISH_DYNAMIC_RECTIFICATION_TOKEN", "server-secret")
    handler = _handler()
    handler.headers = {"Authorization": "Bearer server-secret"}
    scores = {"05:30": 10_000.0, "05:31": 0.0, "05:32": 0.0, "05:33": 0.0}
    evidence = [
        _evidence(
            question_id=f"question-{index}",
            opportunity_id=f"opportunity-{index}",
            partition_id=f"partition-{index}",
            dimension_code=dimension,
            candidate_scores=scores,
            information_gain=1.0,
        )
        for index, dimension in enumerate(
            ["career", "relationship", "education", "relocation"], start=1
        )
    ]

    result = handler._compute_dynamic_rectification_score({
        **_score_request(), "choice_evidence": evidence,
    })

    assert result["confidence"] == "high"
    assert result["can_apply"] is False
    assert "vedastro_validation_required" in result["reasons"]


def test_opportunity_route_defaults_legacy_missing_events_to_empty(monkeypatch) -> None:
    monkeypatch.setenv("JYOTISH_DYNAMIC_RECTIFICATION_TOKEN", "server-secret")
    handler = _handler()
    handler.headers = {"Authorization": "Bearer server-secret"}
    received = []
    monkeypatch.setattr(api_server, "_load_local_module", lambda _name: SimpleNamespace(
        build_difference_packet=lambda request: received.append(request)
        or {"packet": {}, "candidate_model": {}},
    ))

    handler._compute_dynamic_rectification_opportunities(_opportunity_request())

    assert received[0]["events"] == []


def test_opportunity_route_validates_and_forwards_historical_events(monkeypatch) -> None:
    monkeypatch.setenv("JYOTISH_DYNAMIC_RECTIFICATION_TOKEN", "server-secret")
    handler = _handler()
    handler.headers = {"Authorization": "Bearer server-secret"}
    event_id = str(uuid4())
    received = []
    monkeypatch.setattr(api_server, "_load_local_module", lambda _name: SimpleNamespace(
        build_difference_packet=lambda request: received.append(request)
        or {"packet": {}, "candidate_model": {}},
    ))

    handler._compute_dynamic_rectification_opportunities(_opportunity_request(events=[{
        "id": event_id,
        "domain": "career",
        "date": "2020-06",
        "precision": "month",
    }]))

    assert received[0]["events"] == [{
        "id": event_id,
        "domain": "career",
        "date": "2020-06",
        "precision": "month",
    }]
    with pytest.raises(api_server.BadRequest, match="date does not match"):
        handler._compute_dynamic_rectification_opportunities(_opportunity_request(events=[{
            "id": event_id,
            "domain": "career",
            "date": "2020",
            "precision": "month",
        }]))


def test_dynamic_routes_are_not_browser_runnable_technique_examples() -> None:
    endpoints = {
        "/api/dynamic_rectification_opportunities",
        "/api/dynamic_rectification_score",
    }
    assert endpoints.isdisjoint(api_server.TECHNIQUE_EXAMPLE_ENDPOINTS)
    assert endpoints.isdisjoint(api_server.API_COMMAND_MAP.values())
    for endpoint in endpoints:
        with pytest.raises(KeyError):
            _handler()._dispatch_technique_endpoint(endpoint, {})


def test_primary_choice_changes_rankings_and_returns_a_real_range() -> None:
    result = dynamic_rectification.score_choice_evidence(
        {**_score_request(), "choice_evidence": [_evidence()]}
    )

    assert result["effective_answer_count"] == 1
    assert result["winning_segment"] == {
        "start_time": "05:31",
        "end_time": "05:32",
        "representative_time": "05:31",
        "width_minutes": 2,
    }
    assert result["can_apply"] is False
    assert result["evidence"] == []


def test_score_accepts_candidate_membership_independent_of_json_key_order() -> None:
    scores = {"05:33": 0.0, "05:32": 1.0, "05:31": 1.0, "05:30": 0.0}
    result = dynamic_rectification.score_choice_evidence(
        {**_score_request(), "choice_evidence": [_evidence(candidate_scores=scores)]}
    )

    assert result["winning_segment"]["start_time"] == "05:31"


def test_cross_midnight_leaders_form_one_chronological_segment() -> None:
    scores = {"23:58": 0.0, "23:59": 1.0, "00:00": 1.0, "00:01": 0.0}
    evidence = [
        _evidence(
            question_id=f"question-{index}",
            opportunity_id=f"opportunity-{index}",
            partition_id=f"partition-{index}",
            dimension_code=dimension,
            candidate_scores=scores,
            information_gain=1.0,
        )
        for index, dimension in enumerate(
            ["career", "relationship", "education", "career"], start=1
        )
    ]
    result = dynamic_rectification.score_choice_evidence({
        **_score_request(),
        "start_time": "23:58",
        "end_time": "00:01",
        "choice_evidence": evidence,
    })

    assert result["confidence"] == "high"
    assert result["winning_segment"] == {
        "start_time": "23:59",
        "end_time": "00:00",
        "representative_time": "23:59",
        "width_minutes": 2,
    }


def test_opaque_trimmed_question_ids_are_valid_and_duplicates_remain_rejected() -> None:
    evidence = _evidence(question_id="  question-career-window  ")
    result = dynamic_rectification.score_choice_evidence(
        {**_score_request(), "choice_evidence": [evidence]}
    )
    assert result["effective_answer_count"] == 1

    with pytest.raises(ValueError, match="duplicate question"):
        dynamic_rectification.score_choice_evidence({
            **_score_request(),
            "choice_evidence": [
                evidence,
                {**evidence, "question_id": "question-career-window"},
            ],
        })


def test_unknown_and_unmatched_are_never_choice_evidence() -> None:
    with pytest.raises(ValueError, match="partition evidence"):
        dynamic_rectification.score_choice_evidence(
            {**_score_request(), "choice_evidence": [{"kind": "unknown"}]}
        )


def test_high_confidence_requires_versioned_hard_gates() -> None:
    result = dynamic_rectification.adjudicate_choice_rows(
        _decisive_rows(),
        effective_answer_count=4,
        dimension_count=3,
        missing_layers=[],
    )

    assert result["confidence"] == "high"
    assert result["can_apply"] is True
    assert result["winning_segment"]["width_minutes"] <= 5
    assert result["margin_percent"] >= 20
    assert result["algorithm_version"] == "birth-time-choice-scoring-v2"


def test_medium_and_missing_layers_never_allow_application() -> None:
    medium = dynamic_rectification.adjudicate_choice_rows(
        _decisive_rows(), effective_answer_count=3, dimension_count=2, missing_layers=[]
    )
    blocked = dynamic_rectification.adjudicate_choice_rows(
        _decisive_rows(),
        effective_answer_count=4,
        dimension_count=3,
        missing_layers=["D10"],
    )

    assert medium["confidence"] == "medium"
    assert medium["can_apply"] is False
    assert blocked["confidence"] == "low"
    assert blocked["can_apply"] is False


def test_score_rejects_client_fields_duplicates_caps_and_invalid_scores() -> None:
    evidence = _evidence()
    with pytest.raises(ValueError, match="option_id"):
        dynamic_rectification.score_choice_evidence({
            **_score_request(), "choice_evidence": [{**evidence, "option_id": "client"}],
        })
    with pytest.raises(ValueError, match="duplicate question"):
        dynamic_rectification.score_choice_evidence(
            {**_score_request(), "choice_evidence": [evidence, evidence]}
        )
    with pytest.raises(ValueError, match="at most 10"):
        dynamic_rectification.score_choice_evidence({
            **_score_request(),
            "choice_evidence": [
                {**evidence, "question_id": str(uuid4())} for _ in range(11)
            ],
        })
    with pytest.raises(ValueError, match="candidate scores"):
        dynamic_rectification.score_choice_evidence({
            **_score_request(),
            "choice_evidence": [_evidence(candidate_scores={
                **evidence["candidate_scores"], "05:34": 1.0,
            })],
        })
    with pytest.raises(ValueError, match="candidate scores"):
        dynamic_rectification.score_choice_evidence({
            **_score_request(),
            "choice_evidence": [_evidence(candidate_scores={
                **evidence["candidate_scores"], "05:30": -1.0,
            })],
        })
    with pytest.raises(ValueError, match="identifier"):
        dynamic_rectification.score_choice_evidence(
            {**_score_request(), "choice_evidence": [_evidence(partition_id="")]}
        )