1#-------------------------------------------------------------------------- 2# 3# Copyright (c) Microsoft Corporation. All rights reserved. 4# 5# The MIT License (MIT) 6# 7# Permission is hereby granted, free of charge, to any person obtaining a copy 8# of this software and associated documentation files (the ""Software""), to deal 9# in the Software without restriction, including without limitation the rights 10# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 11# copies of the Software, and to permit persons to whom the Software is 12# furnished to do so, subject to the following conditions: 13# 14# The above copyright notice and this permission notice shall be included in 15# all copies or substantial portions of the Software. 16# 17# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 18# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 19# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 20# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 21# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 22# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN 23# THE SOFTWARE. 24# 25#-------------------------------------------------------------------------- 26 27import unittest 28 29from msrest.paging import Paged 30 31class FakePaged(Paged): 32 _attribute_map = { 33 'next_link': {'key': 'nextLink', 'type': 'str'}, 34 'current_page': {'key': 'value', 'type': '[str]'} 35 } 36 37 def __init__(self, *args, **kwargs): 38 super(FakePaged, self).__init__(*args, **kwargs) 39 40class TestPaging(unittest.TestCase): 41 42 def test_basic_paging(self): 43 44 def internal_paging(next_link=None, raw=False): 45 if not next_link: 46 return { 47 'nextLink': 'page2', 48 'value': ['value1.0', 'value1.1'] 49 } 50 else: 51 return { 52 'nextLink': None, 53 'value': ['value2.0', 'value2.1'] 54 } 55 56 deserialized = FakePaged(internal_paging, {}) 57 result_iterated = list(deserialized) 58 self.assertListEqual( 59 ['value1.0', 'value1.1', 'value2.0', 'value2.1'], 60 result_iterated 61 ) 62 63 def test_advance_paging(self): 64 65 def internal_paging(next_link=None, raw=False): 66 if not next_link: 67 return { 68 'nextLink': 'page2', 69 'value': ['value1.0', 'value1.1'] 70 } 71 else: 72 return { 73 'nextLink': None, 74 'value': ['value2.0', 'value2.1'] 75 } 76 77 deserialized = FakePaged(internal_paging, {}) 78 page1 = deserialized.advance_page() 79 self.assertListEqual( 80 ['value1.0', 'value1.1'], 81 page1 82 ) 83 page2 = deserialized.advance_page() 84 self.assertListEqual( 85 ['value2.0', 'value2.1'], 86 page2 87 ) 88 with self.assertRaises(StopIteration): 89 deserialized.advance_page() 90 91 def test_get_paging(self): 92 93 def internal_paging(next_link=None, raw=False): 94 if not next_link: 95 return { 96 'nextLink': 'page2', 97 'value': ['value1.0', 'value1.1'] 98 } 99 elif next_link == 'page2': 100 return { 101 'nextLink': 'page3', 102 'value': ['value2.0', 'value2.1'] 103 } 104 else: 105 return { 106 'nextLink': None, 107 'value': ['value3.0', 'value3.1'] 108 } 109 110 deserialized = FakePaged(internal_paging, {}) 111 page2 = deserialized.get('page2') 112 self.assertListEqual( 113 ['value2.0', 'value2.1'], 114 page2 115 ) 116 page3 = deserialized.get('page3') 117 self.assertListEqual( 118 ['value3.0', 'value3.1'], 119 page3 120 ) 121 122 def test_reset_paging(self): 123 124 def internal_paging(next_link=None, raw=False): 125 if not next_link: 126 return { 127 'nextLink': 'page2', 128 'value': ['value1.0', 'value1.1'] 129 } 130 else: 131 return { 132 'nextLink': None, 133 'value': ['value2.0', 'value2.1'] 134 } 135 136 deserialized = FakePaged(internal_paging, {}) 137 deserialized.reset() 138 result_iterated = list(deserialized) 139 self.assertListEqual( 140 ['value1.0', 'value1.1', 'value2.0', 'value2.1'], 141 result_iterated 142 ) 143 144 deserialized = FakePaged(internal_paging, {}) 145 # Push the iterator to the last element 146 for element in deserialized: 147 if element == "value2.0": 148 break 149 deserialized.reset() 150 result_iterated = list(deserialized) 151 self.assertListEqual( 152 ['value1.0', 'value1.1', 'value2.0', 'value2.1'], 153 result_iterated 154 ) 155 156 def test_none_value(self): 157 def internal_paging(next_link=None, raw=False): 158 return { 159 'nextLink': None, 160 'value': None 161 } 162 163 deserialized = FakePaged(internal_paging, {}) 164 result_iterated = list(deserialized) 165 self.assertEqual(len(result_iterated), 0) 166