def init(self,
           mapper_handler_spec=MAPPER_HANDLER_SPEC,
           mapper_parameters=None,
           hooks_class_name=None,
           output_writer_spec=None,
           shard_count=8,
           mr_params=None,
           shard_retries=0):
    InputReader.reset()

    self.mapreduce_id = "mapreduce0"
    self.mapreduce_spec = self.create_mapreduce_spec(
        self.mapreduce_id,
        shard_count,
        mapper_handler_spec=mapper_handler_spec,
        hooks_class_name=hooks_class_name,
        output_writer_spec=output_writer_spec,
        mapper_parameters=mapper_parameters)
    if mr_params:
      self.mapreduce_spec.params.update(mr_params)

    self.shard_number = 1
    self.slice_id = 0
    self.shard_state = self.create_and_store_shard_state(
        self.mapreduce_id, self.shard_number)
    self.shard_state.retries = shard_retries
    self.shard_state.slice_id = self.slice_id
    self.shard_state.put()
    self.shard_id = self.shard_state.shard_id

    output_writer = None
    if self.mapreduce_spec.mapper.output_writer_class():
      output_writer_cls = self.mapreduce_spec.mapper.output_writer_class()
      output_writer = output_writer_cls.create(self.mapreduce_spec,
                                               self.shard_number,
                                               shard_retries + 1)

    reader_iter = db_iters.RangeIteratorFactory.create_key_ranges_iterator(
        key_ranges.KeyRangesFactory.create_from_list([key_range.KeyRange()]),
        model.QuerySpec("ENTITY_KIND", model_class_path=ENTITY_KIND),
        db_iters.KeyRangeModelIterator)
    self.transient_state = model.TransientShardState(
        "/mapreduce",
        self.mapreduce_spec,
        self.shard_id,
        self.slice_id,
        InputReader(reader_iter),
        InputReader(reader_iter),
        output_writer=output_writer,
        retries=shard_retries)

    self.handler = handlers.MapperWorkerCallbackHandler()
    self.handler._time = MockTime.time
    request = mock_webapp.MockRequest()
    request.headers["X-AppEngine-QueueName"] = "default"
    request.headers["X-AppEngine-TaskName"] = "foo-task-name"
    request.path = "/mapreduce/worker_callback/" + self.shard_id
    request.headers[util._MR_ID_TASK_HEADER] = self.mapreduce_id
    request.headers[util._MR_SHARD_ID_TASK_HEADER] = self.shard_id
    request.headers[model.HugeTask.PAYLOAD_VERSION_HEADER] = (
        model.HugeTask.PAYLOAD_VERSION)

    self.request = request
    self.response = mock_webapp.MockResponse()
    self.handler.initialize(self.request, self.response)

    worker_params = self.transient_state.to_dict()
    for param_name in worker_params:
      self.handler.request.set(param_name, worker_params[param_name])
Init everything needed for testing worker callbacks.
dataset/ETHPy150Open GoogleCloudPlatform/appengine-mapreduce/python/test/mapreduce/handlers_test.py/MapperWorkerCallbackHandlerTest.init