|
7 | 7 | from slapdtest import SlapdTestCase, requires_tls |
8 | 8 |
|
9 | 9 | import ldap |
10 | | -from ldap.controls import RequestControlTuples |
| 10 | +from ldap.controls import RequestControl, RequestControlTuples |
11 | 11 | from ldap.controls.pagedresults import SimplePagedResultsControl |
12 | 12 | from ldap.controls.openldap import SearchNoOpControl |
13 | 13 | from ldap.ldapobject import SimpleLDAPObject |
14 | 14 |
|
| 15 | +from pyasn1.error import PyAsn1Error |
| 16 | + |
15 | 17 |
|
16 | 18 | SENTINEL = object() |
17 | 19 |
|
@@ -181,15 +183,46 @@ def test_network_timeout_attribute(self): |
181 | 183 | finally: |
182 | 184 | self.set_option(option, old) |
183 | 185 |
|
184 | | - # test is failing with: |
185 | | - # pyasn1.error.SubstrateUnderrunError: Short octet stream on tag decoding |
186 | | - @unittest.expectedFailure |
187 | | - def test_client_controls(self): |
188 | | - self._test_controls(ldap.OPT_CLIENT_CONTROLS) |
| 186 | + # These options hold request controls, but get_option() decodes them |
| 187 | + # with the response classes in KNOWN_RESPONSE_CONTROLS, so only |
| 188 | + # controls with identical request and response schemas round-trip. |
| 189 | + # Returning raw tuples like the module-level API is a 4.0 question, #XXX. |
| 190 | + def _test_controls(self, option): |
| 191 | + self._check_option(option, []) |
189 | 192 |
|
190 | | - @unittest.expectedFailure |
191 | | - def test_server_controls(self): |
192 | | - self._test_controls(ldap.OPT_SERVER_CONTROLS) |
| 193 | + self.set_option(option, [ |
| 194 | + SimplePagedResultsControl(criticality=0, size=5, cookie=b'cookie'), |
| 195 | + ]) |
| 196 | + try: |
| 197 | + paged, = self.get_option(option) |
| 198 | + self.assertIsInstance(paged, SimplePagedResultsControl) |
| 199 | + self.assertEqual(paged.criticality, 0) |
| 200 | + self.assertEqual(paged.size, 5) |
| 201 | + self.assertEqual(paged.cookie, b'cookie') |
| 202 | + finally: |
| 203 | + self.set_option(option, []) |
| 204 | + |
| 205 | + # no request value, SEQUENCE response: critical raises, |
| 206 | + # non-critical is dropped |
| 207 | + self.set_option(option, [SearchNoOpControl(criticality=1)]) |
| 208 | + try: |
| 209 | + with self.assertRaises(PyAsn1Error): |
| 210 | + self.get_option(option) |
| 211 | + finally: |
| 212 | + self.set_option(option, []) |
| 213 | + self.set_option(option, [SearchNoOpControl(criticality=0)]) |
| 214 | + try: |
| 215 | + self.assertEqual(self.get_option(option), []) |
| 216 | + finally: |
| 217 | + self.set_option(option, []) |
| 218 | + |
| 219 | + with self.assertRaises(TypeError): |
| 220 | + self.set_option(option, object) |
| 221 | + with self.assertRaises(TypeError): |
| 222 | + # data must be bytes or None |
| 223 | + self.set_option(option, [ |
| 224 | + RequestControl(TEST_CTRL[0][0], TEST_CTRL[0][1], 'data'), |
| 225 | + ]) |
193 | 226 |
|
194 | 227 |
|
195 | 228 | if __name__ == '__main__': |
|
0 commit comments