tests: add explicit coverage for retry members (#308)

See also: #283 This PR adds explicit coverage for each member of `google.cloud.storage.retry`. Together with PR #284, should return this package to 100% unit test coverage.

Author: tseaver

noxfile.py

@@ -70,7 +70,9 @@ def lint_setup_py(session):
 
 def default(session):
  # Install all test dependencies, then install this package in-place.
- session.install("mock", "pytest", "pytest-cov")
+ session.install(
+ "mock", "pytest", "pytest-cov",
+ )
  session.install("-e", ".")
 
  # Run py.test against the unit tests.
@@ -144,7 +146,7 @@ def cover(session):
  test runs (not system test runs), and then erases coverage data.
  """
  session.install("coverage", "pytest-cov")
- session.run("coverage", "report", "--show-missing", "--fail-under=99")
+ session.run("coverage", "report", "--show-missing", "--fail-under=100")
 
  session.run("coverage", "erase")
 

synth.py

@@ -25,7 +25,7 @@
 # Add templated files
 # ----------------------------------------------------------------------------
 templated_files = common.py_library(
- cov_level=99,
+ cov_level=100,
  system_test_external_dependencies=[
  "google-cloud-iam",
  "google-cloud-pubsub < 2.0.0",

tests/unit/test_retry.py

@@ -14,64 +14,222 @@
 
 import unittest
 
-from google.cloud.storage.retry import DEFAULT_RETRY
-from google.cloud.storage.retry import DEFAULT_RETRY_IF_GENERATION_SPECIFIED
-from google.cloud.storage.retry import DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED
-from google.cloud.storage.retry import DEFAULT_RETRY_IF_ETAG_IN_JSON
+import mock
+
+
+class Test_should_retry(unittest.TestCase):
+ def _call_fut(self, exc):
+ from google.cloud.storage import retry
+
+ return retry._should_retry(exc)
+
+ def test_w_retryable_types(self):
+ from google.cloud.storage import retry
+
+ for exc_type in retry._RETRYABLE_TYPES:
+ exc = exc_type("testing")
+ self.assertTrue(self._call_fut(exc))
+
+ def test_w_google_api_call_error_hit(self):
+ from google.api_core import exceptions
+
+ exc = exceptions.GoogleAPICallError("testing")
+ exc.code = 408
+ self.assertTrue(self._call_fut(exc))
+
+ def test_w_google_api_call_error_miss(self):
+ from google.api_core import exceptions
+
+ exc = exceptions.GoogleAPICallError("testing")
+ exc.code = 999
+ self.assertFalse(self._call_fut(exc))
+
+ def test_w_requests_connection_error(self):
+ exc = ValueError("testing")
+ self.assertFalse(self._call_fut(exc))
 
 
 class TestConditionalRetryPolicy(unittest.TestCase):
+ def _make_one(self, retry_policy, conditional_predicate, required_kwargs):
+ from google.cloud.storage import retry
+
+ return retry.ConditionalRetryPolicy(
+ retry_policy, conditional_predicate, required_kwargs
+ )
+
+ def test_ctor(self):
+ retry_policy = mock.Mock()
+ conditional_predicate = mock.Mock()
+ required_kwargs = ("kwarg",)
+
+ policy = self._make_one(retry_policy, conditional_predicate, required_kwargs)
+
+ self.assertIs(policy.retry_policy, retry_policy)
+ self.assertIs(policy.conditional_predicate, conditional_predicate)
+ self.assertEqual(policy.required_kwargs, required_kwargs)
+
+ def test_get_retry_policy_if_conditions_met_single_kwarg_hit(self):
+ retry_policy = mock.Mock()
+ conditional_predicate = mock.Mock(return_value=True)
+ required_kwargs = ("foo",)
+ policy = self._make_one(retry_policy, conditional_predicate, required_kwargs)
+
+ kwargs = {"foo": 1, "bar": 2, "baz": 3}
+ result = policy.get_retry_policy_if_conditions_met(**kwargs)
+
+ self.assertIs(result, retry_policy)
+
+ conditional_predicate.assert_called_once_with(1)
+
+ def test_get_retry_policy_if_conditions_met_multiple_kwargs_miss(self):
+ retry_policy = mock.Mock()
+ conditional_predicate = mock.Mock(return_value=False)
+ required_kwargs = ("foo", "bar")
+ policy = self._make_one(retry_policy, conditional_predicate, required_kwargs)
+
+ kwargs = {"foo": 1, "bar": 2, "baz": 3}
+ result = policy.get_retry_policy_if_conditions_met(**kwargs)
+
+ self.assertIsNone(result)
+
+ conditional_predicate.assert_called_once_with(1, 2)
+
+
+class Test_is_generation_specified(unittest.TestCase):
+ def _call_fut(self, query_params):
+ from google.cloud.storage import retry
+
+ return retry.is_generation_specified(query_params)
+
+ def test_w_empty(self):
+ query_params = {}
+
+ self.assertFalse(self._call_fut(query_params))
+
+ def test_w_generation(self):
+ query_params = {"generation": 123}
+
+ self.assertTrue(self._call_fut(query_params))
+
+ def test_wo_generation_w_if_generation_match(self):
+ query_params = {"if_generation_match": 123}
+
+ self.assertTrue(self._call_fut(query_params))
+
+
+class Test_is_metageneration_specified(unittest.TestCase):
+ def _call_fut(self, query_params):
+ from google.cloud.storage import retry
+
+ return retry.is_metageneration_specified(query_params)
+
+ def test_w_empty(self):
+ query_params = {}
+
+ self.assertFalse(self._call_fut(query_params))
+
+ def test_w_if_metageneration_match(self):
+ query_params = {"if_metageneration_match": 123}
+
+ self.assertTrue(self._call_fut(query_params))
+
+
+class Test_is_etag_in_json(unittest.TestCase):
+ def _call_fut(self, data):
+ from google.cloud.storage import retry
+
+ return retry.is_etag_in_json(data)
+
+ @staticmethod
+ def _make_json_data(**kw):
+ import json
+
+ return json.dumps(kw)
+
+ def test_w_empty(self):
+ data = self._make_json_data()
+
+ self.assertFalse(self._call_fut(data))
+
+ def test_w_etag_in_data(self):
+ data = self._make_json_data(etag="123")
+
+ self.assertTrue(self._call_fut(data))
+
+ def test_w_empty_data(self):
+ data = ""
+
+ self.assertFalse(self._call_fut(data))
+
+
+class Test_default_conditional_retry_policies(unittest.TestCase):
  def test_is_generation_specified_match_metageneration(self):
- conditional_policy = DEFAULT_RETRY_IF_GENERATION_SPECIFIED
+ from google.cloud.storage import retry
+
+ conditional_policy = retry.DEFAULT_RETRY_IF_GENERATION_SPECIFIED
  policy = conditional_policy.get_retry_policy_if_conditions_met(
  query_params={"if_generation_match": 1}
  )
- self.assertEqual(policy, DEFAULT_RETRY)
+ self.assertEqual(policy, retry.DEFAULT_RETRY)
 
  def test_is_generation_specified_match_generation(self):
- conditional_policy = DEFAULT_RETRY_IF_GENERATION_SPECIFIED
+ from google.cloud.storage import retry
+
+ conditional_policy = retry.DEFAULT_RETRY_IF_GENERATION_SPECIFIED
  policy = conditional_policy.get_retry_policy_if_conditions_met(
  query_params={"generation": 1}
  )
- self.assertEqual(policy, DEFAULT_RETRY)
+ self.assertEqual(policy, retry.DEFAULT_RETRY)
 
  def test_is_generation_specified_mismatch(self):
- conditional_policy = DEFAULT_RETRY_IF_GENERATION_SPECIFIED
+ from google.cloud.storage import retry
+
+ conditional_policy = retry.DEFAULT_RETRY_IF_GENERATION_SPECIFIED
  policy = conditional_policy.get_retry_policy_if_conditions_met(
  query_params={"if_metageneration_match": 1}
  )
  self.assertEqual(policy, None)
 
  def test_is_metageneration_specified_match(self):
- conditional_policy = DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED
+ from google.cloud.storage import retry
+
+ conditional_policy = retry.DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED
  policy = conditional_policy.get_retry_policy_if_conditions_met(
  query_params={"if_metageneration_match": 1}
  )
- self.assertEqual(policy, DEFAULT_RETRY)
+ self.assertEqual(policy, retry.DEFAULT_RETRY)
 
  def test_is_metageneration_specified_mismatch(self):
- conditional_policy = DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED
+ from google.cloud.storage import retry
+
+ conditional_policy = retry.DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED
  policy = conditional_policy.get_retry_policy_if_conditions_met(
  query_params={"if_generation_match": 1}
  )
  self.assertEqual(policy, None)
 
  def test_is_etag_in_json_etag_match(self):
- conditional_policy = DEFAULT_RETRY_IF_ETAG_IN_JSON
+ from google.cloud.storage import retry
+
+ conditional_policy = retry.DEFAULT_RETRY_IF_ETAG_IN_JSON
  policy = conditional_policy.get_retry_policy_if_conditions_met(
  query_params={"if_generation_match": 1}, data='{"etag": "12345678"}'
  )
- self.assertEqual(policy, DEFAULT_RETRY)
+ self.assertEqual(policy, retry.DEFAULT_RETRY)
 
  def test_is_etag_in_json_mismatch(self):
- conditional_policy = DEFAULT_RETRY_IF_ETAG_IN_JSON
+ from google.cloud.storage import retry
+
+ conditional_policy = retry.DEFAULT_RETRY_IF_ETAG_IN_JSON
  policy = conditional_policy.get_retry_policy_if_conditions_met(
  query_params={"if_generation_match": 1}, data="{}"
  )
  self.assertEqual(policy, None)
 
  def test_is_meta_or_etag_in_json_invalid(self):
- conditional_policy = DEFAULT_RETRY_IF_ETAG_IN_JSON
+ from google.cloud.storage import retry
+
+ conditional_policy = retry.DEFAULT_RETRY_IF_ETAG_IN_JSON
  policy = conditional_policy.get_retry_policy_if_conditions_met(
  query_params={"if_generation_match": 1}, data="I am invalid JSON!"
  )