mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-11-13 01:48:17 +00:00
feat(detection-core): validate threshold feasibility
This commit is contained in:
Marius Ciepluch
@@ -250,7 +250,7 @@
|
||||
- Reject label exposure, acquisition drift, context substitution, stale readiness, scope mismatch, or prior spend.
|
||||
- Done when each refusal branch fails before feasibility and no digest-only set or lineage claim is accepted.
|
||||
- _Requirements: 12.1, 12.2, 13.1_
|
||||
- [ ] 8.5 Refuse arithmetically impossible threshold configurations
|
||||
- [x] 8.5 Refuse arithmetically impossible threshold configurations
|
||||
- Enumerate reachable counts from label-blind pair, case, semantic-site, and model-call capacities.
|
||||
- Evaluate count, rate, per-100, and Wilson-bound thresholds with the fixed comparator direction for each metric.
|
||||
- Require every grouped threshold to have at least one jointly attainable count assignment.
|
||||
|
||||
@@ -0,0 +1,220 @@
|
||||
"""Label-blind arithmetic feasibility for admission thresholds."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
from openultrasast.core import canonical
|
||||
from openultrasast.core.admission_review import (
|
||||
AdmissionManifestV2,
|
||||
AdmissionThreshold,
|
||||
PopulationCensus,
|
||||
ScopeCensus,
|
||||
validate_reviewed_admission_payloads,
|
||||
)
|
||||
from openultrasast.core.factpack import ReviewIndependencePolicy
|
||||
from openultrasast.core.measurement import (
|
||||
DenominatorPolicy,
|
||||
MeasurementContractError,
|
||||
MetricId,
|
||||
MetricRule,
|
||||
ScopeKey,
|
||||
evaluate_metric,
|
||||
metric_passes,
|
||||
validate_denominator_policy,
|
||||
)
|
||||
|
||||
FEASIBILITY_SCHEMA = "openultrasast.detection-core.threshold-feasibility.v2"
|
||||
_DIGEST = re.compile(r"[0-9a-f]{64}")
|
||||
|
||||
|
||||
class AdmissionFeasibilityError(ValueError):
|
||||
"""An admission threshold has no reachable count assignment."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FeasibleCountAssignment:
|
||||
metric_id: MetricId
|
||||
numerator: int
|
||||
denominator: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ThresholdFeasibility:
|
||||
census_digest: str
|
||||
denominator_policy_digest: str
|
||||
threshold_set_digest: str
|
||||
witnesses: tuple[FeasibleCountAssignment, ...]
|
||||
feasibility_digest: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _Capacities:
|
||||
pairs: int
|
||||
vulnerable: int
|
||||
fixed: int
|
||||
ordinary: int
|
||||
sites: int
|
||||
model_calls: int
|
||||
|
||||
|
||||
def _count(value: object, field: str) -> int:
|
||||
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
||||
raise AdmissionFeasibilityError(f"{field} must be a non-negative integer")
|
||||
return value
|
||||
|
||||
|
||||
def _validate_row(row: object) -> ScopeCensus:
|
||||
if type(row) is not ScopeCensus:
|
||||
raise AdmissionFeasibilityError("census rows must be exact ScopeCensus values")
|
||||
counts = tuple(
|
||||
_count(getattr(row, field), field)
|
||||
for field in (
|
||||
"total_cases",
|
||||
"pair_count",
|
||||
"vulnerable_side_count",
|
||||
"fixed_side_count",
|
||||
"ordinary_case_count",
|
||||
"semantic_site_capacity",
|
||||
"model_call_capacity",
|
||||
)
|
||||
)
|
||||
total, pairs, vulnerable, fixed, ordinary, _, _ = counts
|
||||
if total != vulnerable + fixed + ordinary:
|
||||
raise AdmissionFeasibilityError("census total does not reconcile with aggregate roles")
|
||||
if pairs > min(vulnerable, fixed):
|
||||
raise AdmissionFeasibilityError("census pair count exceeds side capacity")
|
||||
return row
|
||||
|
||||
|
||||
def _capacities(census: PopulationCensus) -> tuple[tuple[ScopeKey, ...], _Capacities]:
|
||||
if type(census) is not PopulationCensus or not census.rows:
|
||||
raise AdmissionFeasibilityError("census must be a nonempty PopulationCensus")
|
||||
if census.case_labels_exposed:
|
||||
raise AdmissionFeasibilityError("feasibility requires a label-blind census")
|
||||
if _DIGEST.fullmatch(census.census_digest) is None:
|
||||
raise AdmissionFeasibilityError("census digest must be a full lowercase SHA-256 digest")
|
||||
rows = tuple(_validate_row(row) for row in census.rows)
|
||||
scopes = tuple(row.scope for row in rows)
|
||||
if scopes != tuple(sorted(scopes)) or len(scopes) != len(set(scopes)):
|
||||
raise AdmissionFeasibilityError("census scopes must be unique and canonically ordered")
|
||||
return scopes, _Capacities(
|
||||
sum(row.pair_count for row in rows),
|
||||
sum(row.vulnerable_side_count for row in rows),
|
||||
sum(row.fixed_side_count for row in rows),
|
||||
sum(row.ordinary_case_count for row in rows),
|
||||
sum(row.semantic_site_capacity for row in rows),
|
||||
sum(row.model_call_capacity for row in rows),
|
||||
)
|
||||
|
||||
|
||||
def _validated_inputs(
|
||||
manifest: AdmissionManifestV2,
|
||||
policy: DenominatorPolicy,
|
||||
review_policy: ReviewIndependencePolicy,
|
||||
resolver: canonical.ContentResolver,
|
||||
) -> tuple[_Capacities, str]:
|
||||
try:
|
||||
validate_reviewed_admission_payloads(manifest.census, manifest, review_policy, resolver)
|
||||
validate_denominator_policy(policy)
|
||||
except (AttributeError, LookupError, TypeError, ValueError) as exc:
|
||||
raise AdmissionFeasibilityError(str(exc)) from exc
|
||||
if policy.policy_digest != manifest.denominator_policy_digest:
|
||||
raise AdmissionFeasibilityError("denominator policy does not match the reviewed manifest")
|
||||
scopes, capacities = _capacities(manifest.census)
|
||||
if any(rule.scopes != scopes for rule in policy.rules):
|
||||
raise AdmissionFeasibilityError("denominator policy scope does not match the census")
|
||||
return capacities, manifest.threshold_set_digest
|
||||
|
||||
|
||||
def _best_counts(metric_id: MetricId, capacities: _Capacities) -> tuple[int, int]:
|
||||
"""Return the jointly optimal reachable point under the fixed comparator."""
|
||||
|
||||
fixed = {
|
||||
MetricId.PAIRED_CORRECTNESS: capacities.pairs,
|
||||
MetricId.VULNERABLE_RECALL: capacities.vulnerable,
|
||||
MetricId.FIXED_FALSE_ALARM_RATE: capacities.fixed,
|
||||
MetricId.ORDINARY_FALSE_ALARMS_PER_100: capacities.ordinary,
|
||||
MetricId.SEMANTIC_STAGE_COVERAGE: capacities.sites,
|
||||
MetricId.MODEL_SUCCESS_RATE: capacities.model_calls,
|
||||
}
|
||||
if metric_id is MetricId.PRECISION:
|
||||
if capacities.vulnerable:
|
||||
return capacities.vulnerable, capacities.vulnerable
|
||||
false_positive_capacity = capacities.fixed + capacities.ordinary
|
||||
return (0, 1) if false_positive_capacity else (0, 0)
|
||||
denominator = fixed[metric_id]
|
||||
false_alarm = metric_id in {
|
||||
MetricId.FIXED_FALSE_ALARM_RATE,
|
||||
MetricId.ORDINARY_FALSE_ALARMS_PER_100,
|
||||
}
|
||||
return (0 if false_alarm else denominator), denominator
|
||||
|
||||
|
||||
def _grouped_thresholds(
|
||||
thresholds: tuple[AdmissionThreshold, ...],
|
||||
) -> tuple[tuple[MetricId, tuple[AdmissionThreshold, ...]], ...]:
|
||||
return tuple(
|
||||
(metric_id, tuple(item for item in thresholds if item.metric_id is metric_id))
|
||||
for metric_id in MetricId
|
||||
if any(item.metric_id is metric_id for item in thresholds)
|
||||
)
|
||||
|
||||
|
||||
def _witness(
|
||||
rule: MetricRule,
|
||||
thresholds: tuple[AdmissionThreshold, ...],
|
||||
capacities: _Capacities,
|
||||
) -> FeasibleCountAssignment:
|
||||
numerator, denominator = _best_counts(rule.metric_id, capacities)
|
||||
try:
|
||||
result = evaluate_metric(rule, numerator=numerator, denominator=denominator)
|
||||
outcomes = tuple(metric_passes(result, item.statistic, item.comparator, item.target) for item in thresholds)
|
||||
except MeasurementContractError as exc:
|
||||
raise AdmissionFeasibilityError(str(exc)) from exc
|
||||
if not all(outcomes):
|
||||
raise AdmissionFeasibilityError(f"{rule.metric_id.value} thresholds are arithmetically impossible")
|
||||
return FeasibleCountAssignment(rule.metric_id, numerator, denominator)
|
||||
|
||||
|
||||
def _feasibility_value(
|
||||
census: PopulationCensus,
|
||||
policy: DenominatorPolicy,
|
||||
threshold_digest: str,
|
||||
witnesses: tuple[FeasibleCountAssignment, ...],
|
||||
) -> canonical.JsonValue:
|
||||
return {
|
||||
"schema_version": FEASIBILITY_SCHEMA,
|
||||
"census_digest": census.census_digest,
|
||||
"denominator_policy_digest": policy.policy_digest,
|
||||
"threshold_set_digest": threshold_digest,
|
||||
"witnesses": tuple(
|
||||
{
|
||||
"metric_id": item.metric_id.value,
|
||||
"numerator": item.numerator,
|
||||
"denominator": item.denominator,
|
||||
}
|
||||
for item in witnesses
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def evaluate_threshold_feasibility(
|
||||
manifest: AdmissionManifestV2,
|
||||
policy: DenominatorPolicy,
|
||||
review_policy: ReviewIndependencePolicy,
|
||||
resolver: canonical.ContentResolver,
|
||||
) -> ThresholdFeasibility:
|
||||
capacities, threshold_digest = _validated_inputs(manifest, policy, review_policy, resolver)
|
||||
rules = {rule.metric_id: rule for rule in policy.rules}
|
||||
witnesses = tuple(_witness(rules[metric_id], group, capacities) for metric_id, group in _grouped_thresholds(manifest.thresholds))
|
||||
census = manifest.census
|
||||
digest = canonical.content_digest(_feasibility_value(census, policy, threshold_digest, witnesses))
|
||||
return ThresholdFeasibility(
|
||||
census.census_digest,
|
||||
policy.policy_digest,
|
||||
threshold_digest,
|
||||
witnesses,
|
||||
digest,
|
||||
)
|
||||
@@ -0,0 +1,308 @@
|
||||
"""Label-blind arithmetic admission feasibility (Requirements 12.3 and 12.4)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import replace
|
||||
|
||||
import pytest
|
||||
from detection_core_fixtures import InMemoryContentResolver, stored_artifact
|
||||
|
||||
from openultrasast.core import canonical
|
||||
from openultrasast.core.admission_feasibility import (
|
||||
AdmissionFeasibilityError,
|
||||
ThresholdFeasibility,
|
||||
evaluate_threshold_feasibility,
|
||||
)
|
||||
from openultrasast.core.admission_review import (
|
||||
AdmissionManifestV2,
|
||||
AdmissionThreshold,
|
||||
PopulationCensus,
|
||||
ScopeCensus,
|
||||
derive_census_digest,
|
||||
derive_census_payload_digest,
|
||||
derive_manifest_digest,
|
||||
derive_manifest_payload_digest,
|
||||
derive_threshold_set_digest,
|
||||
)
|
||||
from openultrasast.core.factpack import (
|
||||
IndependenceAttestation,
|
||||
ReviewIndependencePolicy,
|
||||
derive_independence_attestation_digest,
|
||||
derive_review_independence_policy_digest,
|
||||
)
|
||||
from openultrasast.core.measurement import (
|
||||
BaselineIdentity,
|
||||
Comparator,
|
||||
DenominatorPolicy,
|
||||
IntervalMethod,
|
||||
MetricId,
|
||||
ScopeKey,
|
||||
Statistic,
|
||||
standard_metric_rules,
|
||||
)
|
||||
|
||||
|
||||
def _scope() -> ScopeKey:
|
||||
return ScopeKey("access-control", "python")
|
||||
|
||||
|
||||
def _census(
|
||||
*,
|
||||
pairs: int = 12,
|
||||
vulnerable: int = 12,
|
||||
fixed: int = 12,
|
||||
ordinary: int = 20,
|
||||
sites: int = 8,
|
||||
calls: int = 5,
|
||||
) -> PopulationCensus:
|
||||
row = ScopeCensus(
|
||||
_scope(),
|
||||
vulnerable + fixed + ordinary,
|
||||
pairs,
|
||||
vulnerable,
|
||||
fixed,
|
||||
ordinary,
|
||||
sites,
|
||||
calls,
|
||||
)
|
||||
disclosure, _ = stored_artifact(b'{"conflicts":[]}', artifact_id="feasibility-conflicts")
|
||||
producer = canonical.content_digest("benchmark-producer")
|
||||
reviewer = canonical.content_digest("independent-reviewer")
|
||||
provisional_review = _attestation("0" * 64, producer, reviewer, disclosure)
|
||||
provisional = PopulationCensus(canonical.content_digest("population"), (row,), False, "0" * 64, provisional_review, "0" * 64)
|
||||
payload = derive_census_payload_digest(provisional)
|
||||
reviewed = replace(provisional, census_payload_digest=payload, review=_attestation(payload, producer, reviewer, disclosure))
|
||||
return replace(reviewed, census_digest=derive_census_digest(reviewed))
|
||||
|
||||
|
||||
def _policy(scope: ScopeKey | None = None) -> DenominatorPolicy:
|
||||
rules = standard_metric_rules((scope or _scope(),))
|
||||
return DenominatorPolicy(rules, IntervalMethod.WILSON_95, "")
|
||||
|
||||
|
||||
def _thresholds(*values: AdmissionThreshold) -> tuple[AdmissionThreshold, ...]:
|
||||
return tuple(sorted(values, key=lambda item: (item.metric_id.value, item.statistic.value)))
|
||||
|
||||
|
||||
def _threshold(
|
||||
metric: MetricId,
|
||||
statistic: Statistic,
|
||||
comparator: Comparator,
|
||||
target: float,
|
||||
) -> AdmissionThreshold:
|
||||
return AdmissionThreshold(metric, statistic, comparator, target)
|
||||
|
||||
|
||||
def _attestation(
|
||||
subject: str,
|
||||
producer: str,
|
||||
reviewer: str,
|
||||
disclosure: canonical.ContentRef,
|
||||
) -> IndependenceAttestation:
|
||||
provisional = IndependenceAttestation(subject, producer, reviewer, disclosure, "0" * 64)
|
||||
return replace(provisional, attestation_digest=derive_independence_attestation_digest(provisional))
|
||||
|
||||
|
||||
def _review_policy(reviewer: str) -> ReviewIndependencePolicy:
|
||||
stores = ("trusted-store",)
|
||||
digest = derive_review_independence_policy_digest((reviewer,), (), stores, 4096)
|
||||
return ReviewIndependencePolicy((reviewer,), (), stores, 4096, digest)
|
||||
|
||||
|
||||
def _baseline(
|
||||
census: PopulationCensus,
|
||||
denominator: DenominatorPolicy,
|
||||
threshold_digest: str,
|
||||
) -> BaselineIdentity:
|
||||
return BaselineIdentity(
|
||||
"arm-a",
|
||||
census.population_digest,
|
||||
canonical.content_digest("freeze"),
|
||||
denominator.rules[0].scopes,
|
||||
(canonical.content_digest("contract"),),
|
||||
canonical.content_digest("catalog"),
|
||||
canonical.content_digest("fact-schema"),
|
||||
canonical.content_digest("fact-set"),
|
||||
canonical.content_digest("acquisition"),
|
||||
"analyzer-v2",
|
||||
canonical.content_digest("configuration"),
|
||||
(),
|
||||
None,
|
||||
denominator.policy_digest,
|
||||
threshold_digest,
|
||||
canonical.content_digest("outcomes"),
|
||||
)
|
||||
|
||||
|
||||
def _reviewed_manifest(
|
||||
census: PopulationCensus,
|
||||
denominator: DenominatorPolicy,
|
||||
thresholds: tuple[AdmissionThreshold, ...],
|
||||
) -> tuple[AdmissionManifestV2, ReviewIndependencePolicy, InMemoryContentResolver]:
|
||||
disclosure, resolver = stored_artifact(b'{"conflicts":[]}', artifact_id="feasibility-conflicts")
|
||||
producer = canonical.content_digest("benchmark-producer")
|
||||
reviewer = canonical.content_digest("independent-reviewer")
|
||||
threshold_digest = derive_threshold_set_digest(thresholds)
|
||||
baseline = _baseline(census, denominator, threshold_digest)
|
||||
provisional_review = _attestation("0" * 64, producer, reviewer, disclosure)
|
||||
provisional = AdmissionManifestV2(
|
||||
"0" * 64,
|
||||
provisional_review,
|
||||
True,
|
||||
False,
|
||||
baseline,
|
||||
census,
|
||||
(canonical.content_digest("readiness"),),
|
||||
canonical.content_digest("custody"),
|
||||
denominator.policy_digest,
|
||||
thresholds,
|
||||
threshold_digest,
|
||||
canonical.content_digest("threshold-lineage"),
|
||||
canonical.content_digest("validation-context"),
|
||||
"0" * 64,
|
||||
)
|
||||
payload = derive_manifest_payload_digest(provisional)
|
||||
reviewed = replace(provisional, manifest_payload_digest=payload, review=_attestation(payload, producer, reviewer, disclosure))
|
||||
return replace(reviewed, manifest_digest=derive_manifest_digest(reviewed)), _review_policy(reviewer), resolver
|
||||
|
||||
|
||||
def _evaluate(
|
||||
census: PopulationCensus,
|
||||
denominator: DenominatorPolicy,
|
||||
thresholds: tuple[AdmissionThreshold, ...],
|
||||
) -> ThresholdFeasibility:
|
||||
manifest, review_policy, resolver = _reviewed_manifest(census, denominator, thresholds)
|
||||
return evaluate_threshold_feasibility(manifest, denominator, review_policy, resolver)
|
||||
|
||||
|
||||
def test_stale_reviewed_census_identity_is_refused_before_capacity_use() -> None:
|
||||
thresholds = (_threshold(MetricId.SEMANTIC_STAGE_COVERAGE, Statistic.COUNT, Comparator.AT_LEAST, 50.0),)
|
||||
census = _census(sites=12)
|
||||
denominator = _policy()
|
||||
manifest, review_policy, resolver = _reviewed_manifest(census, denominator, thresholds)
|
||||
changed_row = replace(census.rows[0], semantic_site_capacity=99)
|
||||
stale_census = replace(census, rows=(changed_row,))
|
||||
stale_manifest = replace(manifest, census=stale_census)
|
||||
|
||||
with pytest.raises(AdmissionFeasibilityError, match="census payload"):
|
||||
evaluate_threshold_feasibility(stale_manifest, denominator, review_policy, resolver)
|
||||
|
||||
|
||||
def test_enumerates_capacity_witnesses_for_every_metric_family() -> None:
|
||||
thresholds = _thresholds(
|
||||
_threshold(MetricId.PAIRED_CORRECTNESS, Statistic.COUNT, Comparator.AT_LEAST, 6.0),
|
||||
_threshold(MetricId.VULNERABLE_RECALL, Statistic.RATE, Comparator.AT_LEAST, 0.5),
|
||||
_threshold(MetricId.FIXED_FALSE_ALARM_RATE, Statistic.RATE, Comparator.AT_MOST, 0.1),
|
||||
_threshold(MetricId.FIXED_FALSE_ALARM_RATE, Statistic.WILSON_UPPER, Comparator.AT_MOST, 0.25),
|
||||
_threshold(MetricId.ORDINARY_FALSE_ALARMS_PER_100, Statistic.PER_100, Comparator.AT_MOST, 5.0),
|
||||
_threshold(MetricId.ORDINARY_FALSE_ALARMS_PER_100, Statistic.WILSON_LOWER, Comparator.AT_MOST, 0.0),
|
||||
_threshold(MetricId.PRECISION, Statistic.WILSON_LOWER, Comparator.AT_LEAST, 0.7),
|
||||
_threshold(MetricId.SEMANTIC_STAGE_COVERAGE, Statistic.COUNT, Comparator.AT_LEAST, 4.0),
|
||||
_threshold(MetricId.MODEL_SUCCESS_RATE, Statistic.WILSON_UPPER, Comparator.AT_LEAST, 0.9),
|
||||
)
|
||||
|
||||
result = _evaluate(_census(), _policy(), thresholds)
|
||||
|
||||
witnesses = {item.metric_id: (item.numerator, item.denominator) for item in result.witnesses}
|
||||
assert witnesses == {
|
||||
MetricId.PAIRED_CORRECTNESS: (12, 12),
|
||||
MetricId.VULNERABLE_RECALL: (12, 12),
|
||||
MetricId.FIXED_FALSE_ALARM_RATE: (0, 12),
|
||||
MetricId.ORDINARY_FALSE_ALARMS_PER_100: (0, 20),
|
||||
MetricId.PRECISION: (12, 12),
|
||||
MetricId.SEMANTIC_STAGE_COVERAGE: (8, 8),
|
||||
MetricId.MODEL_SUCCESS_RATE: (5, 5),
|
||||
}
|
||||
assert result.census_digest == _census().census_digest
|
||||
assert result.denominator_policy_digest == _policy().policy_digest
|
||||
assert len(result.threshold_set_digest) == 64
|
||||
assert len(result.feasibility_digest) == 64
|
||||
|
||||
|
||||
def test_one_assignment_must_jointly_satisfy_grouped_statistics() -> None:
|
||||
thresholds = _thresholds(
|
||||
_threshold(MetricId.PAIRED_CORRECTNESS, Statistic.COUNT, Comparator.AT_LEAST, 10.0),
|
||||
_threshold(MetricId.PAIRED_CORRECTNESS, Statistic.RATE, Comparator.AT_LEAST, 0.8),
|
||||
_threshold(MetricId.PAIRED_CORRECTNESS, Statistic.WILSON_LOWER, Comparator.AT_LEAST, 0.5),
|
||||
_threshold(MetricId.PAIRED_CORRECTNESS, Statistic.WILSON_UPPER, Comparator.AT_LEAST, 0.8),
|
||||
)
|
||||
|
||||
result = _evaluate(_census(), _policy(), thresholds)
|
||||
|
||||
assert [(item.metric_id, item.numerator, item.denominator) for item in result.witnesses] == [(MetricId.PAIRED_CORRECTNESS, 12, 12)]
|
||||
|
||||
|
||||
def test_historical_impossible_denominator_is_configuration_refusal() -> None:
|
||||
threshold = _threshold(
|
||||
MetricId.VULNERABLE_RECALL,
|
||||
Statistic.WILSON_LOWER,
|
||||
Comparator.AT_LEAST,
|
||||
0.5,
|
||||
)
|
||||
|
||||
with pytest.raises(AdmissionFeasibilityError, match="vulnerable_recall.*arithmetically impossible"):
|
||||
_evaluate(
|
||||
_census(pairs=2, vulnerable=2, fixed=2, ordinary=4),
|
||||
_policy(),
|
||||
(threshold,),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("census", "threshold"),
|
||||
(
|
||||
(
|
||||
_census(pairs=0, vulnerable=0, fixed=0, ordinary=4),
|
||||
_threshold(MetricId.VULNERABLE_RECALL, Statistic.COUNT, Comparator.AT_LEAST, 0.0),
|
||||
),
|
||||
(
|
||||
_census(calls=0),
|
||||
_threshold(MetricId.MODEL_SUCCESS_RATE, Statistic.RATE, Comparator.AT_LEAST, 0.0),
|
||||
),
|
||||
(
|
||||
_census(pairs=0, vulnerable=0),
|
||||
_threshold(MetricId.PRECISION, Statistic.RATE, Comparator.AT_LEAST, 1.0),
|
||||
),
|
||||
),
|
||||
)
|
||||
def test_zero_or_absent_success_capacity_never_becomes_a_clean_zero(
|
||||
census: PopulationCensus,
|
||||
threshold: AdmissionThreshold,
|
||||
) -> None:
|
||||
with pytest.raises(AdmissionFeasibilityError, match="arithmetically impossible"):
|
||||
_evaluate(census, _policy(), (threshold,))
|
||||
|
||||
|
||||
def test_precision_enumerates_a_false_positive_denominator_when_no_true_case_exists() -> None:
|
||||
threshold = _threshold(MetricId.PRECISION, Statistic.RATE, Comparator.AT_LEAST, 0.0)
|
||||
|
||||
result = _evaluate(_census(pairs=0, vulnerable=0), _policy(), (threshold,))
|
||||
|
||||
assert [(item.numerator, item.denominator) for item in result.witnesses] == [(0, 1)]
|
||||
|
||||
|
||||
def test_fixed_metric_algebra_rejects_wrong_comparator_or_statistic() -> None:
|
||||
reversed_comparator = _threshold(
|
||||
MetricId.VULNERABLE_RECALL,
|
||||
Statistic.RATE,
|
||||
Comparator.AT_MOST,
|
||||
1.0,
|
||||
)
|
||||
wrong_statistic = _threshold(
|
||||
MetricId.ORDINARY_FALSE_ALARMS_PER_100,
|
||||
Statistic.RATE,
|
||||
Comparator.AT_MOST,
|
||||
5.0,
|
||||
)
|
||||
|
||||
with pytest.raises(AdmissionFeasibilityError, match="comparator direction"):
|
||||
_evaluate(_census(), _policy(), (reversed_comparator,))
|
||||
with pytest.raises(AdmissionFeasibilityError, match="per_100"):
|
||||
_evaluate(_census(), _policy(), (wrong_statistic,))
|
||||
|
||||
|
||||
def test_policy_scope_must_match_the_label_blind_census() -> None:
|
||||
threshold = _threshold(MetricId.PAIRED_CORRECTNESS, Statistic.COUNT, Comparator.AT_LEAST, 1.0)
|
||||
|
||||
with pytest.raises(AdmissionFeasibilityError, match="scope"):
|
||||
_evaluate(_census(), _policy(ScopeKey("other", "python")), (threshold,))
|
||||
Reference in New Issue
Block a user