def testGetSetContext(self):
     """Test module's get_context and _set functions."""
     ctx = context.Context(None, None)
     self.assertFalse(context.get())
     context.Context._set(ctx)
     self.assertEquals(ctx, context.get())
     context.Context._set(None)
     self.assertEquals(None, context.get())

def create_channel_assignments(entity):
    
    # Get network graph
    
    logging.info('Creating channel assignments')
    
    network_graph = entity.recreate_network_graph()
    
    # Get parameter values
    
    ctx = context.get()
    params = ctx.mapreduce_spec.mapper.params

    min_links = int(params['min_links'])
    min_neighbors = int(params['min_neighbors'])
    num_channels = int(params['num_channels'])
    
    node_selector = NodeSelector(network_graph, min_links, min_neighbors)    
    selected_nodes = node_selector.select_nodes()
    
    logging.info('Number of selected nodes: ' + str(len(selected_nodes)))
    
    channel_assignment_initializer = ChannelAssignmentInitializer(selected_nodes, num_channels)
    names_to_numbers = channel_assignment_initializer.names_to_numbers
    names_to_numbers_entity = NamesToNumbersModel(names_to_numbers=str(names_to_numbers))
    names_to_numbers_entity.put()
    
    num_nodes = len(names_to_numbers)
    
    for combination in product(range(num_channels), repeat=num_nodes):
        ca_entity = ChannelAssignmentModel(channel_assignment=str(combination))
        ca_entity.put()

  def write(self, data):
    """Write data.

    Args:
      data: actual data yielded from handler. Type is writer-specific.
    """
    ctx = context.get()
    if len(data) != 2:
      logging.error("Got bad tuple of length %d (2-tuple expected): %s",
                    len(data), data)

    try:
      key = str(data[0])
      value = str(data[1])
    except TypeError:
      logging.error("Expecting a tuple, but got %s: %s",
                    data.__class__.__name__, data)

    file_index = key.__hash__() % len(self._filehandles)

    # Work-around: Since we don't have access to the context in the to_json()
    # function, but we need to flush each pool before we serialize the
    # filehandle, we rely on a member variable instead of using context for
    # pool management.
    pool = self._pools[file_index]
    if pool is None:
      filehandle = self._filehandles[file_index]
      pool = output_writers.GCSRecordsPool(filehandle=filehandle, ctx=ctx)
      self._pools[file_index] = pool

    proto = kv_pb.KeyValue()
    proto.set_key(key)
    proto.set_value(value)
    pool.append(proto.Encode())

def build_search_index(readbuffer):
    # readbuffer should be a tuple from GoogleCloudLineInputReader composed of a
    # tuple of the form ((file_name, offset), line)

    # Get namespace from mapreduce job and set it.
    ctx = context.get()
    params = ctx.mapreduce_spec.mapper.params
    namespace = params['namespace']
    index_name = params['index_name']
    rightnow=datetime.now()
    today=rightnow.day
    if today < 10:
      day='0%s' % today
    else:
      day='%s' % today
    thismonth=rightnow.month
    if thismonth < 10:
      month='0%s' % thismonth
    else:
      month='%s' % thismonth
    indexdate='%s-%s-%s'  % (rightnow.year, month, day)

    try:
        # Get the row out of the input buffer
        row=readbuffer[1]
        # Create a dictionary from the HEADER and the row
        data = get_rec_dict(dict(zip(HEADER, row.split('\t'))))
#        logging.info('Data from %s offset %s: %s' % (readbuffer[0][0], readbuffer[0][1], data))
        # Create an index document from the row dictionary
        doc = index_record(data, indexdate)
        # Store the document in the given index
        index_doc(doc, index_name, namespace)
    except Exception, e:
        logging.error('%s\n%s' % (e, readbuffer))

  def write(self, data):
    """Write data.

    Args:
      data: actual data yielded from handler. Type is writer-specific.
    """
    ctx = context.get()
    if len(data) != 2:
      logging.error("Got bad tuple of length %d (2-tuple expected): %s",
                    len(data), data)

    try:
      key = str(data[0])
      value = str(data[1])
    except TypeError:
      logging.error("Expecting a tuple, but got %s: %s",
                    data.__class__.__name__, data)

    file_index = key.__hash__() % len(self._filenames)
    pool_name = "kv_pool%d" % file_index
    filename = self._filenames[file_index]

    if ctx.get_pool(pool_name) is None:
      ctx.register_pool(pool_name,
                        output_writers.RecordsPool(filename=filename, ctx=ctx))
    proto = file_service_pb.KeyValue()
    proto.set_key(key)
    proto.set_value(value)
    ctx.get_pool(pool_name).append(proto.Encode())

def index_map(data):
    """Index map function.

    Args:
        data: Refers to a line from the input files. It is actually composed
        of a tuple (lineinfo, line). This is the return value from the
        ZipLineInputReader, available among the input readers of mapreduce.

    Yields:
        A tuple in the format <word>_SEP<title>_SEP<character>,
        <serial_line_key>.
        line_key needs to be serialized because it is an object and reduce
        expects strings as input.
    """
    info, line = data
    if line.strip() == '':
        return
    _, file_index, offset = info
    ctx = context.get()
    params = ctx.mapreduce_spec.mapper.params
    metadata_blob = params['metadata']
    blob_reader = blobstore.BlobReader(metadata_blob[0].split('/')[-1])
    all_meta = blob_reader.read().split('\n')[:-1] # the last one is empty
    metadata = select_metadata(all_meta, file_index)
    char_map = metadata['pos_to_char']
    sorted_offsets = metadata['sorted_offsets']
    character = get_character(char_map, sorted_offsets, offset)
    title = metadata['title']
    line_db = Line(line=line)
    line_key = line_db.put()
    for word in get_words(line.lower()):
        yield (word + _SEP + title + _SEP + character, pickle.dumps(line_key))

def test_user_on_events(user):
    logging.info("Trying user %s (expired %s)", user.fb_uid, user.expired_oauth_token)
    if user.expired_oauth_token:
        return
    # TODO: Relies on us keeping these up to date!
    if not user.num_auto_added_events and not user.num_hand_added_events:
        return
    ctx = context.get()
    params = ctx.mapreduce_spec.mapper.params
    event_ids = params['event_ids'].split(',')
    fbl = user.get_fblookup()
    fbl.allow_cache = False
    try:
        fb_events = fbl.get_multi(fb_api.LookupEvent, event_ids)
    except fb_api.ExpiredOAuthToken:
        # Not longer a valid source for access_tokens
        return
    found_fb_events = [x for x in fb_events if x and not x['empty']]
    for fb_event in found_fb_events:
        yield (fb_event['info']['id'], user.fb_uid)

    # Found some good stuff, let's save and update the db events
    found_db_events = eventdata.DBEvent.get_by_ids([x['info']['id'] for x in found_fb_events])
    db_fb_events = []
    for db_event, new_fb_event in zip(found_db_events, found_fb_events):
        if db_event.has_content():
            db_fb_events.append((db_event, new_fb_event))
    event_updates.update_and_save_fb_events(db_fb_events)

def process(entity):
  ctx=context.get()
  params=ctx.mapreduce_spec.mapper.params
  dbid=params['dbid']
  viewName=params['viewName']
  viewUrl=params['nodeUrl']+'/'+viewName
  map(entity, dbid, viewName, viewUrl)

def process(org_app):
  ctx = context.get()
  params = ctx.mapreduce_spec.mapper.params

  program_type = params['program_type']
  program_key_str = params['program_key']

  # now the script is used only for GCI
  if program_type != 'gci':
    return

  program = GCIProgram.get_by_key_name(program_key_str)

  survey_query = OrgAppSurvey.all(keys_only=True).filter('program', program)
  survey_key = survey_query.get()

  # We can skip the survey records not belonging to the given program.
  if org_app.survey.key() != survey_key:
    return

  # TODO(daniel): create a MapReduce/Task RequestData
  data = MapreduceRequestData(program, Site.get_by_key_name('site'))

  absolute_url = links.ABSOLUTE_LINKER.program(
      program, gci_url_names.CREATE_GCI_ORG_PROFILE)

  if org_app.status == 'pre-accepted':
    org_app_logic.setStatus(data, org_app, 'accepted', absolute_url)
    yield operation.counters.Increment("proposals_accepted")
  elif org_app.status == 'pre-rejected':
    org_app_logic.setStatus(data, org_app, 'rejected', absolute_url)
    yield operation.counters.Increment("proposals_rejected")
  else:
    yield operation.counters.Increment("proposals_ignored")

def process(comment):
  ctx = context.get()
  params = ctx.mapreduce_spec.mapper.params
  program_key = params['program_key']

  program = GCIProgram.get_by_key_name(program_key)

  if comment.parent().program.key() != program.key():
    yield operation.counters.Increment("prev_program_comment_not_converted")
    return

  if comment.title not in ACTION_TITLES:
    yield operation.counters.Increment("user_comment_not_converted")
    return

  comment_title = ACTION_TITLES[comment.title]

  changes = ACTION_TITLES[comment_title]
  # Task reopening is a special case which could have been performed
  # either by a mentor or by the automated system after the passing of
  # the deadline. So additional inference of the user has to be made.
  if comment_title == 'Task Reopened':
    if comment.created_by:
      user_info = ugettext('User-Mentor')
    else:
      user_info = ugettext('MelangeAutomatic')
    changes = [user_info] + changes

  comment.changes = changes

  yield operation.db.Put(comment)
  yield operation.counters.Increment("action_comment_converted")

def process(task):
  ctx = context.get()
  params = ctx.mapreduce_spec.mapper.params
  program_key = params['program_key']

  try:
    program = GCIProgram.get_by_key_name(program_key)
  except db.BadValueError:
    yield operation.counters.Increment('program_key_is_empty_or_invalid')
    return

  def subscribe_to_task_txn(task_key, subscribe):
    task = GCITask.get(task_key)
    task.subscribers = list(set(task.subscribers + subscribe))
    task.put()
    return task

  if task.program.key() != program.key():
    yield operation.counters.Increment("old_program_task_not_updated")
    return

  mentors = db.get(task.mentors)
  entities = mentors + [task.created_by, task.modified_by]

  subscribe = [ent.key() for ent in entities if ent.automatic_task_subscription]

  result = db.run_in_transaction(subscribe_to_task_txn, task.key(), subscribe)

  if result:
    yield operation.counters.Increment("task_updated")
  else:
    yield operation.counters.Increment("task_not_updated")

 def __call__(self, entity):
   self.slice_count += 1
   yield "%s\n" % entity.int_property
   slice_id = context.get()._shard_state.slice_id
   # Raise exception when processing the 2 item in a slice every 3 slices.
   if (self.slice_count == 2 and
       (slice_id + 1) % 3 == 0):
     raise Exception("Intentionally raise an exception")

  def run(self, event):
    if not self.oldest_last_modified:
      params = context.get().mapreduce_spec.mapper.params
      self.oldest_last_modified = datetime.datetime.utcfromtimestamp(
          params['oldest_last_modified'])

    if event.last_modified < self.oldest_last_modified:
      yield op.db.Delete(event)

def yield_post_jp_event(db_events):
    from mapreduce import context
    ctx = context.get()
    params = ctx.mapreduce_spec.mapper.params
    token_nickname = params.get('token_nickname')
    db_events = [x for x in db_events if x.actual_city_name and x.actual_city_name.endswith('Japan')]
    for db_event in db_events:
        pubsub.eventually_publish_event(db_event.id, token_nickname)

def word_count_reduce(key, values):
  """Word count reduce function."""
  sentences = []
  ctx = context.get()
  title = ctx.mapreduce_spec.mapper.params['output_writer']['book_title']
  for sentence in values:
    sentences.append(models.Sentence(sentence=sentence, book=title))
  word = models.Word(word=key, sentences=sentences)
  word.put()

def denorm_entity_mapper(entity):

    ctx = context.get()
    params = ctx.mapreduce_spec.mapper.params

    entity._denorm_values = params['denorm_values']

    # Instead of naive single save: entity.save(), do the following more efficient batch save:
    yield _batch_save(entity)

def log2csv(l):
  """Convert log API RequestLog object to csv."""
  root_pipeline_id = context.get().mapreduce_spec.mapper.params['root_pipeline_id']
  message(root_pipeline_id, '<span class="label label-warning">pending</span> MapperPipeline.log2csv')
  yield '%s,%s,%s,%s,%s,%s,%s,%s,%s,%s\n' % (l.start_time, l.method, l.resource,
                                             l.status, l.latency, l.response_size,
                                             l.was_loading_request, l.cost,
                                             '"%s"' % l.user_agent if l.user_agent else "NULL",
                                             l.nickname if l.nickname else "NULL")

 def map(item):
     mapper_params = context.get().mapreduce_spec.mapper.params
     user_ids_to_remove = set(mapper_params["user_ids"])
     item_user_ids = set(item.get_user_ids())
     matching = item_user_ids.intersection(user_ids_to_remove)
     if matching:
         item.delete()
         for user_id in matching:
             yield user_id, 1

 def map(student):
     params = context.get().mapreduce_spec.mapper.params
     ns = params['course_namespace']
     app_context = sites.get_course_index().get_app_context_for_namespace(ns)
     course = courses.Course(None, app_context=app_context)
     if student_is_qualified(student, course):
         yield(TOTAL_CERTIFICATES, 1)
     if student.scores:
         yield(TOTAL_ACTIVE_STUDENTS, 1)
     yield(TOTAL_STUDENTS, 1)

  def write(self, data):
    """Write data.

    Args:
      data: actual data yielded from handler. Type is writer-specific.
    """
    ctx = context.get()
    if ctx.get_pool("file_pool") is None:
      ctx.register_pool("file_pool", _FilePool(ctx=ctx))
    ctx.get_pool("file_pool").append(self._filename, str(data))