files: # Programmer: Navraj Chohan <nlake44@gmail.com>
  
  """
   Cassandra Interface for AppScale
  """
  import cassandra
  import logging
  import sys
  import time
  
  from cassandra.cluster import Cluster
  from cassandra.concurrent import execute_concurrent
  from cassandra.policies import FallthroughRetryPolicy
  from cassandra.policies import RetryPolicy
  from cassandra.query import BatchStatement
  from cassandra.query import ConsistencyLevel
  from cassandra.query import SimpleStatement
  from cassandra.query import ValueSequence
  from .. import dbconstants
  from .. import helper_functions
  from ..dbconstants import AppScaleDBConnectionError
  from ..dbconstants import Operations
  from ..dbconstants import TxnActions
  from ..dbinterface import AppDBInterface
  from ..unpackaged import APPSCALE_LIB_DIR
  from ..utils import clean_app_id
  from ..utils import encode_index_pb
  from ..utils import get_composite_index_keys
  from ..utils import get_composite_indexes_rows
  from ..utils import get_entity_key
  from ..utils import get_entity_kind
  from ..utils import get_index_kv_from_tuple
  from ..utils import get_kind_key
  
  sys.path.append(APPSCALE_LIB_DIR)
  import appscale_info
  
  # The directory Cassandra is installed to.
  CASSANDRA_INSTALL_DIR = '/opt/cassandra'
  
  # Full path for the nodetool binary.
  NODE_TOOL = '{}/cassandra/bin/nodetool'.format(CASSANDRA_INSTALL_DIR)
  
  # The keyspace used for all tables
  KEYSPACE = "Keyspace1"
  
  # Cassandra watch name.
  CASSANDRA_MONIT_WATCH_NAME = "cassandra-9999"
  
  # The number of times to retry connecting to Cassandra.
  INITIAL_CONNECT_RETRIES = 20
  
  # The data layout version that the datastore expects.
  EXPECTED_DATA_VERSION = 1.0
  
  # The metadata key for the data layout version.
  VERSION_INFO_KEY = 'version'
  
  # The metadata key indicating that the database has been primed.
  PRIMED_KEY = 'primed'
  
  # The size in bytes that a batch must be to use the batches table.
  LARGE_BATCH_THRESHOLD = 5 << 10
  
  
  def batch_size(batch):
    """ Calculates the size of a batch.
  
    Args:
      batch: A list of dictionaries representing mutations.
    Returns:
      An integer specifying the size in bytes of the batch.
    """
    size = 0
    for mutation in batch:
      size += len(mutation['key'])
      if 'values' in mutation:
        for value in mutation['values'].values():
          size += len(value)
    return size
  
  
  def deletions_for_entity(entity, composite_indices=()):
    """ Get a list of deletions needed across tables for deleting an entity.
  
    Args:
      entity: An entity object.
      composite_indices: A list or tuple of composite indices.
    Returns:
      A list of dictionaries representing mutation operations.
    """
    deletions = []
    app_id = clean_app_id(entity.key().app())
    namespace = entity.key().name_space()
    prefix = dbconstants.KEY_DELIMITER.join([app_id, namespace])
  
    asc_rows = get_index_kv_from_tuple([(prefix, entity)])
    for entry in asc_rows:
      deletions.append({'table': dbconstants.ASC_PROPERTY_TABLE,
                        'key': entry[0],
                        'operation': Operations.DELETE})
  
    dsc_rows = get_index_kv_from_tuple(
      [(prefix, entity)], reverse=True)
    for entry in dsc_rows:
      deletions.append({'table': dbconstants.DSC_PROPERTY_TABLE,
                        'key': entry[0],
                        'operation': Operations.DELETE})
  
    for key in get_composite_indexes_rows([entity], composite_indices):
      deletions.append({'table': dbconstants.COMPOSITE_TABLE,
                        'key': key,
                        'operation': Operations.DELETE})
  
    entity_key = get_entity_key(prefix, entity.key().path())
    deletions.append({'table': dbconstants.APP_ENTITY_TABLE,
                      'key': entity_key,
                      'operation': Operations.DELETE})
  
    kind_key = get_kind_key(prefix, entity.key().path())
    deletions.append({'table': dbconstants.APP_KIND_TABLE,
                      'key': kind_key,
                      'operation': Operations.DELETE})
  
    return deletions
  
  
  def index_deletions(old_entity, new_entity, composite_indices=()):
    """ Get a list of index deletions needed for updating an entity. For changing
    an existing entity, this involves examining the property list of both
    entities to see which index entries need to be removed.
  
    Args:
      old_entity: An entity object.
      new_entity: An entity object.
      composite_indices: A list or tuple of composite indices.
    Returns:
      A list of dictionaries representing mutation operations.
    """
    deletions = []
    app_id = clean_app_id(old_entity.key().app())
    namespace = old_entity.key().name_space()
    kind = get_entity_kind(old_entity.key())
    entity_key = str(encode_index_pb(old_entity.key().path()))
  
    new_props = {}
    for prop in new_entity.property_list():
      if prop.name() not in new_props:
        new_props[prop.name()] = []
      new_props[prop.name()].append(prop)
  
    changed_props = {}
    for prop in old_entity.property_list():
      if prop.name() in new_props and prop in new_props[prop.name()]:
        continue
  
      if prop.name() not in changed_props:
        changed_props[prop.name()] = []
      changed_props[prop.name()].append(prop)
  
      value = str(encode_index_pb(prop.value()))
  
      key = dbconstants.KEY_DELIMITER.join(
        [app_id, namespace, kind, prop.name(), value, entity_key])
      deletions.append({'table': dbconstants.ASC_PROPERTY_TABLE,
                        'key': key,
                        'operation': Operations.DELETE})
  
      reverse_key = dbconstants.KEY_DELIMITER.join(
        [app_id, namespace, kind, prop.name(),
         helper_functions.reverse_lex(value), entity_key])
      deletions.append({'table': dbconstants.DSC_PROPERTY_TABLE,
                        'key': reverse_key,
                        'operation': Operations.DELETE})
  
    changed_prop_names = set(changed_props.keys())
    for index in composite_indices:
      if index.definition().entity_type() != kind:
        continue
  
      index_props = set(prop.name() for prop
                        in index.definition().property_list())
      if index_props.isdisjoint(changed_prop_names):
        continue
  
      old_entries = set(get_composite_index_keys(index, old_entity))
      new_entries = set(get_composite_index_keys(index, new_entity))
      for entry in (old_entries - new_entries):
        deletions.append({'table': dbconstants.COMPOSITE_TABLE,
                          'key': entry,
                          'operation': Operations.DELETE})
  
    return deletions
  
  
  def mutations_for_entity(entity, txn, current_value=None,
                           composite_indices=()):
    """ Get a list of mutations needed across tables for an entity change.
  
    Args:
      entity: An entity object.
      txn: A transaction ID handler.
      current_value: The entity object currently stored.
      composite_indices: A list of composite indices for the entity kind.
    Returns:
      A list of dictionaries representing mutations.
    """
    mutations = []
    if current_value is not None:
      mutations.extend(
        index_deletions(current_value, entity, composite_indices))
  
    app_id = clean_app_id(entity.key().app())
    namespace = entity.key().name_space()
    encoded_path = str(encode_index_pb(entity.key().path()))
    prefix = dbconstants.KEY_DELIMITER.join([app_id, namespace])
    entity_key = dbconstants.KEY_DELIMITER.join([prefix, encoded_path])
    entity_value = {dbconstants.APP_ENTITY_SCHEMA[0]: entity.Encode(),
                    dbconstants.APP_ENTITY_SCHEMA[1]: str(txn)}
    mutations.append({'table': dbconstants.APP_ENTITY_TABLE,
                      'key': entity_key,
                      'operation': Operations.PUT,
                      'values': entity_value})
  
    reference_value = {'reference': entity_key}
  
    kind_key = get_kind_key(prefix, entity.key().path())
    mutations.append({'table': dbconstants.APP_KIND_TABLE,
                      'key': kind_key,
                      'operation': Operations.PUT,
                      'values': reference_value})
  
    asc_rows = get_index_kv_from_tuple([(prefix, entity)])
    for entry in asc_rows:
      mutations.append({'table': dbconstants.ASC_PROPERTY_TABLE,
                        'key': entry[0],
                        'operation': Operations.PUT,
                        'values': reference_value})
  
    dsc_rows = get_index_kv_from_tuple([(prefix, entity)], reverse=True)
    for entry in dsc_rows:
      mutations.append({'table': dbconstants.DSC_PROPERTY_TABLE,
                        'key': entry[0],
                        'operation': Operations.PUT,
                        'values': reference_value})
  
    for key in get_composite_indexes_rows([entity], composite_indices):
      mutations.append({'table': dbconstants.COMPOSITE_TABLE,
                        'key': key,
                        'operation': Operations.PUT,
                        'values': reference_value})
  
    return mutations
  
  
  class FailedBatch(Exception):
    pass
  
  
  class IdempotentRetryPolicy(RetryPolicy):
    """ A policy used for retrying idempotent statements. """
    def on_read_timeout(self, query, consistency, required_responses,
                        received_responses, data_retrieved, retry_num):
      """ This is called when a ReadTimeout occurs.
  
      Args:
        query: A statement that timed out.
        consistency: The consistency level of the statement.
        required_responses: The number of responses required.
        received_responses: The number of responses received.
        data_retrieved: Indicates whether any responses contained data.
        retry_num: The number of times the statement has been tried.
      """
      if retry_num >= 5:
        return self.RETHROW, None
      else:
        return self.RETRY, consistency
  
    def on_write_timeout(self, query, consistency, write_type,
                         required_responses, received_responses, retry_num):
      """ This is called when a WriteTimeout occurs.
  
      Args:
        query: A statement that timed out.
        consistency: The consistency level of the statement.
        required_responses: The number of responses required.
        received_responses: The number of responses received.
        data_retrieved: Indicates whether any responses contained data.
        retry_num: The number of times the statement has been tried.
        """
      if retry_num >= 5:
        return self.RETHROW, None
      else:
        return self.RETRY, consistency
  
  
  class ThriftColumn(object):
    """ Columns created by default with thrift interface. """
    KEY = 'key'
    COLUMN_NAME = 'column1'
    VALUE = 'value'
  
  
  class DatastoreProxy(AppDBInterface):
    """ 
      Cassandra implementation of the AppDBInterface
    """
    def __init__(self, log_level=logging.INFO):
      """
      Constructor.
      """
      class_name = self.__class__.__name__
      self.logger = logging.getLogger(class_name)
      self.logger.setLevel(log_level)
      self.logger.info('Starting {}'.format(class_name))
  
      self.hosts = appscale_info.get_db_ips()
      self.retry_policy = IdempotentRetryPolicy()
      self.no_retries = FallthroughRetryPolicy()
  
      remaining_retries = INITIAL_CONNECT_RETRIES
      while True:
        try:
          self.cluster = Cluster(self.hosts,
                                 default_retry_policy=self.retry_policy)
          self.session = self.cluster.connect(KEYSPACE)
          break
        except cassandra.cluster.NoHostAvailable as connection_error:
          remaining_retries -= 1
          if remaining_retries < 0:
            raise connection_error
          time.sleep(3)
  
      self
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/cassandra_env/cassandra_interface.py
files: # Programmer: Navraj Chohan <nlake44@gmail.com>
  
  """
   Cassandra Interface for AppScale
  """
  import cassandra
  import logging
  import sys
  import time
  
  from cassandra.cluster import Cluster
  from cassandra.concurrent import execute_concurrent
  from cassandra.policies import FallthroughRetryPolicy
  from cassandra.policies import RetryPolicy
  from cassandra.query import BatchStatement
  from cassandra.query import ConsistencyLevel
  from cassandra.query import SimpleStatement
  from cassandra.query import ValueSequence
  from .. import dbconstants
  from .. import helper_functions
  from ..dbconstants import AppScaleDBConnectionError
  from ..dbconstants import TxnActions
  from ..dbinterface import AppDBInterface
  from ..unpackaged import APPSCALE_LIB_DIR
  from ..utils import clean_app_id
  from ..utils import encode_index_pb
  from ..utils import get_composite_index_keys
  from ..utils import get_composite_indexes_rows
  from ..utils import get_entity_key
  from ..utils import get_entity_kind
  from ..utils import get_index_kv_from_tuple
  from ..utils import get_kind_key
  
  sys.path.append(APPSCALE_LIB_DIR)
  import appscale_info
  
  # The directory Cassandra is installed to.
  CASSANDRA_INSTALL_DIR = '/opt/cassandra'
  
  # Full path for the nodetool binary.
  NODE_TOOL = '{}/cassandra/bin/nodetool'.format(CASSANDRA_INSTALL_DIR)
  
  # The keyspace used for all tables
  KEYSPACE = "Keyspace1"
  
  # Cassandra watch name.
  CASSANDRA_MONIT_WATCH_NAME = "cassandra-9999"
  
  # The number of times to retry connecting to Cassandra.
  INITIAL_CONNECT_RETRIES = 20
  
  # The data layout version that the datastore expects.
  EXPECTED_DATA_VERSION = 1.0
  
  # The metadata key for the data layout version.
  VERSION_INFO_KEY = 'version'
  
  # The metadata key indicating that the database has been primed.
  PRIMED_KEY = 'primed'
  
  # The size in bytes that a batch must be to use the batches table.
  LARGE_BATCH_THRESHOLD = 5 << 10
  
  
  def batch_size(batch):
    """ Calculates the size of a batch.
  
    Args:
      batch: A list of dictionaries representing mutations.
    Returns:
      An integer specifying the size in bytes of the batch.
    """
    size = 0
    for mutation in batch:
      size += len(mutation['key'])
      if 'values' in mutation:
        for value in mutation['values'].values():
          size += len(value)
    return size
  
  
  def deletions_for_entity(entity, composite_indices=()):
    """ Get a list of deletions needed across tables for deleting an entity.
  
    Args:
      entity: An entity object.
      composite_indices: A list or tuple of composite indices.
    Returns:
      A list of dictionaries representing mutation operations.
    """
    deletions = []
    app_id = clean_app_id(entity.key().app())
    namespace = entity.key().name_space()
    prefix = dbconstants.KEY_DELIMITER.join([app_id, namespace])
  
    asc_rows = get_index_kv_from_tuple([(prefix, entity)])
    for entry in asc_rows:
      deletions.append({'table': dbconstants.ASC_PROPERTY_TABLE,
                        'key': entry[0],
                        'operation': TxnActions.DELETE})
  
    dsc_rows = get_index_kv_from_tuple(
      [(prefix, entity)], reverse=True)
    for entry in dsc_rows:
      deletions.append({'table': dbconstants.DSC_PROPERTY_TABLE,
                        'key': entry[0],
                        'operation': TxnActions.DELETE})
  
    for key in get_composite_indexes_rows([entity], composite_indices):
      deletions.append({'table': dbconstants.COMPOSITE_TABLE,
                        'key': key,
                        'operation': TxnActions.DELETE})
  
    entity_key = get_entity_key(prefix, entity.key().path())
    deletions.append({'table': dbconstants.APP_ENTITY_TABLE,
                      'key': entity_key,
                      'operation': TxnActions.DELETE})
  
    kind_key = get_kind_key(prefix, entity.key().path())
    deletions.append({'table': dbconstants.APP_KIND_TABLE,
                      'key': kind_key,
                      'operation': TxnActions.DELETE})
  
    return deletions
  
  
  def index_deletions(old_entity, new_entity, composite_indices=()):
    """ Get a list of index deletions needed for updating an entity. For changing
    an existing entity, this involves examining the property list of both
    entities to see which index entries need to be removed.
  
    Args:
      old_entity: An entity object.
      new_entity: An entity object.
      composite_indices: A list or tuple of composite indices.
    Returns:
      A list of dictionaries representing mutation operations.
    """
    deletions = []
    app_id = clean_app_id(old_entity.key().app())
    namespace = old_entity.key().name_space()
    kind = get_entity_kind(old_entity.key())
    entity_key = str(encode_index_pb(old_entity.key().path()))
  
    new_props = {}
    for prop in new_entity.property_list():
      if prop.name() not in new_props:
        new_props[prop.name()] = []
      new_props[prop.name()].append(prop)
  
    changed_props = {}
    for prop in old_entity.property_list():
      if prop.name() in new_props and prop in new_props[prop.name()]:
        continue
  
      if prop.name() not in changed_props:
        changed_props[prop.name()] = []
      changed_props[prop.name()].append(prop)
  
      value = str(encode_index_pb(prop.value()))
  
      key = dbconstants.KEY_DELIMITER.join(
        [app_id, namespace, kind, prop.name(), value, entity_key])
      deletions.append({'table': dbconstants.ASC_PROPERTY_TABLE,
                        'key': key,
                        'operation': TxnActions.DELETE})
  
      reverse_key = dbconstants.KEY_DELIMITER.join(
        [app_id, namespace, kind, prop.name(),
         helper_functions.reverse_lex(value), entity_key])
      deletions.append({'table': dbconstants.DSC_PROPERTY_TABLE,
                        'key': reverse_key,
                        'operation': TxnActions.DELETE})
  
    changed_prop_names = set(changed_props.keys())
    for index in composite_indices:
      if index.definition().entity_type() != kind:
        continue
  
      index_props = set(prop.name() for prop
                        in index.definition().property_list())
      if index_props.isdisjoint(changed_prop_names):
        continue
  
      old_entries = set(get_composite_index_keys(index, old_entity))
      new_entries = set(get_composite_index_keys(index, new_entity))
      for entry in (old_entries - new_entries):
        deletions.append({'table': dbconstants.COMPOSITE_TABLE,
                          'key': entry,
                          'operation': TxnActions.DELETE})
  
    return deletions
  
  
  def mutations_for_entity(entity, txn, current_value=None,
                           composite_indices=()):
    """ Get a list of mutations needed across tables for an entity change.
  
    Args:
      entity: An entity object.
      txn: A transaction ID handler.
      current_value: The entity object currently stored.
      composite_indices: A list of composite indices for the entity kind.
    Returns:
      A list of dictionaries representing mutations.
    """
    mutations = []
    if current_value is not None:
      mutations.extend(
        index_deletions(current_value, entity, composite_indices))
  
    app_id = clean_app_id(entity.key().app())
    namespace = entity.key().name_space()
    encoded_path = str(encode_index_pb(entity.key().path()))
    prefix = dbconstants.KEY_DELIMITER.join([app_id, namespace])
    entity_key = dbconstants.KEY_DELIMITER.join([prefix, encoded_path])
    entity_value = {dbconstants.APP_ENTITY_SCHEMA[0]: entity.Encode(),
                    dbconstants.APP_ENTITY_SCHEMA[1]: str(txn)}
    mutations.append({'table': dbconstants.APP_ENTITY_TABLE,
                      'key': entity_key,
                      'operation': TxnActions.PUT,
                      'values': entity_value})
  
    reference_value = {'reference': entity_key}
  
    kind_key = get_kind_key(prefix, entity.key().path())
    mutations.append({'table': dbconstants.APP_KIND_TABLE,
                      'key': kind_key,
                      'operation': TxnActions.PUT,
                      'values': reference_value})
  
    asc_rows = get_index_kv_from_tuple([(prefix, entity)])
    for entry in asc_rows:
      mutations.append({'table': dbconstants.ASC_PROPERTY_TABLE,
                        'key': entry[0],
                        'operation': TxnActions.PUT,
                        'values': reference_value})
  
    dsc_rows = get_index_kv_from_tuple([(prefix, entity)], reverse=True)
    for entry in dsc_rows:
      mutations.append({'table': dbconstants.DSC_PROPERTY_TABLE,
                        'key': entry[0],
                        'operation': TxnActions.PUT,
                        'values': reference_value})
  
    for key in get_composite_indexes_rows([entity], composite_indices):
      mutations.append({'table': dbconstants.COMPOSITE_TABLE,
                        'key': key,
                        'operation': TxnActions.PUT,
                        'values': reference_value})
  
    return mutations
  
  
  class FailedBatch(Exception):
    pass
  
  
  class IdempotentRetryPolicy(RetryPolicy):
    """ A policy used for retrying idempotent statements. """
    def on_read_timeout(self, query, consistency, required_responses,
                        received_responses, data_retrieved, retry_num):
      """ This is called when a ReadTimeout occurs.
  
      Args:
        query: A statement that timed out.
        consistency: The consistency level of the statement.
        required_responses: The number of responses required.
        received_responses: The number of responses received.
        data_retrieved: Indicates whether any responses contained data.
        retry_num: The number of times the statement has been tried.
      """
      if retry_num >= 5:
        return self.RETHROW, None
      else:
        return self.RETRY, consistency
  
    def on_write_timeout(self, query, consistency, write_type,
                         required_responses, received_responses, retry_num):
      """ This is called when a WriteTimeout occurs.
  
      Args:
        query: A statement that timed out.
        consistency: The consistency level of the statement.
        required_responses: The number of responses required.
        received_responses: The number of responses received.
        data_retrieved: Indicates whether any responses contained data.
        retry_num: The number of times the statement has been tried.
        """
      if retry_num >= 5:
        return self.RETHROW, None
      else:
        return self.RETRY, consistency
  
  
  class ThriftColumn(object):
    """ Columns created by default with thrift interface. """
    KEY = 'key'
    COLUMN_NAME = 'column1'
    VALUE = 'value'
  
  
  class DatastoreProxy(AppDBInterface):
    """ 
      Cassandra implementation of the AppDBInterface
    """
    def __init__(self, log_level=logging.INFO):
      """
      Constructor.
      """
      class_name = self.__class__.__name__
      self.logger = logging.getLogger(class_name)
      self.logger.setLevel(log_level)
      self.logger.info('Starting {}'.format(class_name))
  
      self.hosts = appscale_info.get_db_ips()
      self.retry_policy = IdempotentRetryPolicy()
      self.no_retries = FallthroughRetryPolicy()
  
      remaining_retries = INITIAL_CONNECT_RETRIES
      while True:
        try:
          self.cluster = Cluster(self.hosts,
                                 default_retry_policy=self.retry_policy)
          self.session = self.cluster.connect(KEYSPACE)
          break
        except cassandra.cluster.NoHostAvailable as connection_error:
          remaining_retries -= 1
          if remaining_
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/cassandra_env/cassandra_interface.py
files: """
   Datastore Constants
  """
  import cassandra.cluster
  
  SECRET_LOCATION = "/etc/appscale/secret.key"
  
  ERROR_DEFAULT = "DB_ERROR:"
  NONEXISTANT_TRANSACTION = "0"
  
  # The datastore's default HTTP port.
  DEFAULT_PORT = 4080
  
  # The datastore's default HTTPS port.
  DEFAULT_SSL_PORT = 8443
  
  # The lowest character to separate different fields in a row key.
  KEY_DELIMITER = '\x00'
  
  # The character used to separate kinds in an ancestry.
  KIND_SEPARATOR = '\x01'
  
  # HTTP code to indicate that the request is invalid.
  HTTP_BAD_REQUEST = 400
  
  # The length of an ID string. A constant length allows lexicographic ordering.
  ID_KEY_LENGTH = 10
  
  # The character between the kind and the ID/name of an entity.
  ID_SEPARATOR = ":"
  
  # The maximum number of composite indexes an application can have.
  MAX_NUMBER_OF_COMPOSITE_INDEXES = 1000
  
  # A string used to create end keys when doing range queries.
  TERMINATING_STRING = chr(255) * 500
  
  # Tombstone value for soft deletes.
  TOMBSTONE = "APPSCALE_SOFT_DELETE"
  
  TRANSIENT_CASSANDRA_ERRORS = (
    cassandra.Unavailable, cassandra.Timeout, cassandra.CoordinationFailure,
    cassandra.OperationTimedOut, cassandra.cluster.NoHostAvailable)
  
  # The database backends supported by the AppScale datastore.
  VALID_DATASTORES = ['cassandra']
  
  # Table names
  USERS_TABLE = "USERS__"
  APPS_TABLE = "APPS__"
  JOURNAL_TABLE = "JOURNAL__"
  
  ASC_PROPERTY_TABLE = "ASC_PROPERTY__"
  DSC_PROPERTY_TABLE = "DSC_PROPERTY__"
  COMPOSITE_TABLE = "COMPOSITE_INDEXES__"
  APP_ID_TABLE = "APP_IDS__"
  APP_ENTITY_TABLE = "ENTITIES__"
  APP_KIND_TABLE = "KINDS__"
  METADATA_TABLE = "METADATA__"
  DATASTORE_METADATA_TABLE = "DATASTORE_METADATA__"
  TRANSACTIONS_TABLE = 'TRANSACTIONS__'
  SCHEMA_TABLE = '__key__'
  
  INITIAL_TABLES = [ASC_PROPERTY_TABLE,
                    DSC_PROPERTY_TABLE,
                    APP_ID_TABLE,
                    APP_ENTITY_TABLE,
                    APP_KIND_TABLE,
                    COMPOSITE_TABLE,
                    METADATA_TABLE,
                    USERS_TABLE,
                    APPS_TABLE,
                    SCHEMA_TABLE,
                    DATASTORE_METADATA_TABLE,
                    TRANSACTIONS_TABLE]
  
  ###########################################
  # DB schemas for version 1 of the datastore
  ###########################################
  JOURNAL_SCHEMA = [
    "Encoded_Entity"]
  
  ENTITY_TABLE_SCHEMA = [
    "Encoded_Entity",
    "Txn_Num"]
  
  ###########################################
  # DB schema for version 2 of the datastore
  ###########################################
  
  # The schema of the table which holds the encoded entities
  APP_ENTITY_SCHEMA = [
    "entity",
    "txnID"]
  
  # Index tables store references are to entity table
  PROPERTY_SCHEMA = [
    "reference" ]
  APP_ID_SCHEMA = [
    "next_id" ]
  APP_KIND_SCHEMA = [
    "reference" ]
  COMPOSITE_SCHEMA = [
    "reference" ]
  METADATA_SCHEMA = [
    "data" ]
  
  USERS_SCHEMA = [
    "email",
    "pw",
    "date_creation", 
    "date_change",
    "date_last_login",
    "applications",
    "appdrop_rem_token",
    "appdrop_rem_token_exp",
    "visit_cnt",
    "cookie",
    "cookie_ip",
    "cookie_exp",
    "cksum",
    "enabled",
    "type",
    "is_cloud_admin",
    "capabilities" ]
  
  APPS_SCHEMA = [
    "name",
    "language",
    "version",
    "owner",
    "admins_list",
    "host",
    "port",
    "creation_date",
    "last_time_updated_date",
    "yaml_file",
    "cksum",
    "num_entries",
    "tar_ball",
    "enabled",
    "indexes" ]
  
  DATASTORE_METADATA_SCHEMA = [
    "version"]
  
  TRANSACTIONS_SCHEMA = [
    'operation',
    'operand',
    'exclude_indices'
  ]
  
  # All schema information for the keyspace is stored in the schema table.
  SCHEMA_TABLE_SCHEMA = ['schema']
  
  
  # Possible values in the 'action' column of the transaction table.
  class TxnActions(object):
    DELETE = '0'
    PUT = '1'
    GET = '2'
    ENQUEUE_TASK = '3'
  
  
  class Operations(object):
    """ Possible datastore operations on entities. """
    PUT = 'put'
    DELETE = 'delete'
  
  
  ###############################
  # Generic Datastore Exceptions
  ###############################
  class AppScaleDBError(Exception):
    """ Tossed for generic datastore errors
    """
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
  
    def __str__(self):
      return repr(self.value)
  
  class AppScaleDBConnectionError(Exception):
    """ Tossed when there is a bad connection
    """ 
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
    def __str__(self):
      return repr(self.value)
  
  class AppScaleMisconfiguredQuery(Exception):
    """ Tossed when a query is misconfigured
    """
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
    def __str__(self):
      return repr(self.value)
  
  class AppScaleBadArg(Exception):
    """ Bad Argument given for a function
    """
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
    def __str__(self):
      return repr(self.value)
  
  class ExcessiveTasks(Exception):
    """ Indicates that there are too many tasks for a transaction. """
    pass
  
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/dbconstants.py
files: """
   Datastore Constants
  """
  import cassandra.cluster
  
  SECRET_LOCATION = "/etc/appscale/secret.key"
  
  ERROR_DEFAULT = "DB_ERROR:"
  NONEXISTANT_TRANSACTION = "0"
  
  # The datastore's default HTTP port.
  DEFAULT_PORT = 4080
  
  # The datastore's default HTTPS port.
  DEFAULT_SSL_PORT = 8443
  
  # The lowest character to separate different fields in a row key.
  KEY_DELIMITER = '\x00'
  
  # The character used to separate kinds in an ancestry.
  KIND_SEPARATOR = '\x01'
  
  # HTTP code to indicate that the request is invalid.
  HTTP_BAD_REQUEST = 400
  
  # The length of an ID string. A constant length allows lexicographic ordering.
  ID_KEY_LENGTH = 10
  
  # The character between the kind and the ID/name of an entity.
  ID_SEPARATOR = ":"
  
  # The maximum number of composite indexes an application can have.
  MAX_NUMBER_OF_COMPOSITE_INDEXES = 1000
  
  # A string used to create end keys when doing range queries.
  TERMINATING_STRING = chr(255) * 500
  
  # Tombstone value for soft deletes.
  TOMBSTONE = "APPSCALE_SOFT_DELETE"
  
  TRANSIENT_CASSANDRA_ERRORS = (
    cassandra.Unavailable, cassandra.Timeout, cassandra.CoordinationFailure,
    cassandra.OperationTimedOut, cassandra.cluster.NoHostAvailable)
  
  # The database backends supported by the AppScale datastore.
  VALID_DATASTORES = ['cassandra']
  
  # Table names
  USERS_TABLE = "USERS__"
  APPS_TABLE = "APPS__"
  JOURNAL_TABLE = "JOURNAL__"
  
  ASC_PROPERTY_TABLE = "ASC_PROPERTY__"
  DSC_PROPERTY_TABLE = "DSC_PROPERTY__"
  COMPOSITE_TABLE = "COMPOSITE_INDEXES__"
  APP_ID_TABLE = "APP_IDS__"
  APP_ENTITY_TABLE = "ENTITIES__"
  APP_KIND_TABLE = "KINDS__"
  METADATA_TABLE = "METADATA__"
  DATASTORE_METADATA_TABLE = "DATASTORE_METADATA__"
  TRANSACTIONS_TABLE = 'TRANSACTIONS__'
  SCHEMA_TABLE = '__key__'
  
  INITIAL_TABLES = [ASC_PROPERTY_TABLE,
                    DSC_PROPERTY_TABLE,
                    APP_ID_TABLE,
                    APP_ENTITY_TABLE,
                    APP_KIND_TABLE,
                    COMPOSITE_TABLE,
                    METADATA_TABLE,
                    USERS_TABLE,
                    APPS_TABLE,
                    SCHEMA_TABLE,
                    DATASTORE_METADATA_TABLE,
                    TRANSACTIONS_TABLE]
  
  ###########################################
  # DB schemas for version 1 of the datastore
  ###########################################
  JOURNAL_SCHEMA = [
    "Encoded_Entity"]
  
  ENTITY_TABLE_SCHEMA = [
    "Encoded_Entity",
    "Txn_Num"]
  
  ###########################################
  # DB schema for version 2 of the datastore
  ###########################################
  
  # The schema of the table which holds the encoded entities
  APP_ENTITY_SCHEMA = [
    "entity",
    "txnID"]
  
  # Index tables store references are to entity table
  PROPERTY_SCHEMA = [
    "reference" ]
  APP_ID_SCHEMA = [
    "next_id" ]
  APP_KIND_SCHEMA = [
    "reference" ]
  COMPOSITE_SCHEMA = [
    "reference" ]
  METADATA_SCHEMA = [
    "data" ]
  
  USERS_SCHEMA = [
    "email",
    "pw",
    "date_creation", 
    "date_change",
    "date_last_login",
    "applications",
    "appdrop_rem_token",
    "appdrop_rem_token_exp",
    "visit_cnt",
    "cookie",
    "cookie_ip",
    "cookie_exp",
    "cksum",
    "enabled",
    "type",
    "is_cloud_admin",
    "capabilities" ]
  
  APPS_SCHEMA = [
    "name",
    "language",
    "version",
    "owner",
    "admins_list",
    "host",
    "port",
    "creation_date",
    "last_time_updated_date",
    "yaml_file",
    "cksum",
    "num_entries",
    "tar_ball",
    "enabled",
    "indexes" ]
  
  DATASTORE_METADATA_SCHEMA = [
    "version"]
  
  TRANSACTIONS_SCHEMA = [
    'operation',
    'operand',
    'exclude_indices'
  ]
  
  # All schema information for the keyspace is stored in the schema table.
  SCHEMA_TABLE_SCHEMA = ['schema']
  
  
  # Possible values in the 'action' column of the transaction table.
  class TxnActions(object):
    DELETE = '0'
    PUT = '1'
    GET = '2'
    ENQUEUE_TASK = '3'
  
  
  ###############################
  # Generic Datastore Exceptions
  ###############################
  class AppScaleDBError(Exception):
    """ Tossed for generic datastore errors
    """
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
  
    def __str__(self):
      return repr(self.value)
  
  class AppScaleDBConnectionError(Exception):
    """ Tossed when there is a bad connection
    """ 
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
    def __str__(self):
      return repr(self.value)
  
  class AppScaleMisconfiguredQuery(Exception):
    """ Tossed when a query is misconfigured
    """
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
    def __str__(self):
      return repr(self.value)
  
  class AppScaleBadArg(Exception):
    """ Bad Argument given for a function
    """
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
    def __str__(self):
      return repr(self.value)
  
  class ExcessiveTasks(Exception):
    """ Indicates that there are too many tasks for a transaction. """
    pass
  
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/dbconstants.py
files: #!/usr/bin/env python
  # Programmer: Navraj Chohan <nlake44@gmail.com>
  
  import sys
  import unittest
  
  from appscale.datastore import dbconstants
  from appscale.datastore import utils
  from appscale.datastore.datastore_distributed import DatastoreDistributed
  from appscale.datastore.dbconstants import APP_ENTITY_SCHEMA
  from appscale.datastore.dbconstants import ID_KEY_LENGTH
  from appscale.datastore.dbconstants import JOURNAL_SCHEMA
  from appscale.datastore.dbconstants import TOMBSTONE
  from appscale.datastore.cassandra_env.cassandra_interface import DatastoreProxy
  from appscale.datastore.cassandra_env.cassandra_interface import\
    deletions_for_entity
  from appscale.datastore.cassandra_env.cassandra_interface import\
    index_deletions
  from appscale.datastore.cassandra_env.cassandra_interface import\
    mutations_for_entity
  from appscale.datastore.unpackaged import APPSCALE_LIB_DIR
  from appscale.datastore.unpackaged import APPSCALE_PYTHON_APPSERVER
  from appscale.datastore.utils import encode_index_pb
  from appscale.datastore.utils import get_entity_key
  from appscale.datastore.utils import get_entity_kind
  from appscale.datastore.utils import get_index_key_from_params
  from appscale.datastore.utils import get_index_kv_from_tuple
  from appscale.datastore.utils import get_kind_key
  from appscale.datastore.zkappscale.zktransaction import TX_TIMEOUT
  from appscale.datastore.zkappscale.zktransaction import ZKTransactionException
  from cassandra.cluster import Cluster
  from flexmock import flexmock
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import api_base_pb
  from google.appengine.datastore import entity_pb
  from google.appengine.datastore import datastore_pb
  from google.appengine.ext import db
  
  sys.path.append(APPSCALE_LIB_DIR)
  import appscale_info
  
  
  class Item(db.Model):
    name = db.StringProperty(required = True)
  
  
  class TestDatastoreServer(unittest.TestCase):
    """
    A set of test cases for the datastore server (datastore server v2)
    """
    BASIC_ENTITY = ['guestbook', 'Greeting', 'foo', 'content', 'hello world']
  
    def get_zookeeper(self):
      zookeeper = flexmock()
      zookeeper.should_receive("acquire_lock").and_return(True)
      zookeeper.should_receive("release_lock").and_return(True)
      zookeeper.should_receive("get_transaction_id").and_return(1)
      zookeeper.should_receive("increment_and_get_counter").and_return(0,1000)
      return zookeeper
  
    def test_get_entity_kind(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, None)
      item = Item(name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(get_entity_kind(key), "Item")
  
    def test_kind_key(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, None)
      item = Item(name="Dyan", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(get_kind_key("howdy", key.key().path()), "howdy\x00Item\x01Item:0000000000\x01")
  
      item1 = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item1)
      self.assertEquals(get_kind_key("howdy", key.key().path()), "howdy\x00Item\x01Item:Bob\x01")
     
      item2 = Item(key_name="Frank", name="Frank", _app="hello", parent = item1)
      key = db.model_to_protobuf(item2)
      self.assertEquals(get_kind_key("howdy", key.key().path()),
             "howdy\x00Item\x01Item:Bob\x01Item:Frank\x01")
  
    def test_get_entity_key(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, None)
      item = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(str(get_entity_key("howdy", key.key().path())), "howdy\x00Item:Bob\x01")
  
    def test_validate_key(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, None)
      item = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      dd.validate_key(key.key())
  
    def test_get_table_prefix(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      db_batch.should_receive("batch_put_entity").and_return(None)
      zookeeper = flexmock()
      zookeeper.should_receive("acquire_lock").and_return(True)
      zookeeper.should_receive("release_lock").and_return(True)
      dd = DatastoreDistributed(db_batch, zookeeper)
      item = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(dd.get_table_prefix(key), "hello\x00")
  
    def test_get_index_key_from_params(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, None)
      params = ['a','b','c','d','e']
      self.assertEquals(get_index_key_from_params(params),
                        "a\x00b\x00c\x00d\x00e")
  
    def test_get_index_kv_from_tuple(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, None)
      item1 = Item(key_name="Bob", name="Bob", _app="hello")
      item2 = Item(key_name="Sally", name="Sally", _app="hello")
      key1 = db.model_to_protobuf(item1)
      key2 = db.model_to_protobuf(item2)
      tuples_list = [("a\x00b",key1),("a\x00b",key2)]
      self.assertEquals(get_index_kv_from_tuple(
        tuples_list), (['a\x00b\x00Item\x00name\x00\x9aBob\x01\x01\x00Item:Bob\x01', 
        'a\x00b\x00Item:Bob\x01'], 
        ['a\x00b\x00Item\x00name\x00\x9aSally\x01\x01\x00Item:Sally\x01', 
        'a\x00b\x00Item:Sally\x01']))
  
    def test_get_composite_index_key(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, self.get_zookeeper())
      dd = flexmock(dd)
  
      composite_index = entity_pb.CompositeIndex()
      composite_index.set_id(123)
      composite_index.set_app_id("appid")
  
      definition = composite_index.mutable_definition()
      definition.set_entity_type("kind")
  
      prop1 = definition.add_property()
      prop1.set_name("prop1")
      prop1.set_direction(1) # ascending
      prop2 = definition.add_property()
      prop2.set_name("prop2")
      prop1.set_direction(1) # ascending
  
      ent = self.get_new_entity_proto("appid", "kind", "entity_name", "prop1", "value", ns="")
       
      self.assertEquals(dd.get_composite_index_key(composite_index, ent), 
        "appid\x00\x00123\x00\x9avalue\x01\x01\x00\x00kind:entity_name\x01")
  
    def test_get_indices(self):
      session = flexmock(default_consistency_level=None)
      cluster = flexmock(connect=lambda keyspace: session)
      flexmock(appscale_info).should_receive('get_db_ips')
      flexmock(Cluster).new_instances(cluster)
      flexmock(DatastoreProxy).should_receive('range_query').and_return({})
      db_batch = DatastoreProxy()
  
      self.assertEquals(db_batch.get_indices("appid"), [])
  
    def test_delete_composite_index_metadata(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      db_batch.should_receive("batch_delete").and_return(None)
      dd = DatastoreDistributed(db_batch, self.get_zookeeper())
      dd = flexmock(dd)
      composite_index = entity_pb.CompositeIndex()
      composite_index.set_id(1)
      dd.delete_composite_index_metadata("appid", composite_index)
  
    def test_create_composite_index(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      db_batch.should_receive("batch_put_entity").and_return(None)
      dd = DatastoreDistributed(db_batch, self.get_zookeeper())
      dd = flexmock(dd)
      index = entity_pb.CompositeIndex()
      index.set_app_id("appid")
      index.set_state(2)
      definition = index.mutable_definition()
      definition.set_entity_type("kind")
      definition.set_ancestor(0)
      prop1 = definition.add_property()
      prop1.set_name("prop1")
      prop1.set_direction(1) # ascending
      prop2 = definition.add_property()
      prop2.set_name("prop2")
      prop1.set_direction(1) # ascending
  
      dd.create_composite_index("appid", index)
      assert index.id() > 0 
  
    def test_insert_composite_indexes(self):
      composite_index = entity_pb.CompositeIndex()
      composite_index.set_id(123)
      composite_index.set_app_id("appid")
  
      definition = composite_index.mutable_definition()
      definition.set_entity_type("kind")
  
      prop1 = definition.add_property()
      prop1.set_name("prop1")
      prop1.set_direction(1) # ascending
      prop2 = definition.add_property()
      prop2.set_name("prop2")
      prop1.set_direction(1) # ascending
  
      ent = self.get_new_entity_proto("appid", "kind", "entity_name", "prop1", "value", ns="")
  
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      db_batch.should_receive("batch_put_entity").and_return(None).once()
      dd = DatastoreDistributed(db_batch, self.get_zookeeper())
      dd.insert_composite_indexes([ent], [composite_index])
  
    def test_allocate_ids(self):
      PREFIX = "x"
      BATCH_SIZE = 1000
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, self.get_zookeeper())
      self.assertEquals(dd.allocate_ids(PREFIX, BATCH_SIZE), (1, 1000))
  
      dd = Dat
  
  [... the rest is omitted ...]
files: AppDB/test/unit/test_datastore_server.py
files: #!/usr/bin/env python
  # Programmer: Navraj Chohan <nlake44@gmail.com>
  
  import sys
  import unittest
  
  from appscale.datastore import dbconstants
  from appscale.datastore import utils
  from appscale.datastore.datastore_distributed import DatastoreDistributed
  from appscale.datastore.dbconstants import APP_ENTITY_SCHEMA
  from appscale.datastore.dbconstants import ID_KEY_LENGTH
  from appscale.datastore.dbconstants import JOURNAL_SCHEMA
  from appscale.datastore.dbconstants import TOMBSTONE
  from appscale.datastore.cassandra_env.cassandra_interface import DatastoreProxy
  from appscale.datastore.cassandra_env.cassandra_interface import\
    deletions_for_entity
  from appscale.datastore.cassandra_env.cassandra_interface import\
    index_deletions
  from appscale.datastore.cassandra_env.cassandra_interface import\
    mutations_for_entity
  from appscale.datastore.unpackaged import APPSCALE_LIB_DIR
  from appscale.datastore.unpackaged import APPSCALE_PYTHON_APPSERVER
  from appscale.datastore.utils import encode_index_pb
  from appscale.datastore.utils import get_entity_key
  from appscale.datastore.utils import get_entity_kind
  from appscale.datastore.utils import get_index_key_from_params
  from appscale.datastore.utils import get_index_kv_from_tuple
  from appscale.datastore.utils import get_kind_key
  from appscale.datastore.zkappscale.zktransaction import TX_TIMEOUT
  from appscale.datastore.zkappscale.zktransaction import ZKTransactionException
  from cassandra.cluster import Cluster
  from flexmock import flexmock
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import api_base_pb
  from google.appengine.datastore import entity_pb
  from google.appengine.datastore import datastore_pb
  from google.appengine.ext import db
  
  sys.path.append(APPSCALE_LIB_DIR)
  import appscale_info
  
  
  class Item(db.Model):
    name = db.StringProperty(required = True)
  
  
  class TestDatastoreServer(unittest.TestCase):
    """
    A set of test cases for the datastore server (datastore server v2)
    """
    BASIC_ENTITY = ['guestbook', 'Greeting', 'foo', 'content', 'hello world']
  
    def get_zookeeper(self):
      zookeeper = flexmock()
      zookeeper.should_receive("acquire_lock").and_return(True)
      zookeeper.should_receive("release_lock").and_return(True)
      zookeeper.should_receive("get_transaction_id").and_return(1)
      zookeeper.should_receive("increment_and_get_counter").and_return(0,1000)
      return zookeeper
  
    def test_get_entity_kind(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, None)
      item = Item(name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(get_entity_kind(key), "Item")
  
    def test_kind_key(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, None)
      item = Item(name="Dyan", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(get_kind_key("howdy", key.key().path()), "howdy\x00Item\x01Item:0000000000\x01")
  
      item1 = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item1)
      self.assertEquals(get_kind_key("howdy", key.key().path()), "howdy\x00Item\x01Item:Bob\x01")
     
      item2 = Item(key_name="Frank", name="Frank", _app="hello", parent = item1)
      key = db.model_to_protobuf(item2)
      self.assertEquals(get_kind_key("howdy", key.key().path()),
             "howdy\x00Item\x01Item:Bob\x01Item:Frank\x01")
  
    def test_get_entity_key(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, None)
      item = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(str(get_entity_key("howdy", key.key().path())), "howdy\x00Item:Bob\x01")
  
    def test_validate_key(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, None)
      item = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      dd.validate_key(key.key())
  
    def test_get_table_prefix(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      db_batch.should_receive("batch_put_entity").and_return(None)
      zookeeper = flexmock()
      zookeeper.should_receive("acquire_lock").and_return(True)
      zookeeper.should_receive("release_lock").and_return(True)
      dd = DatastoreDistributed(db_batch, zookeeper)
      item = Item(key_name="Bob", name="Bob", _app="hello")
      key = db.model_to_protobuf(item)
      self.assertEquals(dd.get_table_prefix(key), "hello\x00")
  
    def test_get_index_key_from_params(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, None)
      params = ['a','b','c','d','e']
      self.assertEquals(get_index_key_from_params(params),
                        "a\x00b\x00c\x00d\x00e")
  
    def test_get_index_kv_from_tuple(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, None)
      item1 = Item(key_name="Bob", name="Bob", _app="hello")
      item2 = Item(key_name="Sally", name="Sally", _app="hello")
      key1 = db.model_to_protobuf(item1)
      key2 = db.model_to_protobuf(item2)
      tuples_list = [("a\x00b",key1),("a\x00b",key2)]
      self.assertEquals(get_index_kv_from_tuple(
        tuples_list), (['a\x00b\x00Item\x00name\x00\x9aBob\x01\x01\x00Item:Bob\x01', 
        'a\x00b\x00Item:Bob\x01'], 
        ['a\x00b\x00Item\x00name\x00\x9aSally\x01\x01\x00Item:Sally\x01', 
        'a\x00b\x00Item:Sally\x01']))
  
    def test_get_composite_index_key(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, self.get_zookeeper())
      dd = flexmock(dd)
  
      composite_index = entity_pb.CompositeIndex()
      composite_index.set_id(123)
      composite_index.set_app_id("appid")
  
      definition = composite_index.mutable_definition()
      definition.set_entity_type("kind")
  
      prop1 = definition.add_property()
      prop1.set_name("prop1")
      prop1.set_direction(1) # ascending
      prop2 = definition.add_property()
      prop2.set_name("prop2")
      prop1.set_direction(1) # ascending
  
      ent = self.get_new_entity_proto("appid", "kind", "entity_name", "prop1", "value", ns="")
       
      self.assertEquals(dd.get_composite_index_key(composite_index, ent), 
        "appid\x00\x00123\x00\x9avalue\x01\x01\x00\x00kind:entity_name\x01")
  
    def test_get_indices(self):
      session = flexmock(default_consistency_level=None)
      cluster = flexmock(connect=lambda keyspace: session)
      flexmock(appscale_info).should_receive('get_db_ips')
      flexmock(Cluster).new_instances(cluster)
      flexmock(DatastoreProxy).should_receive('range_query').and_return({})
      db_batch = DatastoreProxy()
  
      self.assertEquals(db_batch.get_indices("appid"), [])
  
    def test_delete_composite_index_metadata(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      db_batch.should_receive("batch_delete").and_return(None)
      dd = DatastoreDistributed(db_batch, self.get_zookeeper())
      dd = flexmock(dd)
      composite_index = entity_pb.CompositeIndex()
      composite_index.set_id(1)
      dd.delete_composite_index_metadata("appid", composite_index)
  
    def test_create_composite_index(self):
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      db_batch.should_receive("batch_put_entity").and_return(None)
      dd = DatastoreDistributed(db_batch, self.get_zookeeper())
      dd = flexmock(dd)
      index = entity_pb.CompositeIndex()
      index.set_app_id("appid")
      index.set_state(2)
      definition = index.mutable_definition()
      definition.set_entity_type("kind")
      definition.set_ancestor(0)
      prop1 = definition.add_property()
      prop1.set_name("prop1")
      prop1.set_direction(1) # ascending
      prop2 = definition.add_property()
      prop2.set_name("prop2")
      prop1.set_direction(1) # ascending
  
      dd.create_composite_index("appid", index)
      assert index.id() > 0 
  
    def test_insert_composite_indexes(self):
      composite_index = entity_pb.CompositeIndex()
      composite_index.set_id(123)
      composite_index.set_app_id("appid")
  
      definition = composite_index.mutable_definition()
      definition.set_entity_type("kind")
  
      prop1 = definition.add_property()
      prop1.set_name("prop1")
      prop1.set_direction(1) # ascending
      prop2 = definition.add_property()
      prop2.set_name("prop2")
      prop1.set_direction(1) # ascending
  
      ent = self.get_new_entity_proto("appid", "kind", "entity_name", "prop1", "value", ns="")
  
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      db_batch.should_receive("batch_put_entity").and_return(None).once()
      dd = DatastoreDistributed(db_batch, self.get_zookeeper())
      dd.insert_composite_indexes([ent], [composite_index])
  
    def test_allocate_ids(self):
      PREFIX = "x"
      BATCH_SIZE = 1000
      db_batch = flexmock()
      db_batch.should_receive('valid_data_version').and_return(True)
      dd = DatastoreDistributed(db_batch, self.get_zookeeper())
      self.assertEquals(dd.allocate_ids(PREFIX, BATCH_SIZE), (1, 1000))
  
      dd = Dat
  
  [... the rest is omitted ...]
files: AppDB/test/unit/test_datastore_server.py