|
from copy import deepcopy |
|
from typing import Optional, Any, List, Dict |
|
|
|
from flows import logging |
|
from flows.base_flows.abstract import AtomicFlow |
|
|
|
log = logging.get_logger(__name__) |
|
|
|
|
|
class CodeTesting(AtomicFlow): |
|
REQUIRED_KEYS_CONFIG = [] |
|
REQUIRED_KEYS_CONSTRUCTOR = [] |
|
|
|
def __init__(self, **kwargs): |
|
super().__init__(**kwargs) |
|
|
|
def _get_test_data(self, input_data: Dict): |
|
"""This function retrieves (or generates) input-output pairs that will be used to test the implementation.""" |
|
raise NotImplementedError() |
|
|
|
def _run_tests(self, input_data: Dict, test_data: Dict) -> Dict[str, Any]: |
|
raise NotImplementedError() |
|
|
|
@classmethod |
|
def instantiate_from_config(cls, config): |
|
flow_config = deepcopy(config) |
|
|
|
kwargs = {"flow_config": flow_config} |
|
kwargs["input_data_transformations"] = cls._set_up_data_transformations(config["input_data_transformations"]) |
|
kwargs["output_data_transformations"] = cls._set_up_data_transformations(config["output_data_transformations"]) |
|
|
|
|
|
return cls(**kwargs) |
|
|
|
def run(self, |
|
input_data: Dict[str, Any], |
|
private_keys: Optional[List[str]] = [], |
|
keys_to_ignore_for_hash: Optional[List[str]] = []) -> Dict[str, Any]: |
|
|
|
|
|
test_data = self._get_test_data(input_data) |
|
|
|
|
|
response = self._run_tests(input_data, test_data) |
|
|
|
return response |
|
|
