CC_flows / CodeTesting.py

Refactoring.

1a3765a about 3 years ago

1.43 kB

from copy import deepcopy
from typing import Optional, Any, List, Dict

from flows import logging
from flows.base_flows 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]) -> Dict[str, Any]:
        # ~~~ Retrieve the test data ~~~
        test_data = self._get_test_data(input_data)

        # ~~~ Run tests ~~~
        response: Dict[str, Any] = self._run_tests(input_data, test_data)

        return response

	from copy import deepcopy
	from typing import Optional, Any, List, Dict

	from flows import logging
	from flows.base_flows 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]) -> Dict[str, Any]:
	# ~~~ Retrieve the test data ~~~
	test_data = self._get_test_data(input_data)

	# ~~~ Run tests ~~~
	response: Dict[str, Any] = self._run_tests(input_data, test_data)

	return response