tools/telemetry/third_party/gsutilz/gslib/tests/test_notification.py

   1 # -*- coding: utf-8 -*-
   2 # Copyright 2013 Google Inc. All Rights Reserved.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 """Integration tests for notification command."""
  16
  17 from __future__ import absolute_import
  18
  19 import re
  20 import uuid
  21
  22 import boto
  23
  24 import gslib.tests.testcase as testcase
  25 from gslib.tests.util import ObjectToURI as suri
  26 from gslib.tests.util import unittest
  27
  28
  29 def _LoadNotificationUrl():
  30   return boto.config.get_value('GSUtil', 'test_notification_url')
  31
  32 NOTIFICATION_URL = _LoadNotificationUrl()
  33
  34
  35 class TestNotification(testcase.GsUtilIntegrationTestCase):
  36   """Integration tests for notification command."""
  37
  38   @unittest.skipUnless(NOTIFICATION_URL,
  39                        'Test requires notification URL configuration.')
  40   def test_watch_bucket(self):
  41     """Tests creating a notification channel on a bucket."""
  42     bucket_uri = self.CreateBucket()
  43     self.RunGsUtil([
  44         'notification', 'watchbucket', NOTIFICATION_URL, suri(bucket_uri)])
  45
  46     identifier = str(uuid.uuid4())
  47     token = str(uuid.uuid4())
  48     stderr = self.RunGsUtil([
  49         'notification', 'watchbucket', '-i', identifier, '-t', token,
  50         NOTIFICATION_URL, suri(bucket_uri)], return_stderr=True)
  51     self.assertIn('token: %s' % token, stderr)
  52     self.assertIn('identifier: %s' % identifier, stderr)
  53
  54   @unittest.skipUnless(NOTIFICATION_URL,
  55                        'Test requires notification URL configuration.')
  56   def test_stop_channel(self):
  57     """Tests stopping a notification channel on a bucket."""
  58     bucket_uri = self.CreateBucket()
  59     stderr = self.RunGsUtil(
  60         ['notification', 'watchbucket', NOTIFICATION_URL, suri(bucket_uri)],
  61         return_stderr=True)
  62
  63     channel_id = re.findall(r'channel identifier: (?P<id>.*)', stderr)
  64     self.assertEqual(len(channel_id), 1)
  65     resource_id = re.findall(r'resource identifier: (?P<id>.*)', stderr)
  66     self.assertEqual(len(resource_id), 1)
  67
  68     channel_id = channel_id[0]
  69     resource_id = resource_id[0]
  70
  71     self.RunGsUtil(['notification', 'stopchannel', channel_id, resource_id])
  72
  73   def test_invalid_subcommand(self):
  74     stderr = self.RunGsUtil(['notification', 'foo', 'bar', 'baz'],
  75                             return_stderr=True, expected_status=1)
  76     self.assertIn('Invalid subcommand', stderr)