DataOps-platform/tests/performance/test_rule_execution_capacity.py

from __future__ import annotations

import json
from pathlib import Path

from scripts.validation.rule_execution_capacity import (
    CapacityLimits,
    CapacityObservation,
    assert_capacity_within_limits,
    write_capacity_evidence,
)


def test_capacity_evidence_is_recalculable_and_fails_closed(tmp_path):
    limits = CapacityLimits(
        runner_memory_bytes=512 * 1024 * 1024,
        datasource_pool_budget=4,
        artifact_bytes=128 * 1024 * 1024,
    )
    observations = [
        CapacityObservation(
            backend="sql_pushdown",
            rows=100_000,
            elapsed_seconds=2.0,
            peak_memory_bytes=64 * 1024 * 1024,
            pool_peak=2,
            artifact_bytes=0,
            node_durations_seconds=(1.8, 2.0, 2.2),
        ),
        CapacityObservation(
            backend="polars_batch",
            rows=100_000,
            elapsed_seconds=4.0,
            peak_memory_bytes=128 * 1024 * 1024,
            pool_peak=1,
            artifact_bytes=8 * 1024 * 1024,
            node_durations_seconds=(3.8, 4.0, 4.2),
        ),
    ]
    for observation in observations:
        assert_capacity_within_limits(observation, limits)

    output = tmp_path / "capacity.json"
    write_capacity_evidence(
        output,
        observations=observations,
        limits=limits,
        machine={"source": "pytest", "logical_cpu_count": 8},
    )
    evidence = json.loads(output.read_text(encoding="utf-8"))
    assert evidence["schema_version"] == "1.0"
    assert evidence["limits"]["datasource_pool_budget"] == 4
    assert evidence["observations"][0]["throughput_rows_per_second"] == 50_000
    assert evidence["observations"][0]["p95_node_seconds"] == 2.2
    assert evidence["evidence_sha256"] == (
        write_capacity_evidence.digest_payload(evidence)
    )


def test_capacity_gate_rejects_runner_pool_and_artifact_overruns():
    limits = CapacityLimits(
        runner_memory_bytes=100,
        datasource_pool_budget=2,
        artifact_bytes=100,
    )
    cases = [
        CapacityObservation(
            "polars_batch", 100, 1.0, 101, 1, 10, (1.0,)
        ),
        CapacityObservation(
            "sql_pushdown", 100, 1.0, 10, 3, 0, (1.0,)
        ),
        CapacityObservation(
            "polars_batch", 100, 1.0, 10, 1, 101, (1.0,)
        ),
    ]
    for observation in cases:
        try:
            assert_capacity_within_limits(observation, limits)
        except ValueError as exc:
            assert "capacity limit exceeded" in str(exc)
        else:
            raise AssertionError("capacity overrun was accepted")


def test_capacity_targets_declare_100k_1m_and_10m_or_hardware_limit():
    manifest = Path("docs/validation/data-rule-m5-capacity-targets.json")
    assert manifest.is_file()
    payload = json.loads(manifest.read_text(encoding="utf-8"))
    assert payload["requested_rows"] == [100_000, 1_000_000, 10_000_000]
    assert payload["hardware_limit"]["maximum_measured_rows"] >= 100_000
    if payload["hardware_limit"]["maximum_measured_rows"] < 10_000_000:
        assert payload["hardware_limit"]["reason"]