File size: 3,185 Bytes
23272f4 e81ee32 23272f4 e81ee32 23272f4 e81ee32 23272f4 e81ee32 23272f4 e81ee32 23272f4 e81ee32 23272f4 |
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 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 |
import os
import time
import subprocess
from typing import Dict, Any
from flow_modules.aiflows.InterpreterFlowModule import InterpreterAtomicFlow
class CodeTestingAtomicFlow(InterpreterAtomicFlow):
"""This class inherits from InterpreterAtomicFlow and is used to test code.
*Input Interface*:
- `temp_code_file_location`: Location of the file containing the code to be tested.
*Output Interface*:
- `feedback`: Feedback from the test (i.e. test results).
*Configuration Parameters*:
- `input_interface`: the input interface of the atomic flow.
- `output_interface`: the output interface of the atomic flow.
"""
def _open_file_and_wait_for_file_update(self, file_location, check_interval=1):
"""Opens the file in VSCode and waits for the file to be updated.
:param file_location: The location of the file to be opened.
:type file_location: str
:param check_interval: The interval at which the file is checked for updates.
:type check_interval: int
"""
process = subprocess.Popen(["code", "--wait", file_location])
while True:
if process.poll() is not None:
break
time.sleep(check_interval)
def _prepare_code_and_lang(self, input_data: Dict[str, Any]):
"""Prepares the code and language to be passed to the interpreter.
:param input_data: The input data to the atomic flow.
:type 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]):
"""Checks if the input data contains the required keys.
:param input_data: The input data to the atomic flow.
:type input_data: Dict[str, Any]
:raises AssertionError: If the input data does not contain the required keys.
"""
assert "temp_code_file_location" in input_data, "temp_code_file_location not passed to CodeTestingAtomicFlow"
def _delete_file(self, file_location):
"""Deletes the file at the given location.
:param file_location: The location of the file to be deleted.
:type file_location: str
"""
if os.path.exists(file_location):
os.remove(file_location)
def run(
self,
input_data: Dict[str, Any]):
"""Runs the atomic flow.
:param input_data: The input data to the atomic flow.
:type input_data: Dict[str, Any]
:return: The output data of the atomic flow.
:rtype: Dict[str, Any]
"""
self._check_input(input_data)
file_loc = input_data["temp_code_file_location"]
self._open_file_and_wait_for_file_update(file_loc)
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}
|