File size: 1,753 Bytes
e8c208d 2c14431 e8c208d aef98f4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 |
import os
import time
from typing import Dict, Any
from flow_modules.Tachi67.InterpreterFlowModule import InterpreterAtomicFlow
class CodeTestingAtomicFlow(InterpreterAtomicFlow):
def _wait_for_file_update(self, file_location, last_check_time, check_interval=1):
while True:
time.sleep(check_interval)
current_time = os.path.getmtime(file_location)
if current_time != last_check_time:
break
def _prepare_code_and_lang(self, input_data: Dict[str, Any]):
file_location = input_data["temp_code_file_location"]
input_data["language"] = "python"
with open(file_location, "r") as file:
code_str = file.read()
input_data["code"] = code_str
def _check_input(self, input_data: Dict[str, Any]):
assert "temp_code_file_location" in input_data, "temp_code_file_location not passed to CodeTestingAtomicFlow"
assert "temp_code_file_written_timestamp" in input_data, "temp_code_file_written_timestamp not passed to CodeTestingAtomicFlow"
def _delete_file(self, file_location):
if os.path.exists(file_location):
os.remove(file_location)
def run(
self,
input_data: Dict[str, Any]):
self._check_input(input_data)
file_loc = input_data["temp_code_file_location"]
last_timestamp = input_data["temp_code_file_written_timestamp"]
self._wait_for_file_update(file_loc, last_timestamp)
self._prepare_code_and_lang(input_data)
self._process_input_data(input_data)
response = self._call()
self._delete_file(file_loc)
if response.strip() == '':
response = "test passed"
return {"feedback": response}
|