import itertools
from io import BytesIO
from itertools import product
from typing import Callable, List, Sequence, Any, Dict

from mrjob.job import MRJob
from approvaltests import verify, verify_all_combinations


def verify_map_reduce(mr_job_under_test: MRJob, test_data: str) -> None:
    storyboard = print_map_reduce_job(mr_job_under_test, test_data)
    verify(storyboard)


def print_map_reduce_job(mr_job_under_test: MRJob, test_data: str) -> str:
    storyboard = f"{test_data}\n\nMap reduces to:\n\n"
    mr_job_under_test.sandbox(stdin=BytesIO(test_data.encode("utf-8")))
    with mr_job_under_test.make_runner() as runner:
        runner.run()
        results = mr_job_under_test.parse_output(runner.cat_output())
        for key, value in sorted(results):
            storyboard += f"{key}:{value}\n"
    return storyboard


def verify_templated_map_reduce(map_reduction: MRJob, input_creator: Callable[[Sequence[Any]], str],
                                params: Sequence[Any]) -> None:
    def map_reducer_creator(*args):
        return map_reduction

    verify_templated_map_reduce_with_customized_job(map_reducer_creator, input_creator, params)


def verify_templated_map_reduce_with_customized_job(map_reduce_creator: Callable[[Sequence[Any]], MRJob],
                                                    input_creator: Callable[[Sequence[Any]], str],
                                                    params: Sequence[Sequence[Any]]) -> None:
    inputs = product(*params)
    storyboard = ""
    for input in inputs:
        storyboard += f"===================\n\n{input} =>\n"
        data = input_creator(*input)
        map_reduction = map_reduce_creator(*input)

        storyboard += f"{print_map_reduce_job(map_reduction, data)}\n"
    verify(storyboard)


def verify_templated_map_reduce_with_customized_job_with_dictionary_args(
        map_reduce_creator: Callable[[Dict[str, Any]], MRJob],
        input_creator: Callable[[Dict[str, Any]], str],
        params: Dict[str, Sequence[Any]]) -> None:

    def product_dict(**kwargs):
        keys = kwargs.keys()
        vals = kwargs.values()
        for instance in itertools.product(*vals):
            yield dict(zip(keys, instance))

    inputs = product_dict(**params)
    storyboard = ""
    for input in inputs:
        storyboard += f"===================\n\n{input} =>\n"
        data = input_creator(input)
        map_reduction = map_reduce_creator(input)

        storyboard += f"{print_map_reduce_job(map_reduction, data)}\n"
    verify(storyboard)

@@ -1496,124 +1496,195 @@
     
-storyboard = %22%22%0A    for input in inputs:%0A        storyboard += f%22===================%5Cn%5Cn%7Binput%7D =%3E%5Cn%22%0A        data =
+verify_templated_map_reduce_with_customized_job_with_dictionary_args2(lambda i: map_reduce_creator(*i),%0A                                                                          lambda i:
  inp
@@ -1699,159 +1699,186 @@
 or(*
+i), 
 input
+s
 )%0A
-        map_reduction = map_reduce_creator(*input)%0A%0A        storyboard += f%22%7Bprint_map_reduce_job(map_reduction, data)%7D%5Cn%22%0A    verify(storyboard
+%0A%0Adef product_dict(**kwargs):%0A    keys = kwargs.keys()%0A    vals = kwargs.values()%0A    for instance in itertools.product(*vals):%0A        yield dict(zip(keys, instance)
 )%0A%0A%0A
@@ -2121,24 +2121,28 @@
 %3E None:%0A
-%0A
     
-def
+inputs =
  product
@@ -2153,207 +2153,285 @@
 t(**
-kwarg
+param
 s)
-:
 %0A    
-    keys = kwargs.keys()%0A        vals = kwargs.values()%0A        f
+verify_templated_map_reduce_with_customized_job_with_dictionary_args2(map_reduce_creat
 or
+,
  in
-stance in itertools.product(*vals):%0A            yield dict(zip(keys, instance))%0A%0A    inputs = product_dict(**params)
+put_creator, inputs)%0A%0A%0Adef verify_templated_map_reduce_with_customized_job_with_dictionary_args2(%0A        map_reduce_creator,%0A        input_creator,%0A        inputs) -%3E None:
 %0A   

R!! remove duplication