test_apps_utils.py

import pytest
import asyncio
import time
import json
from pathlib import Path
import sys

# Add the parent directory to the path so we can import from control
sys.path.append(str(Path(__file__).parent.parent))
from control.apps_utils import are_tests_correct, Test, remove_comments_and_format, format_code


@pytest.mark.asyncio
async def test_correct_code():
    """Test that a correct solution passes all tests."""
    tests = [
        Test(input="5", expected_output="120"),
        Test(input="3", expected_output="6"),
    ]
    code = """
def factorial(n):
    if n == 0 or n == 1:
        return 1
    else:
        return n * factorial(n-1)

n = int(input().strip())
print(factorial(n))
"""
    result, metadata = await are_tests_correct(tests, code, fn_name=None)
    assert result is not None
    assert all(result)
    assert "test_outputs" in metadata
    assert len(metadata["test_outputs"]) == 2
    assert metadata["test_outputs"][0]["stdout"].strip() == "120"
    assert metadata["test_outputs"][1]["stdout"].strip() == "6"


@pytest.mark.asyncio
async def test_incorrect_function_code():
    """Test that an incorrect function solution fails some tests."""
    tests = [
        Test(input=[5], expected_output=120),
        Test(input=[0], expected_output=1),
        Test(input=[3], expected_output=6),
        Test(input=[6], expected_output="aaa"),
    ]
    code = """
def factorial(n):
    # This implementation is incorrect for n=0
    if n == 1:
        return 1
    else:
        return n * factorial(n-1)
"""
    result, metadata = await are_tests_correct(tests, code, fn_name="factorial")

    print("stdout:\n")
    print(metadata["stdout"])
    print("stderr:\n")
    print(metadata["stderr"])

    assert result is not None
    assert result[0] == True  # Test for n=5 passes
    assert result[1] == -1  # Test for n=0 fails (will cause runtime error)
    assert result[2] == True  # Test for n=3 passes
    assert result[3] == False  # Test for n=6 fails
    assert "test_outputs" in metadata
    assert len(metadata["test_outputs"]) == 4
    assert metadata["test_outputs"][0]["fn_output"] == 120
    assert metadata["test_outputs"][1]["fn_output"] is None
    assert metadata["test_outputs"][2]["fn_output"] == 6
    assert metadata["test_outputs"][3]["fn_output"] == 720
    print(metadata["test_outputs"][1]["stderr"])
    assert "RecursionError" in metadata["test_outputs"][1]["stderr"]



@pytest.mark.asyncio
async def test_incorrect_code():
    """Test that an incorrect solution fails some tests."""
    tests = [
        Test(input="5", expected_output="120"),
        Test(input="0", expected_output="1"),
        Test(input="3", expected_output="6"),
        Test(input="6", expected_output="111"), # Should return false
    ]
    code = """
def factorial(n):
    # This implementation is incorrect for n=0
    if n == 1:
        return 1
    else:
        return n * factorial(n-1)

n = int(input().strip())
print(factorial(n))
"""
    result, metadata = await are_tests_correct(tests, code, fn_name=None)

    print("stdout:\n")
    print(metadata["stdout"])
    print("stderr:\n")
    print(metadata["stderr"])

    assert result is not None
    assert result[0] == True  # Test for n=5 passes
    assert result[1] == -1  # Test for n=0 fails (will cause runtime error)
    assert result[2] == True  # Test for n=3 passes
    assert result[3] == False  # Test for n=6 fails (returns 111 instead of false)
    assert "test_outputs" in metadata
    assert len(metadata["test_outputs"]) == 4
    assert metadata["test_outputs"][0]["stdout"].strip() == "120"
    assert metadata["test_outputs"][2]["stdout"].strip() == "6"
    assert metadata["test_outputs"][3]["stdout"].strip() == "720"
    print(metadata["test_outputs"][1]["stderr"])
    assert "RecursionError" in metadata["test_outputs"][1]["stderr"]


# TODO: check the case when the fn_name is not found!

@pytest.mark.asyncio
async def test_stdout_stderr_capture():
    """Test that stdout and stderr are captured correctly."""
    tests = [Test(input="test", expected_output="TEST")]
    code = """
s = input().strip()
print(s.upper())
print("Debug info", file=sys.stderr)
"""
    result, metadata = await are_tests_correct(tests, code, fn_name=None)
    assert result is not None
    assert all(result)
    assert "test_outputs" in metadata
    assert metadata["test_outputs"][0]["stdout"].strip() == "TEST"
    assert "stderr" in metadata
    assert metadata["test_outputs"][0]["stderr"].strip() == "Debug info"


@pytest.mark.asyncio
async def test_compile_error():
    """Test behavior with code that has compilation errors."""
    tests = [Test(input="5", expected_output="120")]
    code = """
def factorial(n):
    if n == 0 or n == 1:
        return 1
    else:
        return n * factorial(n-1

n = int(input().strip())
print(factorial(n))
"""
    result, metadata = await are_tests_correct(tests, code, fn_name=None)
    assert result is not None
    assert result[0] == -2
    assert "test_outputs" in metadata
    print(metadata["stderr"])
    print(metadata["test_outputs"][0]["stderr"])
    assert "SyntaxError" in metadata["test_outputs"][0]["stderr"].strip()


@pytest.mark.asyncio
async def test_compile_error_in_function():
    """Test behavior with function code that has compilation errors."""
    tests = [Test(input="5", expected_output="120")]
    code = """
def factorial(n):
    if n == 0 or n == 1:
        return 1
    else:
        return n * factorial(n-1
"""
    result, metadata = await are_tests_correct(tests, code, fn_name="factorial")
    assert result is not None
    assert result[0] == -2
    print(metadata["test_outputs"][0]["stderr"])
    assert "test_outputs" in metadata
    assert "SyntaxError" in metadata["test_outputs"][0]["stderr"].strip()



@pytest.mark.asyncio
async def test_runtime_error():
    """Test behavior with code that has runtime errors."""
    tests = [Test(input="5", expected_output="120")]
    code = """
def factorial(n):
    if n == 0 or n == 1:
        return 1
    else:
        return n * factorial(n-1)

n = int(input().strip())
# Intentional error: division by zero
result = factorial(n) / 0
print(result)
"""
    result, metadata = await are_tests_correct(tests, code, fn_name=None)
    assert result is not None
    assert result[0] == -1
    assert "ZeroDivisionError" in metadata["test_outputs"][0]["stderr"]


@pytest.mark.asyncio
async def test_timeout():
    """Test that code execution times out after the specified timeout."""
    tests = [Test(input="5", expected_output="120")]
    code = """
import time
n = int(input().strip())
time.sleep(10)  # Sleep longer than the timeout
print(n)
"""
    start_time = time.time()
    result, metadata = await are_tests_correct(tests, code, fn_name=None, timeout_seconds=2)
    elapsed_time = time.time() - start_time
    
    assert result is not None
    assert result[0] == -1
    print(metadata["stderr"])
    print(metadata["test_outputs"][0]["stderr"])
    assert "TimeoutException" in metadata["test_outputs"][0]["stderr"]
    # Check that it took approximately the timeout duration (with some margin)
    print(f"Elapsed time: {elapsed_time}")
    assert elapsed_time < 4  # Should be less than 2x the timeout


@pytest.mark.asyncio
async def test_numpy_available():
    """Test that numpy is available as np without explicit import."""
    tests = [Test(input="5", expected_output="15")]
    code = """
n = int(input().strip())
# Use numpy without importing it (it's imported in testing_util.py)
result = np.sum(np.array([n, n, n]))
print(result)
"""
    result, metadata = await are_tests_correct(tests, code, fn_name=None)
    assert result is not None
    assert result[0] == True
    assert "ImportError" not in metadata.get("stderr", "")


@pytest.mark.asyncio
async def test_memory_limit():
    """Test that code execution respects memory limits."""
    tests = [Test(input="5", expected_output="5")]
    code = """
n = int(input().strip())
# Try to allocate a large amount of memory
large_list = [0] * (1024 * 1024 * 256)  # Attempt to allocate ~2GB
print(n)
"""
    # Too low memory limits will cause import errors (numpy complains)
    result, metadata = await are_tests_correct(tests, code, fn_name=None, memory_limit_mb=512)
    # The exact error message might vary by platform, but should indicate a memory issue
    print(metadata["stderr"])
    # print(metadata["test_outputs"][0]["stderr"])

    assert result is not None
    assert result[0] == -1
    assert any(
        error_type in metadata["test_outputs"][0]["stderr"] 
        for error_type in ["MemoryError", "Killed", "Resource"]
    )


def test_remove_comments_normal_code():
    """Test remove_comments with normal code containing comments."""
    code = '''
def add(a, b):
    # This function adds two numbers
    return a + b
'''
    expected = '''
def add(a, b):
    
    return a + b
'''
    print(remove_comments_and_format(code))
    assert remove_comments_and_format(code) == format_code(expected)


def test_remove_comments_hash_in_strings():
    """Test remove_comments with code containing hash symbols in strings."""
    code = '''
def say_hash(): # This is a comment
    print("This is a hash symbol: # not a comment")
    return "End of function # still not a comment"
'''
    expected = '''
def say_hash():
    print("This is a hash symbol: # not a comment")
    return "End of function # still not a comment"
'''
    assert remove_comments_and_format(code) == format_code(expected)