CodeTesting.py

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"])

        # ~~~ Instantiate flow ~~~
        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]:

        # ~~~ Retrieve the test data ~~~
        test_data = self._get_test_data(input_data)

        # ~~~ Run tests ~~~
        response = self._run_tests(input_data, test_data)

        return response

# from typing import Optional, Any, List, Dict
#
# from flows.base_flows.abstract import AtomicFlow
# from flows.utils.general_helpers import validate_parameters
#
# class CodeTester(AtomicFlow):
#     REQUIRED_KEYS_CONFIG = []
#     REQUIRED_KEYS_CONSTRUCTOR = []
#
#     def __init__(self, **kwargs):
#         super().__init__(**kwargs)
#
#     @classmethod
#     def _validate_parameters(cls, kwargs):
#         validate_parameters(cls, kwargs)  # is this necessary?
#
#     def run(self,
#             input_data: Dict[str, Any],
#             private_keys: Optional[List[str]] = [],
#             keys_to_ignore_for_hash: Optional[List[str]] = []) -> Dict[str, Any]: