Added gsutil to depot tools

third_party/gsutil/boto/tests/unit/cloudfront/test_invalidation_list.py

Index: third_party/gsutil/boto/tests/unit/cloudfront/test_invalidation_list.py
diff --git a/third_party/gsutil/boto/tests/unit/cloudfront/test_invalidation_list.py b/third_party/gsutil/boto/tests/unit/cloudfront/test_invalidation_list.py
new file mode 100644
index 0000000000000000000000000000000000000000..2801178c500a5779dd2d6355361ed7c5341c4302
--- /dev/null
+++ b/third_party/gsutil/boto/tests/unit/cloudfront/test_invalidation_list.py
@@ -0,0 +1,113 @@
+#!/usr/bin/env python
+import random
+import string
+from tests.unit import unittest
+
+import mock
+import boto
+
+
+RESPONSE_TEMPLATE = r"""
+<InvalidationList>
+   <Marker/>
+   <NextMarker>%(next_marker)s</NextMarker>
+   <MaxItems>%(max_items)s</MaxItems>
+   <IsTruncated>%(is_truncated)s</IsTruncated>
+   %(inval_summaries)s
+</InvalidationList>
+"""
+
+INVAL_SUMMARY_TEMPLATE = r"""
+   <InvalidationSummary>
+      <Id>%(cfid)s</Id>
+      <Status>%(status)s</Status>
+   </InvalidationSummary>
+"""
+
+
+class CFInvalidationListTest(unittest.TestCase):
+
+    cloudfront = True
+
+    def setUp(self):
+        self.cf = boto.connect_cloudfront('aws.aws_access_key_id',
+                                          'aws.aws_secret_access_key')
+
+    def _get_random_id(self, length=14):
+        return ''.join([random.choice(string.ascii_letters) for i in
+                        range(length)])
+
+    def _group_iter(self, iterator, n):
+        accumulator = []
+        for item in iterator:
+            accumulator.append(item)
+            if len(accumulator) == n:
+                yield accumulator
+                accumulator = []
+        if len(accumulator) != 0:
+            yield accumulator
+
+    def _get_mock_responses(self, num, max_items):
+        max_items = min(max_items, 100)
+        cfid_groups = list(self._group_iter([self._get_random_id() for i in
+                                             range(num)], max_items))
+        cfg = dict(status='Completed', max_items=max_items, next_marker='')
+        responses = []
+        is_truncated = 'true'
+        for i, group in enumerate(cfid_groups):
+            next_marker = group[-1]
+            if (i + 1) == len(cfid_groups):
+                is_truncated = 'false'
+                next_marker = ''
+            invals = ''
+            cfg.update(dict(next_marker=next_marker,
+                            is_truncated=is_truncated))
+            for cfid in group:
+                cfg.update(dict(cfid=cfid))
+                invals += INVAL_SUMMARY_TEMPLATE % cfg
+            cfg.update(dict(inval_summaries=invals))
+            mock_response = mock.Mock()
+            mock_response.read.return_value = RESPONSE_TEMPLATE % cfg
+            mock_response.status = 200
+            responses.append(mock_response)
+        return responses
+
+    def test_manual_pagination(self, num_invals=30, max_items=4):
+        """
+        Test that paginating manually works properly
+        """
+        self.assertGreater(num_invals, max_items)
+        responses = self._get_mock_responses(num=num_invals,
+                                             max_items=max_items)
+        self.cf.make_request = mock.Mock(side_effect=responses)
+        ir = self.cf.get_invalidation_requests('dist-id-here',
+                                               max_items=max_items)
+        all_invals = list(ir)
+        self.assertEqual(len(all_invals), max_items)
+        while ir.is_truncated:
+            ir = self.cf.get_invalidation_requests('dist-id-here',
+                                                   marker=ir.next_marker,
+                                                   max_items=max_items)
+            invals = list(ir)
+            self.assertLessEqual(len(invals), max_items)
+            all_invals.extend(invals)
+        remainder = num_invals % max_items
+        if remainder != 0:
+            self.assertEqual(len(invals), remainder)
+        self.assertEqual(len(all_invals), num_invals)
+
+    def test_auto_pagination(self, num_invals=1024):
+        """
+        Test that auto-pagination works properly
+        """
+        max_items = 100
+        self.assertGreaterEqual(num_invals, max_items)
+        responses = self._get_mock_responses(num=num_invals,
+                                             max_items=max_items)
+        self.cf.make_request = mock.Mock(side_effect=responses)
+        ir = self.cf.get_invalidation_requests('dist-id-here')
+        self.assertEqual(len(ir._inval_cache), max_items)
+        self.assertEqual(len(list(ir)), num_invals)
+
+if __name__ == '__main__':
+    unittest.main()