def from_request(cls, request):
    """Create new TransientShardState from webapp request."""
    mapreduce_spec = MapreduceSpec.from_json_str(request.get("mapreduce_spec"))
    mapper_spec = mapreduce_spec.mapper
    input_reader_spec_dict = simplejson.loads(request.get("input_reader_state"))
    input_reader = mapper_spec.input_reader_class().from_json(
        input_reader_spec_dict)
    initial_input_reader_spec_dict = simplejson.loads(
        request.get("initial_input_reader_state"))
    initial_input_reader = mapper_spec.input_reader_class().from_json(
        initial_input_reader_spec_dict)

    output_writer = None
    if mapper_spec.output_writer_class():
      output_writer = mapper_spec.output_writer_class().from_json(
          simplejson.loads(request.get("output_writer_state", "{}")))
      assert isinstance(output_writer, mapper_spec.output_writer_class()), (
          "%s.from_json returned an instance of wrong class: %s" % (
              mapper_spec.output_writer_class(),
              output_writer.__class__))

    request_path = request.path
    base_path = request_path[:request_path.rfind("/")]

    return cls(base_path,
               mapreduce_spec,
               str(request.get("shard_id")),
               int(request.get("slice_id")),
               input_reader,
               initial_input_reader,
               output_writer=output_writer,
               retries=int(request.get("retries")))

  def from_json(json_str):
    """Deserialize KeyRange from its json representation.

    Args:
      json_str: string with json representation created by key_range_to_json.

    Returns:
      deserialized KeyRange instance.
    """
    if simplejson is None:
      raise SimplejsonUnavailableError(
          "JSON functionality requires simplejson to be available")

    def key_from_str(key_str):
      if key_str:
        return db.Key(key_str)
      else:
        return None

    json = simplejson.loads(json_str)
    return KeyRange(key_from_str(json["key_start"]),
                    key_from_str(json["key_end"]),
                    json["direction"],
                    json["include_start"],
                    json["include_end"],
                    json.get("namespace"),
                    _app=json.get("_app"))

  def setUp(self):
    """Sets up the test harness."""
    testutil.HandlerTestBase.setUp(self)

    for _ in range(100):
      TestKind().put()
    self.start = handlers.StartJobHandler()
    self.start.initialize(mock_webapp.MockRequest(),
                          mock_webapp.MockResponse())
    self.start.request.path = "/mapreduce/command/start"
    self.start.request.set("name", "my job 1")
    self.start.request.set(
        "mapper_input_reader",
        "mapreduce.input_readers.DatastoreInputReader")
    self.start.request.set("mapper_handler", "__main__.TestMap")
    self.start.request.set("mapper_params.entity_kind", "__main__.TestKind")

    self.start.request.headers["X-Requested-With"] = "XMLHttpRequest"

    self.start.post()
    result = simplejson.loads(self.start.response.out.getvalue())
    self.mapreduce_id = result["mapreduce_id"]

    self.handler = status.GetJobDetailHandler()
    self.handler.initialize(mock_webapp.MockRequest(),
                            mock_webapp.MockResponse())
    self.handler.request.path = "/mapreduce/command/list"

    self.handler.request.headers["X-Requested-With"] = "XMLHttpRequest"

  def testBasic(self):
    """Tests when there are fewer than the max results to render."""
    TestKind().put()
    self.start.request.set("name", "my job 1")
    self.start.post()
    time.sleep(.1)
    self.start.request.set("name", "my job 2")
    self.start.post()
    time.sleep(.1)
    self.start.request.set("name", "my job 3")
    self.start.post()

    self.handler.get()
    result = simplejson.loads(self.handler.response.out.getvalue())
    expected_args = set([
        "active",
        "active_shards",
        "chart_url",
        "chart_width",
        "mapreduce_id",
        "name",
        "shards",
        "start_timestamp_ms",
        "updated_timestamp_ms",
        ])
    self.assertEquals(3, len(result["jobs"]))
    self.assertEquals("my job 3", result["jobs"][0]["name"])
    self.assertEquals("my job 2", result["jobs"][1]["name"])
    self.assertEquals("my job 1", result["jobs"][2]["name"])
    self.assertEquals(expected_args, set(result["jobs"][0].keys()))
    self.assertEquals(expected_args, set(result["jobs"][1].keys()))
    self.assertEquals(expected_args, set(result["jobs"][2].keys()))

 def testBadJobId(self):
   """Tests when an invalid job ID is supplied."""
   self.handler.request.set("mapreduce_id", "does not exist")
   self.handler.get()
   result = simplejson.loads(self.handler.response.out.getvalue())
   self.assertEquals(
       {"error_message": "\"Could not find job with ID 'does not exist'\"",
        "error_class": "KeyError"},
       result)

  def from_json_str(cls, json_str):
    """Convert json string representation into class instance.

    Args:
      json_str: json representation as string.

    Returns:
      New instance of the class with data loaded from json string.
    """
    return cls.from_json(simplejson.loads(json_str, cls=JsonDecoder))

    def input_reader(self, mapper_spec):
        """Get the reader from mapper_spec initialized with the request's state.

    Args:
      mapper_spec: a mapper spec containing the immutable mapper state.

    Returns:
      An initialized InputReader.
    """
        input_reader_spec_dict = simplejson.loads(self.request.get("input_reader_state"))
        return mapper_spec.input_reader_class().from_json(input_reader_spec_dict)

  def testBeforeKickOff(self):
    """Tests getting the job details."""
    self.handler.request.set("mapreduce_id", self.mapreduce_id)
    self.handler.get()
    result = simplejson.loads(self.handler.response.out.getvalue())

    expected_keys = set([
        "active", "chart_url", "counters", "mapper_spec", "mapreduce_id",
        "name", "result_status", "shards", "start_timestamp_ms",
        "updated_timestamp_ms", "params", "hooks_class_name"])

    self.assertEquals(expected_keys, set(result.keys()))

  def testCursor(self):
    """Tests when a job cursor is present."""
    TestKind().put()
    self.start.request.set("name", "my job 1")
    self.start.post()
    self.start.request.set("name", "my job 2")
    self.start.post()

    self.handler.request.set("count", "1")
    self.handler.get()
    result = simplejson.loads(self.handler.response.out.getvalue())
    self.assertEquals(1, len(result["jobs"]))
    self.assertTrue("cursor" in result)

    self.handler.response.out.truncate(0)
    self.handler.request.set("count", "1")
    self.handler.request.set("cursor", result['cursor'])
    self.handler.get()
    result2 = simplejson.loads(self.handler.response.out.getvalue())
    self.assertEquals(1, len(result2["jobs"]))
    self.assertFalse("cursor" in result2)

    def value(self):
        """Returns the value of this Slot."""
        if hasattr(self, "_value_decoded"):
            return self._value_decoded

        if self.value_blob is not None:
            encoded_value = self.value_blob.open().read()
        else:
            encoded_value = self.value_text

        self._value_decoded = simplejson.loads(encoded_value, cls=util.JsonDecoder)
        return self._value_decoded

  def make_value_from_datastore(self, value):
    """Convert a value from its Datastore representation.

    Args:
      value: Datastore value.

    Returns:
      value to return to user code.
    """
    if value is None:
      return None
    return simplejson.loads(value)

  def make_value_from_datastore(self, value):
    """Convert value from datastore representation.

    Args:
      value: datastore value.

    Returns:
      value to store in the model.
    """

    if value is None:
      return None
    return self.data_type.from_json(simplejson.loads(value))

  def from_request(cls, request):
    """Create new TransientShardState from webapp request."""
    mapreduce_spec = MapreduceSpec.from_json_str(request.get("mapreduce_spec"))
    mapper_spec = mapreduce_spec.mapper
    input_reader_spec_dict = simplejson.loads(request.get("input_reader_state"),
                                              cls=json_util.JsonDecoder)
    input_reader = mapper_spec.input_reader_class().from_json(
        input_reader_spec_dict)
    initial_input_reader_spec_dict = simplejson.loads(
        request.get("initial_input_reader_state"), cls=json_util.JsonDecoder)
    initial_input_reader = mapper_spec.input_reader_class().from_json(
        initial_input_reader_spec_dict)

    output_writer = None
    if mapper_spec.output_writer_class():
      output_writer = mapper_spec.output_writer_class().from_json(
          simplejson.loads(request.get("output_writer_state", "{}"),
                           cls=json_util.JsonDecoder))
      assert isinstance(output_writer, mapper_spec.output_writer_class()), (
          "%s.from_json returned an instance of wrong class: %s" % (
              mapper_spec.output_writer_class(),
              output_writer.__class__))

    handler = util.try_deserialize_handler(request.get("serialized_handler"))
    if not handler:
      handler = mapreduce_spec.mapper.handler

    return cls(mapreduce_spec.params["base_path"],
               mapreduce_spec,
               str(request.get("shard_id")),
               int(request.get("slice_id")),
               input_reader,
               initial_input_reader,
               output_writer=output_writer,
               retries=int(request.get("retries")),
               handler=handler)

  def make_value_from_datastore(self, value):
    """Convert value from datastore representation.

    Args:
      value: datastore value.

    Returns:
      value to store in the model.
    """

    if value is None:
      return None
    json = simplejson.loads(value, cls=JsonDecoder)
    if self.data_type == dict:
      return json
    return self.data_type.from_json(json)

  def post(self):
    """Handles kick off request."""
    spec = model.MapreduceSpec.from_json_str(
        self._get_required_param("mapreduce_spec"))
    input_readers_json = simplejson.loads(
        self._get_required_param("input_readers"))

    queue_name = os.environ.get("HTTP_X_APPENGINE_QUEUENAME", "default")

    mapper_input_reader_class = spec.mapper.input_reader_class()
    input_readers = [mapper_input_reader_class.from_json_str(reader_json)
                     for reader_json in input_readers_json]

    KickOffJobHandler._schedule_shards(
        spec, input_readers, queue_name, self.base_path())

    ControllerCallbackHandler.reschedule(
        self.base_path(), spec, queue_name=queue_name, serial_id=0)

    def params(self):
        """Returns the dictionary of parameters for this Pipeline."""
        if hasattr(self, "_params_decoded"):
            return self._params_decoded

        if self.params_blob is not None:
            value_encoded = self.params_blob.open().read()
        else:
            value_encoded = self.params_text

        value = simplejson.loads(value_encoded, cls=util.JsonDecoder)
        if isinstance(value, dict):
            kwargs = value.get("kwargs")
            if kwargs:
                adjusted_kwargs = {}
                for arg_key, arg_value in kwargs.iteritems():
                    # Python only allows non-unicode strings as keyword arguments.
                    adjusted_kwargs[str(arg_key)] = arg_value
                value["kwargs"] = adjusted_kwargs

        self._params_decoded = value
        return self._params_decoded

 def testNoJobs(self):
   """Tests when there are no jobs."""
   self.handler.get()
   result = simplejson.loads(self.handler.response.out.getvalue())
   self.assertEquals({'jobs': []}, result)