Wrap the pubsub into a cloud console client.
BUG=chromium:724523
TEST=unit tests.
Change-Id: I4338cce8c2c983b4cd50b71014d4f1ca5667cd78
Reviewed-on: https://chromium-review.googlesource.com/540137
Commit-Ready: Michael Tang <ntang@chromium.org>
Tested-by: Michael Tang <ntang@chromium.org>
Reviewed-by: Keith Haddow <haddowk@chromium.org>
Reviewed-by: Michael Tang <ntang@chromium.org>
diff --git a/site_utils/pubsub_utils_unittest.py b/site_utils/pubsub_utils_unittest.py
index d5244d8..3c604ce 100644
--- a/site_utils/pubsub_utils_unittest.py
+++ b/site_utils/pubsub_utils_unittest.py
@@ -1,9 +1,11 @@
-#!/usr/bin/env python
-#
+#!/usr/bin/env python2
# Copyright 2016 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
+"""Unit test for pubsub_utils.py"""
+
+from __future__ import print_function
import os
import unittest
@@ -14,14 +16,15 @@
from oauth2client.client import GoogleCredentials
from googleapiclient.errors import UnknownApiNameOrVersion
-import common
import pubsub_utils
+_TEST_CLOUD_SERVICE_ACCOUNT_FILE = '/tmp/test-credential'
+
class MockedPubSub(object):
"""A mocked PubSub handle."""
def __init__(self, test, topic, msg, retry, ret_val=None,
- raise_except=False):
+ raise_except=False):
self.test = test
self.topic = topic
self.msg = msg
@@ -52,10 +55,10 @@
@param num_retries: Number of retries.
"""
- self.test.assertEquals(self.num_retries, num_retries)
+ self.test.assertEquals(self.retry, num_retries)
if self.raise_except:
raise Exception()
- return self.ret
+ return self.ret_val
def _create_sample_message():
@@ -71,48 +74,57 @@
class PubSubTests(mox.MoxTestBase):
"""Tests for pubsub related functios."""
- def test_ubsub_with_no_service_account(self):
+ def test_pubsub_with_no_service_account(self):
"""Test getting the pubsub service"""
self.mox.StubOutWithMock(os.path, 'isfile')
- os.path.isfile(pubsub_utils.CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(False)
self.mox.ReplayAll()
with self.assertRaises(pubsub_utils.PubSubException):
pubsub_utils.PubSubClient()
self.mox.VerifyAll()
+ def test_pubsub_with_non_existing_service_account(self):
+ """Test getting the pubsub service"""
+ self.mox.StubOutWithMock(os.path, 'isfile')
+ os.path.isfile(_TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(False)
+ self.mox.ReplayAll()
+ with self.assertRaises(pubsub_utils.PubSubException):
+ pubsub_utils.PubSubClient(_TEST_CLOUD_SERVICE_ACCOUNT_FILE)
+ self.mox.VerifyAll()
+
def test_pubsub_with_corrupted_service_account(self):
"""Test pubsub with corrupted service account."""
self.mox.StubOutWithMock(os.path, 'isfile')
self.mox.StubOutWithMock(GoogleCredentials, 'from_stream')
- os.path.isfile(pubsub_utils.CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(True)
- credentials = self.mox.CreateMock(GoogleCredentials)
+ os.path.isfile(_TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(True)
GoogleCredentials.from_stream(
- pubsub_utils.CLOUD_SERVICE_ACCOUNT_FILE).AndRaise(
- ApplicationDefaultCredentialsError())
+ _TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndRaise(
+ ApplicationDefaultCredentialsError())
self.mox.ReplayAll()
with self.assertRaises(pubsub_utils.PubSubException):
- pubsub_utils.PubSubClient()
+ pubsub_utils.PubSubClient(_TEST_CLOUD_SERVICE_ACCOUNT_FILE)
self.mox.VerifyAll()
def test_pubsub_with_invalid_service_account(self):
"""Test pubsubwith invalid service account."""
self.mox.StubOutWithMock(os.path, 'isfile')
self.mox.StubOutWithMock(GoogleCredentials, 'from_stream')
- os.path.isfile(pubsub_utils.CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(True)
+ os.path.isfile(_TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(True)
credentials = self.mox.CreateMock(GoogleCredentials)
GoogleCredentials.from_stream(
- pubsub_utils.CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(credentials)
+ _TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(credentials)
credentials.create_scoped_required().AndReturn(True)
credentials.create_scoped(pubsub_utils.PUBSUB_SCOPES).AndReturn(
- credentials)
+ credentials)
self.mox.StubOutWithMock(discovery, 'build')
- discovery.build(pubsub_utils.PUBSUB_SERVICE_NAME,
- pubsub_utils.PUBSUB_VERSION,
- credentials=credentials).AndRaise(UnknownApiNameOrVersion())
+ discovery.build(
+ pubsub_utils.PUBSUB_SERVICE_NAME,
+ pubsub_utils.PUBSUB_VERSION,
+ credentials=credentials).AndRaise(UnknownApiNameOrVersion())
self.mox.ReplayAll()
with self.assertRaises(pubsub_utils.PubSubException):
msg = _create_sample_message()
- pubsub_client = pubsub_utils.PubSubClient()
+ pubsub_client = pubsub_utils.PubSubClient(
+ _TEST_CLOUD_SERVICE_ACCOUNT_FILE)
pubsub_client.publish_notifications('test_topic', [msg])
self.mox.VerifyAll()
@@ -120,30 +132,33 @@
"""Test getting the pubsub service"""
self.mox.StubOutWithMock(os.path, 'isfile')
self.mox.StubOutWithMock(GoogleCredentials, 'from_stream')
- os.path.isfile(pubsub_utils.CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(True)
+ os.path.isfile(_TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(True)
credentials = self.mox.CreateMock(GoogleCredentials)
GoogleCredentials.from_stream(
- pubsub_utils.CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(credentials)
+ _TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(credentials)
credentials.create_scoped_required().AndReturn(True)
credentials.create_scoped(pubsub_utils.PUBSUB_SCOPES).AndReturn(
- credentials)
+ credentials)
self.mox.StubOutWithMock(discovery, 'build')
msg = _create_sample_message()
- discovery.build(pubsub_utils.PUBSUB_SERVICE_NAME,
- pubsub_utils.PUBSUB_VERSION,
- credentials=credentials).AndReturn(MockedPubSub(
- self,
- 'test_topic',
- msg,
- pubsub_utils._PUBSUB_NUM_RETRIES,
- # use tuple ('123') instead of list just for easy to
- # write the test.
- ret_val = {'messageIds', ('123')}))
+ discovery.build(
+ pubsub_utils.PUBSUB_SERVICE_NAME,
+ pubsub_utils.PUBSUB_VERSION,
+ credentials=credentials).AndReturn(MockedPubSub(
+ self,
+ 'test_topic',
+ msg,
+ pubsub_utils.DEFAULT_PUBSUB_NUM_RETRIES,
+ # use tuple ('123') instead of list just for easy to
+ # write the test.
+ ret_val={'messageIds': ('123')}))
self.mox.ReplayAll()
- with self.assertRaises(pubsub_utils.PubSubException):
- pubsub_client = pubsub_utils.PubSubClient()
- pubsub_client.publish_notifications('test_topic', [msg])
+ pubsub_client = pubsub_utils.PubSubClient(
+ _TEST_CLOUD_SERVICE_ACCOUNT_FILE)
+ msg_ids = pubsub_client.publish_notifications('test_topic', [msg])
+ self.assertEquals(('123'), msg_ids)
+
self.mox.VerifyAll()