| Server IP : 88.222.222.117 / Your IP : 216.73.217.106 Web Server : LiteSpeed System : Linux id-dci-web1986.main-hosting.eu 5.14.0-611.26.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jan 29 05:24:47 EST 2026 x86_64 User : u686484674 ( 686484674) PHP Version : 8.0.30 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /proc/self/./root/proc/self/root/opt/gsutil/gslib/tests/ |
Upload File : |
# -*- coding: utf-8 -*-
# Copyright 2017 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Integration tests for notification command."""
from __future__ import absolute_import
from __future__ import print_function
from __future__ import division
from __future__ import unicode_literals
import logging
import unittest
from gslib.cs_api_map import ApiSelector
from gslib.project_id import PopulateProjectId
from gslib.pubsub_api import PubsubApi
import gslib.tests.testcase as testcase
from gslib.tests.testcase.integration_testcase import SkipForS3
from gslib.tests.util import ObjectToURI as suri
@SkipForS3('gsutil doesn\'t support S3 notifications')
class TestNotificationPubSub(testcase.GsUtilIntegrationTestCase):
"""Integration tests for notification command (the Cloud Pub/Sub parts)."""
def setUp(self):
super(TestNotificationPubSub, self).setUp()
self.pubsub_api = PubsubApi(logger=logging.getLogger())
self.created_topic = None
def tearDown(self):
super(TestNotificationPubSub, self).tearDown()
# Cleanup any created topics.
if self.created_topic:
self.pubsub_api.DeleteTopic(self.created_topic)
self.created_topic = None
def _RegisterDefaultTopicCreation(self, bucket_name):
"""Records the name of a topic we expect to create, for cleanup."""
if self.test_api == ApiSelector.XML:
return unittest.skip('Notifications only work with the JSON API.')
expected_topic_name = 'projects/%s/topics/%s' % (PopulateProjectId(None),
bucket_name)
self.created_topic = expected_topic_name
return expected_topic_name
def test_list_new_bucket(self):
"""Tests listing notification configs on a new bucket."""
if self.test_api == ApiSelector.XML:
return unittest.skip('Notifications only work with the JSON API.')
bucket_uri = self.CreateBucket()
stdout = self.RunGsUtil(
['notification', 'list', suri(bucket_uri)], return_stdout=True)
self.assertFalse(stdout)
def test_delete_with_no_notifications(self):
"""Tests deleting all notification configs when there are none."""
if self.test_api == ApiSelector.XML:
return unittest.skip('Notifications only work with the JSON API.')
bucket_uri = self.CreateBucket()
stdout = self.RunGsUtil(
['notification', 'delete', suri(bucket_uri)], return_stdout=True)
self.assertFalse(stdout)
def test_create_basic(self):
"""Tests the create command succeeds in normal circumstances."""
if self.test_api == ApiSelector.XML:
return unittest.skip('Notifications only work with the JSON API.')
bucket_uri = self.CreateBucket()
topic_name = self._RegisterDefaultTopicCreation(bucket_uri.bucket_name)
stdout, stderr = self.RunGsUtil(
['notification', 'create', '-f', 'json',
suri(bucket_uri)],
return_stdout=True,
return_stderr=True)
if self._use_gcloud_storage:
self.assertIn('kind: storage#notification', stdout)
self.assertIn(topic_name, stdout)
else:
self.assertIn('Created notification', stderr)
self.assertIn(topic_name, stderr)
def test_list_one_entry(self):
"""Tests notification config listing with one entry."""
if self.test_api == ApiSelector.XML:
return unittest.skip('Notifications only work with the JSON API.')
bucket_uri = self.CreateBucket()
bucket_name = bucket_uri.bucket_name
topic_name = self._RegisterDefaultTopicCreation(bucket_uri.bucket_name)
self.RunGsUtil([
'notification', 'create', '-f', 'json', '-e', 'OBJECT_FINALIZE', '-e',
'OBJECT_DELETE', '-m', 'someKey:someValue', '-p', 'somePrefix',
suri(bucket_uri)
],
return_stderr=True)
stdout = self.RunGsUtil(
['notification', 'list', suri(bucket_uri)], return_stdout=True)
if self._use_gcloud_storage:
# gcloud and gsutil look the same in the terminal, but, for some reason,
# the shim test script detects an extra newline.
trailing_space = '\n'
else:
trailing_space = ''
self.assertEqual(
stdout,
('projects/_/buckets/{bucket_name}/notificationConfigs/1\n'
'\tCloud Pub/Sub topic: {topic_name}\n'
'\tCustom attributes:\n'
'\t\tsomeKey: someValue\n'
'\tFilters:\n'
'\t\tEvent Types: OBJECT_FINALIZE, OBJECT_DELETE\n'
'\t\tObject name prefix: \'somePrefix\'\n{trailing_space}'.format(
bucket_name=bucket_name,
topic_name=topic_name,
trailing_space=trailing_space)))
def test_delete(self):
"""Tests the create command succeeds in normal circumstances."""
if self.test_api == ApiSelector.XML:
return unittest.skip('Notifications only work with the JSON API.')
bucket_uri = self.CreateBucket()
self._RegisterDefaultTopicCreation(bucket_uri.bucket_name)
self.RunGsUtil(['notification', 'create', '-f', 'json', suri(bucket_uri)])
self.RunGsUtil(['notification', 'delete', suri(bucket_uri)])