Merge codex/pr2-final-response into staging integration
This commit is contained in:
Jesse_Chen
@@ -0,0 +1,231 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from scripts.consultation_plan_contract import (
|
||||
ConsultationPlanContractError,
|
||||
apply_plan_precision_boundary,
|
||||
validate_consultation_plan_contract,
|
||||
)
|
||||
|
||||
|
||||
def timing_body(**overrides):
|
||||
body = {
|
||||
"plan_version": "consultation-plan-v2",
|
||||
"strict_workflow_route": "timing",
|
||||
"required_layers": ["Vimshottari", "Narayana", "Transit", "Varga", "negative holdout gate"],
|
||||
"claim_boundary": "candidate_day_month_window_only_until_holdout_passes",
|
||||
"plan_depth": "standard",
|
||||
"requested_domains": ["timing"],
|
||||
"timing_horizon": "next_12_months",
|
||||
"precision_boundary": "precise_timing_blocked",
|
||||
"required_evidence_categories": ["natal_foundation", "timing", "validation"],
|
||||
}
|
||||
body.update(overrides)
|
||||
return body
|
||||
|
||||
|
||||
def test_versioned_plan_is_validated_against_server_route_allowlist():
|
||||
contract = validate_consultation_plan_contract(
|
||||
timing_body(),
|
||||
themes=["timing"],
|
||||
route_packet={"question_type": "timing"},
|
||||
)
|
||||
|
||||
assert contract == {
|
||||
"plan_version": "consultation-plan-v2",
|
||||
"strict_workflow_route": "timing",
|
||||
"plan_depth": "standard",
|
||||
"requested_domains": ["timing"],
|
||||
"timing_horizon": "next_12_months",
|
||||
"required_evidence_categories": ["natal_foundation", "timing", "validation"],
|
||||
"required_layers": ["Vimshottari", "Narayana", "Transit", "Varga", "negative holdout gate"],
|
||||
"claim_boundary": "candidate_day_month_window_only_until_holdout_passes",
|
||||
"precision_boundary": "precise_timing_blocked",
|
||||
"enforcement": "server_allowlist_validated",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("override", "message"),
|
||||
[
|
||||
({"plan_version": "consultation-plan-v1"}, "version"),
|
||||
({"strict_workflow_route": "free_script"}, "route"),
|
||||
({"required_layers": ["D1"]}, "required layers"),
|
||||
({"claim_boundary": "exact_date_allowed"}, "claim boundary"),
|
||||
({"requested_domains": ["career"]}, "requested domains"),
|
||||
({"required_evidence_categories": ["natal_foundation"]}, "evidence categories"),
|
||||
({"plan_depth": "unbounded"}, "depth"),
|
||||
({"timing_horizon": None}, "requires a horizon"),
|
||||
({"precision_boundary": "precise_timing_allowed"}, "precision boundary"),
|
||||
],
|
||||
)
|
||||
def test_versioned_plan_rejects_free_form_or_tampered_workflow_control(override, message):
|
||||
with pytest.raises(ConsultationPlanContractError, match=message):
|
||||
validate_consultation_plan_contract(
|
||||
timing_body(**override),
|
||||
themes=["timing"],
|
||||
route_packet={"question_type": "timing"},
|
||||
)
|
||||
|
||||
|
||||
def test_versioned_plan_rejects_theme_and_resolved_route_mismatch():
|
||||
with pytest.raises(ConsultationPlanContractError, match="theme mismatch"):
|
||||
validate_consultation_plan_contract(
|
||||
timing_body(),
|
||||
themes=["marriage"],
|
||||
route_packet={"question_type": "timing"},
|
||||
)
|
||||
with pytest.raises(ConsultationPlanContractError, match="route mismatch"):
|
||||
validate_consultation_plan_contract(
|
||||
timing_body(),
|
||||
themes=["timing"],
|
||||
route_packet={"question_type": "career"},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("route", "required_layers", "claim_boundary", "evidence_categories", "timing_horizon"),
|
||||
[
|
||||
(
|
||||
"health",
|
||||
["D1", "D6", "D8", "6th/8th houses", "Dasha", "non-medical boundary"],
|
||||
"wellbeing_pressure_patterns_not_medical_diagnosis",
|
||||
["natal_foundation", "domain"],
|
||||
None,
|
||||
),
|
||||
(
|
||||
"education",
|
||||
["D1", "D24", "5th/9th houses", "Mercury/Jupiter", "Dasha"],
|
||||
"learning_pattern_and_broad_timing_only",
|
||||
["natal_foundation", "domain"],
|
||||
None,
|
||||
),
|
||||
(
|
||||
"migration",
|
||||
["D1", "D4", "D12", "4th/12th houses", "Dasha", "Narayana"],
|
||||
"migration_and_home_direction_broad_window_only",
|
||||
["natal_foundation", "domain"],
|
||||
None,
|
||||
),
|
||||
(
|
||||
"family",
|
||||
["D1", "D7", "D12", "4th/5th/9th houses", "Dasha"],
|
||||
"family_pattern_not_deterministic_event_claim",
|
||||
["natal_foundation", "domain"],
|
||||
None,
|
||||
),
|
||||
(
|
||||
"annual",
|
||||
["D1", "Annual chart boundary", "Dasha", "Transit", "Tajika candidate"],
|
||||
"annual_report_broad_periods_only",
|
||||
["natal_foundation", "timing", "validation"],
|
||||
"next_12_months",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_expanded_staging_domains_use_canonical_plan_contracts(
|
||||
route, required_layers, claim_boundary, evidence_categories, timing_horizon,
|
||||
):
|
||||
body = {
|
||||
"plan_version": "consultation-plan-v2",
|
||||
"strict_workflow_route": route,
|
||||
"required_layers": required_layers,
|
||||
"claim_boundary": claim_boundary,
|
||||
"plan_depth": "standard",
|
||||
"requested_domains": [route],
|
||||
"timing_horizon": timing_horizon,
|
||||
"precision_boundary": "server_evidence_required",
|
||||
"required_evidence_categories": evidence_categories,
|
||||
}
|
||||
contract = validate_consultation_plan_contract(
|
||||
body,
|
||||
themes=[route],
|
||||
route_packet={"question_type": route},
|
||||
)
|
||||
|
||||
assert contract is not None
|
||||
assert contract["strict_workflow_route"] == route
|
||||
assert contract["requested_domains"] == [route]
|
||||
assert contract["enforcement"] == "server_allowlist_validated"
|
||||
|
||||
|
||||
def test_legacy_research_callers_without_plan_metadata_remain_compatible():
|
||||
assert validate_consultation_plan_contract(
|
||||
{"question": "事业如何"},
|
||||
themes=["career"],
|
||||
route_packet={"question_type": "career"},
|
||||
) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"partial_body",
|
||||
[
|
||||
{"strict_workflow_route": "timing"},
|
||||
{"required_layers": ["Vimshottari"]},
|
||||
{"claim_boundary": "candidate_day_month_window_only_until_holdout_passes"},
|
||||
{"plan_version": "consultation-plan-v2"},
|
||||
],
|
||||
)
|
||||
def test_partial_plan_metadata_cannot_downgrade_to_legacy(partial_body):
|
||||
with pytest.raises(ConsultationPlanContractError, match="incomplete consultation plan metadata"):
|
||||
validate_consultation_plan_contract(
|
||||
partial_body,
|
||||
themes=["career"],
|
||||
route_packet={"question_type": "timing"},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("missing_key", list(timing_body()))
|
||||
def test_versioned_plan_requires_the_complete_metadata_set(missing_key):
|
||||
body = timing_body()
|
||||
body.pop(missing_key)
|
||||
with pytest.raises(ConsultationPlanContractError, match="incomplete consultation plan metadata"):
|
||||
validate_consultation_plan_contract(
|
||||
body,
|
||||
themes=["timing"],
|
||||
route_packet={"question_type": "timing"},
|
||||
)
|
||||
|
||||
|
||||
def test_blocked_plan_can_only_restrict_evidence_owned_precision_policy():
|
||||
context = {
|
||||
"answer_policy": {
|
||||
"can_answer_precise_timing": True,
|
||||
"should_lead_with_limitations": False,
|
||||
}
|
||||
}
|
||||
blocked = apply_plan_precision_boundary(context, {"precision_boundary": "precise_timing_blocked"})
|
||||
evidence_owned = apply_plan_precision_boundary(context, {"precision_boundary": "server_evidence_required"})
|
||||
|
||||
assert blocked["answer_policy"]["can_answer_precise_timing"] is False
|
||||
assert blocked["answer_policy"]["should_lead_with_limitations"] is True
|
||||
assert evidence_owned == context
|
||||
|
||||
|
||||
def test_api_dry_run_exposes_validated_plan_contract_and_rejects_tampering():
|
||||
from scripts.jyotish_api_server import BadRequest, JyotishAPIHandler
|
||||
|
||||
handler = JyotishAPIHandler.__new__(JyotishAPIHandler)
|
||||
body = {
|
||||
**timing_body(),
|
||||
"dry_run": True,
|
||||
"entry_mode": "direct_chart",
|
||||
"question": "应期与阶段问题：未来哪些阶段值得把握？",
|
||||
"theme": ["timing"],
|
||||
"year": 1990,
|
||||
"month": 1,
|
||||
"day": 1,
|
||||
"hour": 12,
|
||||
"minute": 0,
|
||||
"lat": 25,
|
||||
"lon": 121,
|
||||
"tz": 8,
|
||||
}
|
||||
|
||||
result = handler._compute_consultation_workflow(body)
|
||||
assert result["routing"]["question_type"] == "timing"
|
||||
assert result["consultation_plan_contract"]["enforcement"] == "server_allowlist_validated"
|
||||
|
||||
with pytest.raises(BadRequest, match="claim boundary mismatch"):
|
||||
handler._compute_consultation_workflow({**body, "claim_boundary": "exact_date_allowed"})
|
||||
Reference in New Issue
Block a user