[PY3][Autotest]Moving pubsub unittest off mox
BUG=chromium:990593
TEST=python unittest_suite.py --py_version=2 --py_version=3
Change-Id: I11b097421bd153420522dba6b4bb3ce25a982e62
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/third_party/autotest/+/3010703
Tested-by: Derek Beckett <dbeckett@chromium.org>
Auto-Submit: Derek Beckett <dbeckett@chromium.org>
Commit-Queue: Derek Beckett <dbeckett@chromium.org>
Reviewed-by: Jeremy Bettis <jbettis@chromium.org>
diff --git a/site_utils/pubsub_utils_unittest.py b/site_utils/pubsub_utils_unittest.py
index 3bfde6c..c33031f 100644
--- a/site_utils/pubsub_utils_unittest.py
+++ b/site_utils/pubsub_utils_unittest.py
@@ -9,7 +9,10 @@
import os
import unittest
-import mox
+from mock import patch
+from mock import MagicMock
+
+import common
# TODO(crbug.com/1050892): The unittests rely on apiclient in chromite.
import autotest_lib.utils.frozen_chromite # pylint: disable=unused-import
@@ -19,7 +22,7 @@
from oauth2client.client import GoogleCredentials
from googleapiclient.errors import UnknownApiNameOrVersion
-import pubsub_utils
+from autotest_lib.site_utils import pubsub_utils
_TEST_CLOUD_SERVICE_ACCOUNT_FILE = '/tmp/test-credential'
@@ -74,95 +77,99 @@
return msg_payload
-class PubSubTests(mox.MoxTestBase):
+class PubSubTests(unittest.TestCase):
"""Tests for pubsub related functios."""
+ def setUp(self):
+ patcher = patch.object(os.path, 'isfile')
+ self.isfile_mock = patcher.start()
+ self.addCleanup(patcher.stop)
+ creds_patcher = patch.object(GoogleCredentials, 'from_stream')
+ self.creds_mock = creds_patcher.start()
+ self.addCleanup(creds_patcher.stop)
+
def test_pubsub_with_no_service_account(self):
"""Test getting the pubsub service"""
- self.mox.StubOutWithMock(os.path, 'isfile')
- 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()
+ self.isfile_mock.return_value = False
with self.assertRaises(pubsub_utils.PubSubException):
pubsub_utils.PubSubClient(_TEST_CLOUD_SERVICE_ACCOUNT_FILE)
- self.mox.VerifyAll()
+ self.isfile_mock.assert_called_with(_TEST_CLOUD_SERVICE_ACCOUNT_FILE)
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(_TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(True)
- GoogleCredentials.from_stream(
- _TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndRaise(
- ApplicationDefaultCredentialsError())
- self.mox.ReplayAll()
+
+ self.isfile_mock.return_value = True
+ self.creds_mock.side_effect = ApplicationDefaultCredentialsError
+
with self.assertRaises(pubsub_utils.PubSubException):
pubsub_utils.PubSubClient(_TEST_CLOUD_SERVICE_ACCOUNT_FILE)
- self.mox.VerifyAll()
+
+ self.creds_mock.assert_called_with(_TEST_CLOUD_SERVICE_ACCOUNT_FILE)
+ self.isfile_mock.assert_called_with(_TEST_CLOUD_SERVICE_ACCOUNT_FILE)
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(_TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(True)
- credentials = self.mox.CreateMock(GoogleCredentials)
- GoogleCredentials.from_stream(
- _TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(credentials)
- credentials.create_scoped_required().AndReturn(True)
- credentials.create_scoped(pubsub_utils.PUBSUB_SCOPES).AndReturn(
- credentials)
- self.mox.StubOutWithMock(discovery, 'build')
- 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(
- _TEST_CLOUD_SERVICE_ACCOUNT_FILE)
- pubsub_client.publish_notifications('test_topic', [msg])
- self.mox.VerifyAll()
+ self.isfile_mock.return_value = True
+ credentials = MagicMock(GoogleCredentials)
+ self.creds_mock.return_value = credentials
+
+ credentials.create_scoped_required.return_value = True
+ credentials.create_scoped.return_value = credentials
+
+ with patch.object(discovery, 'build') as discovery_mock:
+ discovery_mock.side_effect = UnknownApiNameOrVersion
+
+ with self.assertRaises(pubsub_utils.PubSubException):
+ msg = _create_sample_message()
+ pubsub_client = pubsub_utils.PubSubClient(
+ _TEST_CLOUD_SERVICE_ACCOUNT_FILE)
+ pubsub_client.publish_notifications('test_topic', [msg])
+
+ credentials.create_scoped.assert_called_with(
+ pubsub_utils.PUBSUB_SCOPES)
+ discovery_mock.assert_called_with(pubsub_utils.PUBSUB_SERVICE_NAME,
+ pubsub_utils.PUBSUB_VERSION,
+ credentials=credentials)
+ self.creds_mock.assert_called_with(_TEST_CLOUD_SERVICE_ACCOUNT_FILE)
+ self.isfile_mock.assert_called_with(_TEST_CLOUD_SERVICE_ACCOUNT_FILE)
def test_publish_notifications(self):
"""Test getting the pubsub service"""
- self.mox.StubOutWithMock(os.path, 'isfile')
- self.mox.StubOutWithMock(GoogleCredentials, 'from_stream')
- os.path.isfile(_TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(True)
- credentials = self.mox.CreateMock(GoogleCredentials)
- GoogleCredentials.from_stream(
- _TEST_CLOUD_SERVICE_ACCOUNT_FILE).AndReturn(credentials)
- credentials.create_scoped_required().AndReturn(True)
- credentials.create_scoped(pubsub_utils.PUBSUB_SCOPES).AndReturn(
- 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.DEFAULT_PUBSUB_NUM_RETRIES,
- # use tuple ('123') instead of list just for easy to
- # write the test.
- ret_val={'messageIds': ('123')}))
+ self.isfile_mock.return_value = True
+ credentials = MagicMock(GoogleCredentials)
+ self.creds_mock.return_value = credentials
- self.mox.ReplayAll()
- pubsub_client = pubsub_utils.PubSubClient(
- _TEST_CLOUD_SERVICE_ACCOUNT_FILE)
- msg_ids = pubsub_client.publish_notifications('test_topic', [msg])
- self.assertEquals(('123'), msg_ids)
+ credentials.create_scoped_required.return_value = True
+ credentials.create_scoped.return_value = credentials
- self.mox.VerifyAll()
+ with patch.object(discovery, 'build') as discovery_mock:
+ msg = _create_sample_message()
+ discovery_mock.return_value = 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')})
+
+ pubsub_client = pubsub_utils.PubSubClient(
+ _TEST_CLOUD_SERVICE_ACCOUNT_FILE)
+ msg_ids = pubsub_client.publish_notifications('test_topic', [msg])
+ self.assertEquals(('123'), msg_ids)
+
+ credentials.create_scoped.assert_called_with(
+ pubsub_utils.PUBSUB_SCOPES)
+ discovery_mock.assert_called_with(pubsub_utils.PUBSUB_SERVICE_NAME,
+ pubsub_utils.PUBSUB_VERSION,
+ credentials=credentials)
+ self.creds_mock.assert_called_with(_TEST_CLOUD_SERVICE_ACCOUNT_FILE)
+ self.isfile_mock.assert_called_with(_TEST_CLOUD_SERVICE_ACCOUNT_FILE)
if __name__ == '__main__':
