text: # coding: utf-8
  # -------------------------------------------------------------------------
  # Copyright (c) Microsoft Corporation. All rights reserved.
  # Licensed under the MIT License. See License.txt in the project root for
  # license information.
  # --------------------------------------------------------------------------
  from __future__ import division
  
  import functools
  import logging
  import os
  import os.path
  import random
  import re
  import sys
  import time
  import zlib
  
  from azure_devtools.scenario_tests import RecordingProcessor
  from devtools_testutils import AzureTestCase, PowerShellPreparer
  from fake_credentials import STORAGE_DATA_LAKE_ACCOUNT_FAKE_KEY
  
  try:
      from cStringIO import StringIO      # Python 2
  except ImportError:
      from io import StringIO
  
  from azure.core.credentials import AccessToken
  
  try:
      import settings_real as settings
  except ImportError:
      settings = None
  
  
  LOGGING_FORMAT = '%(asctime)s %(name)-20s %(levelname)-5s %(message)s'
  
  
  DataLakePreparer = functools.partial(
      PowerShellPreparer, "storage",
      datalake_storage_account_name="storagename",
      datalake_storage_account_key=STORAGE_DATA_LAKE_ACCOUNT_FAKE_KEY
  )
  
  
  class FakeTokenCredential(object):
      """Protocol for classes able to provide OAuth tokens.
      :param str scopes: Lets you specify the type of access needed.
      """
      def __init__(self):
          self.token = AccessToken("YOU SHALL NOT PASS", 0)
  
      def get_token(self, *args):
          return self.token
  
  
  class XMSRequestIDBody(RecordingProcessor):
      """This process is used for Storage batch call only, to avoid the echo policy.
      """
      def process_response(self, response):
          content_type = None
          for key, value in response.get('headers', {}).items():
              if key.lower() == 'content-type':
                  content_type = (value[0] if isinstance(value, list) else value).lower()
                  break
  
          if content_type and 'multipart/mixed' in content_type:
              response['body']['string'] = re.sub(b"x-ms-client-request-id: [a-f0-9-]+\r\n", b"", response['body']['string'])
  
          return response
  
  
  class StorageTestCase(AzureTestCase):
      def __init__(self, *args, **kwargs):
          super(StorageTestCase, self).__init__(*args, **kwargs)
          self.replay_processors.append(XMSRequestIDBody())
          self.logger = logging.getLogger('azure.storage')
          self.configure_logging()
  
      def configure_logging(self):
          self.enable_logging() if os.environ.get('ENABLE_LOGGING', False) else self.disable_logging()
  
      def enable_logging(self):
          handler = logging.StreamHandler()
          handler.setFormatter(logging.Formatter(LOGGING_FORMAT))
          self.logger.handlers = [handler]
          self.logger.setLevel(logging.DEBUG)
          self.logger.propagate = True
          self.logger.disabled = False
  
      def disable_logging(self):
          self.logger.propagate = False
          self.logger.disabled = True
          self.logger.handlers = []
  
      def sleep(self, seconds):
          if not self.is_playback():
              time.sleep(seconds)
  
      def get_resource_name(self, prefix=''):
          # Append a suffix to the name, based on the fully qualified test name
          # We use a checksum of the test name so that each test gets different
          # resource names, but each test will get the same name on repeat runs,
          # which is needed for playback.
          # Most resource names have a length limit, so we use a crc32
  
          checksum = zlib.adler32(self.qualified_test_name.encode()) & 0xffffffff
          name = '{}{}'.format(prefix, hex(checksum)[2:])
          if name.endswith('L'):
              name = name[:-1]
          return name
  
      def get_random_bytes(self, size):
          return b'a'*size
  
      def get_random_text_data(self, size):
          '''Returns random unicode text data exceeding the size threshold for
          chunking blob upload.'''
          checksum = zlib.adler32(self.qualified_test_name.encode()) & 0xffffffff
          rand = random.Random(checksum)
          text = u''
          words = [u'hello', u'world', u'python', u'啊齄丂狛狜']
          while (len(text) < size):
              index = int(rand.random()*(len(words) - 1))
              text = text + u' ' + words[index]
  
          return text
  
      @staticmethod
      def _set_test_proxy(service, settings):
          if settings.USE_PROXY:
              service.set_proxy(
                  settings.PROXY_HOST,
                  settings.PROXY_PORT,
                  settings.PROXY_USER,
                  settings.PROXY_PASSWORD,
              )
  
      def _get_shared_key_credential(self, account_name, account_key):
          return {
              "account_name": account_name,
              "account_key": account_key
          }
  
      def _get_account_url(self, account_name):
          return "https://{}.dfs.core.windows.net".format(
              account_name
          )
  
      def _get_oauth_account_url(self, account_name):
          return "https://{}.blob.core.windows.net".format(
              account_name
          )
  
      def assertNamedItemInContainer(self, container, item_name, msg=None):
          def _is_string(obj):
              if sys.version_info >= (3,):
                  return isinstance(obj, str)
              else:
                  return isinstance(obj, basestring)
          for item in container:
              if _is_string(item):
                  if item == item_name:
                      return
              elif item.name == item_name:
                  return
              elif hasattr(item, 'snapshot') and item.snapshot == item_name:
                  return
  
  
          standardMsg = '{0} not found in {1}'.format(
              repr(item_name), [str(c) for c in container])
          self.fail(self._formatMessage(msg, standardMsg))
  
      def assertNamedItemNotInContainer(self, container, item_name, msg=None):
          for item in container:
              if item.name == item_name:
                  standardMsg = '{0} unexpectedly found in {1}'.format(
                      repr(item_name), repr(container))
                  self.fail(self._formatMessage(msg, standardMsg))
  
      def generate_oauth_token(self):
          if self.is_live:
              from azure.identity import ClientSecretCredential
              return ClientSecretCredential(
                  self.get_settings_value("TENANT_ID"),
                  self.get_settings_value("CLIENT_ID"),
                  self.get_settings_value("CLIENT_SECRET"),
              )
          return self.generate_fake_token()
  
      def generate_fake_token(self):
          return FakeTokenCredential()
  