# Path: google/datalab/contrib/pipeline/composer/_composer.py
# class Composer(object):
#   """ Represents a Composer object that encapsulates a set of functionality relating to the
#   Cloud Composer service.
# 
#   This object can be used to generate the python airflow spec.
#   """
# 
#   gcs_file_regexp = re.compile('gs://.*')
# 
#   def __init__(self, zone, environment):
#     """ Initializes an instance of a Composer object.
# 
#     Args:
#       zone: Zone in which Composer environment has been created.
#       environment: Name of the Composer environment.
#     """
#     self._zone = zone
#     self._environment = environment
#     self._gcs_dag_location = None
# 
#   def deploy(self, name, dag_string):
#     bucket_name, file_path = self.gcs_dag_location.split('/', 3)[2:]  # setting maxsplit to 3
#     file_name = '{0}{1}.py'.format(file_path, name)
# 
#     bucket = storage.Bucket(bucket_name)
#     file_object = bucket.object(file_name)
#     file_object.write_stream(dag_string, 'text/plain')
# 
#   @property
#   def gcs_dag_location(self):
#     if not self._gcs_dag_location:
#       environment_details = Api.get_environment_details(self._zone, self._environment)
# 
#       if ('config' not in environment_details or
#               'gcsDagLocation' not in environment_details.get('config')):
#         raise ValueError('Dag location unavailable from Composer environment {0}'.format(
#           self._environment))
#       gcs_dag_location = environment_details['config']['gcsDagLocation']
# 
#       if gcs_dag_location is None or not self.gcs_file_regexp.match(gcs_dag_location):
#         raise ValueError(
#           'Dag location {0} from Composer environment {1} is in incorrect format'.format(
#             gcs_dag_location, self._environment))
# 
#       self._gcs_dag_location = gcs_dag_location
#       if gcs_dag_location.endswith('/') is False:
#         self._gcs_dag_location = self._gcs_dag_location + '/'
# 
#     return self._gcs_dag_location


import unittest
import mock
from google.datalab.contrib.pipeline.composer._composer import Composer
      # Composer returns None result
      mock_environment_details.return_value = {
        'config': {
          'gcsDagLocation': None
        }
      }
      test_composer = Composer('foo_zone', 'foo_environment')
      with self.assertRaisesRegexp(
              ValueError,
              'Dag location None from Composer environment foo_environment is in incorrect format'):
        test_composer.gcs_dag_location

      # Composer returns incorrect formats
      mock_environment_details.return_value = {
        'config': {
          'gcsDagLocation': 'gs:/foo_bucket'
        }
      }
      test_composer = Composer('foo_zone', 'foo_environment')
      with self.assertRaisesRegexp(
              ValueError,
              ('Dag location gs:/foo_bucket from Composer environment foo_environment is in'
               ' incorrect format')):
        test_composer.gcs_dag_location

      mock_environment_details.return_value = {
        'config': {
          'gcsDagLocation': 'as://foo_bucket'
        }
      }

# Path: google/datalab/contrib/pipeline/composer/_composer.py
# class Composer(object):
#   """ Represents a Composer object that encapsulates a set of functionality relating to the
#   Cloud Composer service.
# 
#   This object can be used to generate the python airflow spec.
#   """
# 
#   gcs_file_regexp = re.compile('gs://.*')
# 
#   def __init__(self, zone, environment):
#     """ Initializes an instance of a Composer object.
# 
#     Args:
#       zone: Zone in which Composer environment has been created.
#       environment: Name of the Composer environment.
#     """
#     self._zone = zone
#     self._environment = environment
#     self._gcs_dag_location = None
# 
#   def deploy(self, name, dag_string):
#     bucket_name, file_path = self.gcs_dag_location.split('/', 3)[2:]  # setting maxsplit to 3
#     file_name = '{0}{1}.py'.format(file_path, name)
# 
#     bucket = storage.Bucket(bucket_name)
#     file_object = bucket.object(file_name)
#     file_object.write_stream(dag_string, 'text/plain')
# 
#   @property
#   def gcs_dag_location(self):
#     if not self._gcs_dag_location:
#       environment_details = Api.get_environment_details(self._zone, self._environment)
# 
#       if ('config' not in environment_details or
#               'gcsDagLocation' not in environment_details.get('config')):
#         raise ValueError('Dag location unavailable from Composer environment {0}'.format(
#           self._environment))
#       gcs_dag_location = environment_details['config']['gcsDagLocation']
# 
#       if gcs_dag_location is None or not self.gcs_file_regexp.match(gcs_dag_location):
#         raise ValueError(
#           'Dag location {0} from Composer environment {1} is in incorrect format'.format(
#             gcs_dag_location, self._environment))
# 
#       self._gcs_dag_location = gcs_dag_location
#       if gcs_dag_location.endswith('/') is False:
#         self._gcs_dag_location = self._gcs_dag_location + '/'
# 
#     return self._gcs_dag_location


# Path: tests/pipeline/composer_tests.py
import unittest
import mock
from google.datalab.contrib.pipeline.composer._composer import Composer

      # Composer returns None result
      mock_environment_details.return_value = {
        'config': {
          'gcsDagLocation': None
        }
      }
      test_composer = Composer('foo_zone', 'foo_environment')
      with self.assertRaisesRegexp(
              ValueError,
              'Dag location None from Composer environment foo_environment is in incorrect format'):
        test_composer.gcs_dag_location

      # Composer returns incorrect formats
      mock_environment_details.return_value = {
        'config': {
          'gcsDagLocation': 'gs:/foo_bucket'
        }
      }
      test_composer = Composer('foo_zone', 'foo_environment')
      with self.assertRaisesRegexp(
              ValueError,
              ('Dag location gs:/foo_bucket from Composer environment foo_environment is in'
               ' incorrect format')):
        test_composer.gcs_dag_location

      mock_environment_details.return_value = {
        'config': {
          'gcsDagLocation': 'as://foo_bucket'
        }
      }

      test_composer = Composer('foo_zone', 'foo_environment')