mcp-forge/tests/pod_executor/security/test_resource_limits.py

"""
Tests for pod_executor resource limits module.

Tests cover all parsing and validation requirements.
"""

import pytest


def test_parse_memory_string_megabytes():
    """Test parsing memory string with megabytes suffix."""
    from pod_executor.security.resource_limits import parse_memory_string
    
    result = parse_memory_string("512m")
    assert result == 536870912  # 512 * 1024 * 1024


def test_parse_memory_string_gigabytes():
    """Test parsing memory string with gigabytes suffix."""
    from pod_executor.security.resource_limits import parse_memory_string
    
    result = parse_memory_string("2g")
    assert result == 2147483648  # 2 * 1024 * 1024 * 1024


def test_parse_memory_string_kilobytes():
    """Test parsing memory string with kilobytes suffix."""
    from pod_executor.security.resource_limits import parse_memory_string
    
    result = parse_memory_string("1024k")
    assert result == 1048576  # 1024 * 1024


def test_parse_memory_string_case_insensitive():
    """Test that memory string parsing is case-insensitive."""
    from pod_executor.security.resource_limits import parse_memory_string
    
    assert parse_memory_string("512M") == 536870912
    assert parse_memory_string("2G") == 2147483648
    assert parse_memory_string("1024K") == 1048576


def test_parse_memory_string_invalid_format_raises_value_error():
    """Test that invalid format raises ValueError."""
    from pod_executor.security.resource_limits import parse_memory_string
    
    with pytest.raises(ValueError) as exc_info:
        parse_memory_string("invalid")
    assert "invalid" in str(exc_info.value).lower() or "format" in str(exc_info.value).lower()
    
    with pytest.raises(ValueError):
        parse_memory_string("512x")  # Invalid suffix
    
    with pytest.raises(ValueError):
        parse_memory_string("abc")  # Not a number


def test_parse_memory_string_negative_value_raises_value_error():
    """Test that negative values raise ValueError."""
    from pod_executor.security.resource_limits import parse_memory_string
    
    with pytest.raises(ValueError) as exc_info:
        parse_memory_string("-512m")
    assert "positive" in str(exc_info.value).lower() or "negative" in str(exc_info.value).lower()


def test_parse_memory_string_zero_value_raises_value_error():
    """Test that zero value raises ValueError."""
    from pod_executor.security.resource_limits import parse_memory_string
    
    with pytest.raises(ValueError) as exc_info:
        parse_memory_string("0m")
    assert "positive" in str(exc_info.value).lower() or "zero" in str(exc_info.value).lower()


def test_parse_cpu_quota_valid_value():
    """Test that valid CPU quota values are accepted."""
    from pod_executor.security.resource_limits import parse_cpu_quota
    
    result = parse_cpu_quota(50000)
    assert result == 50000
    
    result = parse_cpu_quota(100000)  # 100% of one core
    assert result == 100000


def test_parse_cpu_quota_max_limit():
    """Test that CPU quota has a reasonable maximum (10 cores)."""
    from pod_executor.security.resource_limits import parse_cpu_quota
    
    # Should accept up to 1000000 (10 cores)
    result = parse_cpu_quota(1000000)
    assert result == 1000000
    
    # Should reject more than 10 cores
    with pytest.raises(ValueError) as exc_info:
        parse_cpu_quota(1000001)
    assert "1000000" in str(exc_info.value) or "maximum" in str(exc_info.value).lower()


def test_parse_cpu_quota_negative_raises_value_error():
    """Test that negative CPU quota raises ValueError."""
    from pod_executor.security.resource_limits import parse_cpu_quota
    
    with pytest.raises(ValueError) as exc_info:
        parse_cpu_quota(-1)
    assert "positive" in str(exc_info.value).lower() or "negative" in str(exc_info.value).lower()


def test_parse_cpu_quota_zero_raises_value_error():
    """Test that zero CPU quota raises ValueError."""
    from pod_executor.security.resource_limits import parse_cpu_quota
    
    with pytest.raises(ValueError) as exc_info:
        parse_cpu_quota(0)
    assert "positive" in str(exc_info.value).lower() or "zero" in str(exc_info.value).lower()


def test_parse_storage_string_same_as_memory():
    """Test that storage parsing works the same as memory parsing."""
    from pod_executor.security.resource_limits import parse_storage_string
    
    assert parse_storage_string("1g") == 1073741824
    assert parse_storage_string("512m") == 536870912
    assert parse_storage_string("2048k") == 2097152


def test_resource_limits_class_initialization():
    """Test ResourceLimits class initializes correctly."""
    from pod_executor.security.resource_limits import ResourceLimits
    
    limits = ResourceLimits(
        memory="512m",
        storage="1g",
        cpu_quota=50000,
        timeout=300
    )
    
    assert limits.memory_bytes == 536870912
    assert limits.storage_bytes == 1073741824
    assert limits.cpu_quota == 50000
    assert limits.timeout == 300


def test_resource_limits_validates_memory():
    """Test that ResourceLimits validates memory string."""
    from pod_executor.security.resource_limits import ResourceLimits
    
    with pytest.raises(ValueError):
        ResourceLimits(
            memory="invalid",
            storage="1g",
            cpu_quota=50000,
            timeout=300
        )


def test_resource_limits_validates_storage():
    """Test that ResourceLimits validates storage string."""
    from pod_executor.security.resource_limits import ResourceLimits
    
    with pytest.raises(ValueError):
        ResourceLimits(
            memory="512m",
            storage="invalid",
            cpu_quota=50000,
            timeout=300
        )


def test_resource_limits_validates_cpu_quota():
    """Test that ResourceLimits validates CPU quota."""
    from pod_executor.security.resource_limits import ResourceLimits
    
    with pytest.raises(ValueError):
        ResourceLimits(
            memory="512m",
            storage="1g",
            cpu_quota=-1,
            timeout=300
        )


def test_resource_limits_to_podman_params():
    """Test conversion to Podman container parameters."""
    from pod_executor.security.resource_limits import ResourceLimits
    
    limits = ResourceLimits(
        memory="512m",
        storage="1g",
        cpu_quota=50000,
        timeout=300
    )
    
    params = limits.to_podman_params()
    
    assert isinstance(params, dict)
    assert "mem_limit" in params
    assert params["mem_limit"] == "536870912"  # Should be string for Podman
    # CPU quota is set via cpu_quota parameter
    assert "cpu_quota" in params
    assert params["cpu_quota"] == 50000


def test_resource_limits_default_timeout():
    """Test that ResourceLimits has a default timeout."""
    from pod_executor.security.resource_limits import ResourceLimits
    
    limits = ResourceLimits(
        memory="512m",
        storage="1g",
        cpu_quota=50000
    )
    
    assert limits.timeout == 300  # Default from signature


def test_parse_memory_string_with_spaces():
    """Test parsing memory strings that have spaces."""
    from pod_executor.security.resource_limits import parse_memory_string
    
    # Should handle spaces gracefully (strip them)
    result = parse_memory_string(" 512m ")
    assert result == 536870912


def test_parse_memory_string_bytes_suffix():
    """Test parsing memory string with bytes suffix (no multiplier)."""
    from pod_executor.security.resource_limits import parse_memory_string
    
    # Just a number (bytes) - should this be supported?
    # Based on architecture, we support k, m, g suffixes
    # Plain numbers should probably raise an error for safety
    with pytest.raises(ValueError):
        parse_memory_string("1024")


def test_resource_limits_storage_quota_in_podman_params():
    """Test that storage limits are tracked internally but not in Podman params."""
    from pod_executor.security.resource_limits import ResourceLimits
    
    limits = ResourceLimits(
        memory="512m",
        storage="1g",
        cpu_quota=50000,
        timeout=300
    )
    
    # Storage is tracked internally
    assert limits.storage_bytes == 1073741824
    
    params = limits.to_podman_params()
    
    # Storage is NOT included in Podman params as it's not directly supported
    # by Podman API for runtime limits. It's tracked for monitoring/validation.
    assert "mem_limit" in params
    assert "cpu_quota" in params


def test_cpu_quota_explanation():
    """Test that CPU quota values have clear meaning."""
    from pod_executor.security.resource_limits import parse_cpu_quota
    
    # 100000 = 100% of one CPU core
    # 50000 = 50% of one CPU core
    # 200000 = 200% = 2 CPU cores
    
    assert parse_cpu_quota(50000) == 50000  # 0.5 cores
    assert parse_cpu_quota(100000) == 100000  # 1 core
    assert parse_cpu_quota(200000) == 200000  # 2 cores


def test_parse_memory_with_decimal():
    """Test parsing memory strings with decimal values."""
    from pod_executor.security.resource_limits import parse_memory_string
    
    # Should handle decimals
    result = parse_memory_string("1.5g")
    assert result == 1610612736  # 1.5 * 1024 * 1024 * 1024