#!/usr/bin/env python
#
# Copyright 2011 Google Inc. All Rights Reserved.

"""Testing mapreduce functionality end to end."""




import datetime
import logging
import random
import string
import unittest

from google.appengine.ext import ndb

from google.appengine.api import files
from google.appengine.ext import db
from mapreduce import context
from mapreduce import control
from mapreduce import handlers
from mapreduce import model
from mapreduce import output_writers
from mapreduce import parameters
from mapreduce import records
from mapreduce import test_support
from testlib import testutil
from mapreduce.tools import gcs_file_seg_reader


# pylint: disable=g-import-not-at-top
try:
  import cloudstorage
  # In 25 runtime, the above code will be scrubbed to import the stub version
  # of cloudstorage. All occurences of the following if condition in MR
  # codebase is to tell it apart.
  # TODO(user): Remove after 25 runtime MR is abondoned.
  if hasattr(cloudstorage, "_STUB"):
    cloudstorage = None
  from cloudstorage import storage_api
except ImportError:
  cloudstorage = None  # CloudStorage library not available

# pylint: disable=g-bad-name


def random_string(length):
  """Generate a random string of given length."""
  return "".join(
      random.choice(string.letters + string.digits) for _ in range(length))


class TestEntity(db.Model):
  """Test entity class."""
  int_property = db.IntegerProperty()
  dt = db.DateTimeProperty(default=datetime.datetime(2000, 1, 1))


class NdbTestEntity(ndb.Model):
  """Test entity class for NDB."""


class TestHandler(object):
  """Test handler which stores all processed entities keys.

  Properties:
    processed_entites: all processed entities.
  """

  processed_entites = []

  def __call__(self, entity):
    """Main handler process function.

    Args:
      entity: entity to process.
    """
    TestHandler.processed_entites.append(entity)

  @staticmethod
  def reset():
    """Clear processed_entites & reset delay to 0."""
    TestHandler.processed_entites = []


class SerializableHandler(object):
  """Handler that utilize serialization."""

  _next_instance_id = 0
  # The first few instances will keep raising errors.
  # This is to test that upon shard retry, shard creates a new handler.
  INSTANCES_THAT_RAISE_ERRORS = 3
  # The first instance is created for validation and not used by any shard.
  FAILURES_INDUCED_BY_INSTANCE = INSTANCES_THAT_RAISE_ERRORS - 1

  def __init__(self):
    self.count = 0
    self.instance = self.__class__._next_instance_id
    self.__class__._next_instance_id += 1

  def __call__(self, entity):
    if self.instance < self.INSTANCES_THAT_RAISE_ERRORS:
      raise files.FinalizationError("Injected error.")
    # Increment the int property by one on every call.
    entity.int_property = self.count
    entity.put()
    self.count += 1

  @classmethod
  def reset(cls):
    cls._next_instance_id = 0


def test_handler_yield_key(entity):
  """Test handler which yields entity key."""
  yield entity.key()


def test_handler_yield_ndb_key(entity):
  """Test handler which yields entity key (NDB version)."""