Test a specific endpoint directly
(self, params: Union[str, dict], endpoint: str, timeout: Optional[int] = 30, **kwargs)
| 112 | return f"[Python Interpreter Error]: {str(e)}" |
| 113 | |
| 114 | def call_specific_endpoint(self, params: Union[str, dict], endpoint: str, timeout: Optional[int] = 30, **kwargs) -> tuple: |
| 115 | """Test a specific endpoint directly""" |
| 116 | try: |
| 117 | if type(params) is str: |
| 118 | params = json5.loads(params) |
| 119 | code = params.get('code', '') |
| 120 | if not code: |
| 121 | code = params.get('raw', '') |
| 122 | triple_match = re.search(r'```[^\n]*\n(.+?)```', code, re.DOTALL) |
| 123 | if triple_match: |
| 124 | code = triple_match.group(1) |
| 125 | except Exception: |
| 126 | code = extract_code(params) |
| 127 | |
| 128 | if not code.strip(): |
| 129 | return False, '[Python Interpreter Error]: Empty code.' |
| 130 | |
| 131 | try: |
| 132 | start_time = time.time() |
| 133 | code_result = run_code(RunCodeRequest(code=code, language='python', run_timeout=timeout), |
| 134 | max_attempts=1, client_timeout=timeout, endpoint=endpoint) |
| 135 | end_time = time.time() |
| 136 | |
| 137 | result = [] |
| 138 | if code_result.run_result.stdout: |
| 139 | result.append(f"stdout:\n{code_result.run_result.stdout}") |
| 140 | if code_result.run_result.stderr: |
| 141 | result.append(f"stderr:\n{code_result.run_result.stderr}") |
| 142 | |
| 143 | result = '\n'.join(result) |
| 144 | execution_time = end_time - start_time |
| 145 | return True, result if result.strip() else 'Finished execution.', execution_time |
| 146 | |
| 147 | except Timeout as e: |
| 148 | return False, f'[Python Interpreter Error] TimeoutError: Execution timed out.', None |
| 149 | except Exception as e: |
| 150 | return False, f'[Python Interpreter Error]: {str(e)}', None |
nothing calls this directly
no test coverage detected