testcase_utils.py

from typing import List, Dict, Any
from loguru import logger
import ast
import re
import json
from tqdm import tqdm


def get_parameter_names(prompt: str, entry_point: str) -> List[str]:
    """
    Extract parameter names from the function signature in the prompt.
    """
    # logger.debug(f"Prompt: {prompt}")
    # logger.debug(f"Entry point: {entry_point}")
    tree = ast.parse(prompt)
    for node in ast.walk(tree):
        # logger.debug(f"Node name: {node.name if hasattr(node, 'name') else None}")
        if isinstance(node, ast.FunctionDef) and node.name == entry_point:
            # Return the parameter names from the function definition that matches the entry point
            return [param.arg for param in node.args.args]
    return []


def parse_tests(test: str, parameter_names: List[str], entry_point: str) -> Dict[str, List[Dict[str, Any]]]:
    """
    Parse the test string into a structured format using AST.
    """
    # Remove the METADATA section
    test = re.sub(r'METADATA = \{[^}]*\}', '', test)

    # Parse the entire test string
    tree = ast.parse(test)

    test_cases = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Assert):
            # Process each assert statement
            test_case = process_assert(node, entry_point, parameter_names)
            if test_case:
                test_cases.append(test_case)

    return {"test_cases": test_cases}


def process_assert(node: ast.Assert, entry_point: str, parameter_names: List[str]) -> Dict[str, Any]:
    """
    Process a single assert statement and extract input and expected output.
    """
    if isinstance(node.test, ast.Compare) and isinstance(node.test.ops[0], ast.Eq):
        left = node.test.left
        right = node.test.comparators[0]

        if isinstance(left, ast.Call) and isinstance(left.func, ast.Name) and left.func.id == "candidate":
            input_dict = process_input(left.args, parameter_names)
            # logger.debug(f"Input: {input_dict}")
            # logger.debug(f"right: {right}")
            # logger.debug(f"right type: {type(right)}")
            # logger.debug(f"right value: {right.name if isinstance(right, ast.Name) else right.s if isinstance(right, ast.Str) else None}")

            try:
                # Attempt to evaluate using literal_eval
                expected_output = ast.literal_eval(right)
            except ValueError:
                # Fallback to eval if literal_eval fails
                # logger.warning("Falling back to eval due to failure in literal_eval")
                expected_output = eval(compile(ast.Expression(right), filename="<ast>", mode="eval"))

            return {"input": input_dict, "expected_output": expected_output}

    return None


def process_input(args: List[ast.expr], parameter_names: List[str]) -> Dict[str, Any]:
    """
    Process the input arguments and match them with parameter names.
    """
    input_dict = {}

    for i, arg in enumerate(args):
        try:
            # Attempt to evaluate using literal_eval for simpler cases
            evaluated_arg = ast.literal_eval(arg)
        except ValueError:
            # Fallback to eval if literal_eval fails
            # logger.warning("Falling back to eval due to failure in literal_eval")
            evaluated_arg = eval(compile(ast.Expression(arg), filename="<ast>", mode="eval"))

        if i < len(parameter_names):
            input_dict[parameter_names[i]] = evaluated_arg
        else:
            # Handle extra arguments if any
            input_dict[f"arg_{i}"] = evaluated_arg

    return input_dict


def parse_all_problems(problems):
    success_count = 0
    unhandled_failures = 0
    for problem in problems:
        try:
            problem = json.loads(problem)

            # logger.info(f"Problem: {problem}")
            # logger.debug(f"Test: {problem['test']}")

            entry_point = problem["entry_point"]
            parameter_names = get_parameter_names(problem["prompt"], entry_point)
            # logger.info(f"Parameter names: {parameter_names}")

            given_tests_raw = "\n".join(problem["given_tests"]).replace(entry_point, "candidate")
            given_tests = parse_tests(given_tests_raw, parameter_names, entry_point)

            # Parse the test cases using the parameter names
            parsed_tests = parse_tests(problem["test"], parameter_names, entry_point)
            # logger.info(f"Parsed tests: {parsed_tests}")
            success_count += 1
        except:
            logger.exception(f"Error processing problem {problem['task_id']}")
            if problem['is_solved'] == False:
                unhandled_failures += 1
            continue

    logger.info(f"Success count: {success_count}")
    logger.info(f"Total problems: {len(problems)}")
    logger.info(f"Unhandled failures: {unhandled_failures}")


def parse_specific_problem(problem):
    try:
        if isinstance(problem, str):
            problem = json.loads(problem)

        logger.info(f"Problem: {problem}")
        logger.debug(f"Test: {problem['test']}")
        logger.debug(f"Given Test: {problem['given_tests']}")

        entry_point = problem["entry_point"]
        parameter_names = get_parameter_names(problem["prompt"], entry_point)
        logger.debug(f"Parameter names: {parameter_names}")

        given_tests_raw = "\n".join(problem["given_tests"]).replace(entry_point, "candidate")
        given_tests = parse_tests(given_tests_raw, parameter_names, entry_point)
        logger.debug(f"Given tests: {given_tests}")

        # Parse the test cases using the parameter names
        all_tests = parse_tests(problem["test"], parameter_names, entry_point)
        logger.debug(f"Parsed tests: {all_tests}")
        return all_tests
    except:
        logger.exception(f"Error processing problem {problem['task_id']}")
        return None

#assert next_smallest([]) is None
#assert decode_cyclic(encode_cyclic("abc")) == "abc"
#assert round(find_zero([-6, 11, -6, 1]), 2) == 1.0
#assert abs(candidate(1.33) - 0.33) < 1e-6

def check_all_problems(problems):
    problems_q = []
    success_count = 0
    fail_count = 0
    for problem in tqdm(problems):
        try:
            problem = json.loads(problem)

            logger.info(f"Problem: {problem}")
            logger.debug(f"Test: {problem['test']}")
            logger.debug(f"All Test: {problem['given_tests']}")

            entry_point = problem["entry_point"]
            parameter_names = get_parameter_names(problem["prompt"], entry_point)
            logger.info(f"Parameter names: {parameter_names}")

            # given_tests_len = len(problem["given_tests"])
            # given_tests_raw = "\n".join(problem["given_tests"]).replace(entry_point, "candidate")
            # given_tests = parse_tests(given_tests_raw, parameter_names, entry_point)
            # parsed_given_tests_len = len(given_tests['test_cases'])
            # assert given_tests_len == parsed_given_tests_len
            # success_count += 1

            #Parse the test cases using the parameter names
            tests_len_candidate =  problem["test"].count('candidate')
            parsed_tests = parse_tests(problem["test"], parameter_names, entry_point)
            parsed_test_len = len(parsed_tests['test_cases'])
            #assert parsed_test_len != 0 
            assert tests_len_candidate - 1 == parsed_test_len
            logger.info(f"Parsed tests: {parsed_tests}")
            success_count += 1
        except:
            logger.exception(f"Error processing problem {problem['task_id']}")
            if problem['is_solved'] == False:
                fail_count += 1
                problems_q.append(problem['task_id'])
            continue
    
    with open('output_data/humaneval/seed/deepseek-coder-v2-lite-instruct/20240828-174550/dscoder_debugged_seeds_deepseek-coder-v2-lite-instruct_1_1_10.jsonl', "r") as f:
        fixed = f.readlines()
    for fix_problem in fixed:
        fix_problem = json.loads(fix_problem)
        if fix_problem['task_id'] in problems_q:
            print(1)

    logger.info(f"Success count: {success_count}")
    logger.info(f"Total problems: {len(problems)}")
    logger.info(f"Unhandled failures: {fail_count}")

if __name__ == "__main__":
    input_seeds = "input_data/humaneval/seed/deepseek-coder-v2-lite-instruct/seed.jsonl"

    with open(input_seeds, "r") as f:
        problems = f.readlines()

    check_all_problems(problems)
    #parse_all_problems(problems)

    # parse the one with 'task_id': 'HumanEval/32'
    # for problem in problems:
    #     problem = json.loads(problem)
    #     if problem['task_id'] == 'HumanEval/33':
    #         parsed_tests = parse_specific_problem(problem)
    #         break