154 lines
6.2 KiB
Python
154 lines
6.2 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from lead_scoring.cli.polling_contract import (
|
|
DEFAULT_SCORING_CONFIG,
|
|
canonical_json,
|
|
generate_contract_files,
|
|
run_polling_contract_cli,
|
|
)
|
|
from lead_scoring.schemas.scoring_run import (
|
|
AiMetadataSuccess,
|
|
ScoringRunResponse,
|
|
ScoringRunStatus,
|
|
)
|
|
from lead_scoring.scoring import (
|
|
ScoringMode,
|
|
)
|
|
|
|
|
|
def test_polling_contract_files_are_deterministic_and_validate_against_models():
|
|
files = generate_contract_files()
|
|
|
|
assert files == generate_contract_files()
|
|
assert set(files) == {
|
|
"contract.json",
|
|
"scoring-run-response.schema.json",
|
|
"error-response.schema.json",
|
|
"fixtures/pending.json",
|
|
"fixtures/running.json",
|
|
"fixtures/completed.json",
|
|
"fixtures/failed.json",
|
|
}
|
|
|
|
manifest = json.loads(files["contract.json"])
|
|
assert manifest["api_major_version"] == "v1"
|
|
assert manifest["contract_pack_version"] == "0.1.0"
|
|
assert manifest["endpoint_path"] == "/v1/scoring-runs/{run_id}"
|
|
assert manifest["method"] == "GET"
|
|
assert manifest["headers"]["authorization"] == "Bearer <service-token>"
|
|
assert manifest["headers"]["tenant_id"] == "X-Tenant-Id"
|
|
assert manifest["path_params"]["run_id"]["type"] == "string"
|
|
assert manifest["supported_statuses"] == [
|
|
"pending",
|
|
"running",
|
|
"completed",
|
|
"failed",
|
|
]
|
|
assert manifest["reserved_statuses"] == []
|
|
assert (
|
|
"missing/invalid required headers or path parameters"
|
|
in manifest["http_responses"]["422"]["description"]
|
|
)
|
|
|
|
response_schema = json.loads(files["scoring-run-response.schema.json"])
|
|
assert "status" in response_schema["required"]
|
|
assert "run_id" in response_schema["required"]
|
|
assert "tenant_id" in response_schema["required"]
|
|
assert "subject_ref" in response_schema["required"]
|
|
assert "rule_score" in response_schema["required"]
|
|
assert "ai_score" in response_schema["required"]
|
|
assert "final_score" in response_schema["required"]
|
|
assert "effective_score" in response_schema["required"]
|
|
assert "scoring_mode" in response_schema["required"]
|
|
assert "degraded" in response_schema["required"]
|
|
assert "ai_metadata" in response_schema["required"]
|
|
assert "confidence" in response_schema["required"]
|
|
assert "confidence_threshold" in response_schema["required"]
|
|
assert "effective_config" in response_schema["required"]
|
|
assert "error_code" in response_schema["required"]
|
|
assert "retryable" in response_schema["required"]
|
|
assert "version" in response_schema["required"]
|
|
assert "started_at" in response_schema["required"]
|
|
assert "completed_at" in response_schema["required"]
|
|
assert "created_at" in response_schema["required"]
|
|
assert "updated_at" in response_schema["required"]
|
|
|
|
error_schema = json.loads(files["error-response.schema.json"])
|
|
assert "error_code" in error_schema["required"]
|
|
assert "detail" in error_schema["required"]
|
|
|
|
# Public polling exposes the selected version, not internal prompt/config contents.
|
|
for name in ["pending", "running", "completed", "failed"]:
|
|
raw = json.loads(files[f"fixtures/{name}.json"])
|
|
res = ScoringRunResponse.model_validate(raw)
|
|
assert res.effective_config is None
|
|
assert res.confidence_threshold == 0.80
|
|
|
|
pending = ScoringRunResponse.model_validate(json.loads(files["fixtures/pending.json"]))
|
|
assert pending.status is ScoringRunStatus.PENDING
|
|
assert pending.scoring_mode is None
|
|
assert pending.ai_metadata is None
|
|
assert pending.confidence is None
|
|
assert pending.started_at is None
|
|
assert pending.completed_at is None
|
|
|
|
running = ScoringRunResponse.model_validate(json.loads(files["fixtures/running.json"]))
|
|
assert running.status is ScoringRunStatus.RUNNING
|
|
assert running.scoring_mode is None
|
|
assert running.ai_metadata is None
|
|
assert running.confidence is None
|
|
assert running.started_at is not None
|
|
assert running.completed_at is None
|
|
|
|
completed = ScoringRunResponse.model_validate(json.loads(files["fixtures/completed.json"]))
|
|
assert completed.status is ScoringRunStatus.COMPLETED
|
|
assert completed.scoring_mode is ScoringMode.MEDIUM_AI
|
|
assert completed.rule_score == 82
|
|
assert completed.ai_score == 78
|
|
assert completed.final_score == 80
|
|
assert completed.confidence == 0.91
|
|
assert completed.degraded is False
|
|
assert isinstance(completed.ai_metadata, AiMetadataSuccess)
|
|
assert completed.ai_metadata.model == DEFAULT_SCORING_CONFIG.ai_model
|
|
assert completed.ai_metadata.effort_profile is DEFAULT_SCORING_CONFIG.effort_profile
|
|
assert completed.ai_metadata.pass_count == 2
|
|
|
|
failed = ScoringRunResponse.model_validate(json.loads(files["fixtures/failed.json"]))
|
|
assert failed.status is ScoringRunStatus.FAILED
|
|
assert failed.scoring_mode is None
|
|
assert failed.ai_metadata is None
|
|
assert failed.error_code == "worker_interrupted"
|
|
assert failed.retryable is True
|
|
|
|
|
|
def test_scoring_run_response_forbids_extra_fields():
|
|
files = generate_contract_files()
|
|
raw = json.loads(files["fixtures/completed.json"])
|
|
raw["unexpected_extra_field"] = "bad"
|
|
with pytest.raises(ValidationError):
|
|
ScoringRunResponse.model_validate(raw)
|
|
|
|
|
|
def test_polling_contract_cli_generates_and_checks_contract_pack(tmp_path: Path):
|
|
output_dir = tmp_path / "polling-contract"
|
|
|
|
assert run_polling_contract_cli(["--output", str(output_dir)]) == 0
|
|
assert run_polling_contract_cli(["--output", str(output_dir), "--check"]) == 0
|
|
|
|
|
|
def test_polling_contract_cli_rejects_stale_or_unexpected_artifacts(tmp_path: Path):
|
|
output_dir = tmp_path / "polling-contract"
|
|
assert run_polling_contract_cli(["--output", str(output_dir)]) == 0
|
|
|
|
fixture = output_dir / "fixtures" / "completed.json"
|
|
fixture.write_text(canonical_json({"stale": True}), encoding="utf-8")
|
|
assert run_polling_contract_cli(["--output", str(output_dir), "--check"]) == 1
|
|
|
|
assert run_polling_contract_cli(["--output", str(output_dir)]) == 0
|
|
(output_dir / "unexpected.json").write_text("{}\n", encoding="utf-8")
|
|
assert run_polling_contract_cli(["--output", str(output_dir), "--check"]) == 1
|