• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
17from __future__ import absolute_import
18
19import re
20import uuid
21
22import boto
23
24import gslib.tests.testcase as testcase
25from gslib.tests.util import ObjectToURI as suri
26from gslib.tests.util import unittest
27
28
29def _LoadNotificationUrl():
30  return boto.config.get_value('GSUtil', 'test_notification_url')
31
32NOTIFICATION_URL = _LoadNotificationUrl()
33
34
35class 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)
77