zhifu-platform/tests/test_scale_capacity.py

import pytest
from pydantic import ValidationError

from zhifu.operations.scale_capacity import ScaleRequest, inspect_capacity


@pytest.mark.parametrize("values", [
    {"replicas": 0}, {"replicas": 17}, {"replicas": 2, "pool_size": 0},
    {"replicas": 2, "max_overflow": -1}, {"replicas": 2, "reserved_connections": -1},
])
def test_scale_budget_rejects_invalid_bounds(values):
    with pytest.raises(ValidationError):
        ScaleRequest(**values)


async def test_scale_budget_reads_real_mysql_capacity():
    request = ScaleRequest(replicas=2, pool_size=5, max_overflow=5)
    result = await inspect_capacity(request)
    assert result["mysql_max_connections"] > 0
    assert result["currently_connected"] >= 1
    assert result["planned_peak_connections"] == 20
    expected = result["mysql_max_connections"] - result["currently_connected"] - 40
    assert result["remaining_connections"] == expected
    assert result["status"] == ("READY" if expected >= 0 else "BLOCKED")
    assert result["database_configuration_changed"] is False