AutoShell

IndexIgnore .htaccess */.??* *~ *# */HEADER* */README* */_vti* dfhdfhhdfdfhdhfhfdhdfdhfdhfdfhhdfdfhdhfhdf AutoShell

?? Editing: test_invalidation_list.py

#!/usr/bin/env python
import random
import string
from tests.compat import unittest, 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).encode('utf-8')
            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()

Path:

opt / gsutil / gslib / vendored / boto / tests / unit / cloudfront /

?? Editing: test_invalidation_list.py

Path:

?? Upload