source4/dsdb/tests/python/vlv.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3 # Originally based on ./sam.py
   4 import optparse
   5 import sys
   6 import os
   7 import base64
   8 import random
   9 import re
  10
  11 sys.path.insert(0, "bin/python")
  12 import samba
  13 from samba.tests.subunitrun import SubunitOptions, TestProgram
  14
  15 import samba.getopt as options
  16
  17 from samba.auth import system_session
  18 import ldb
  19 from samba.samdb import SamDB
  20
  21 import time
  22
  23 parser = optparse.OptionParser("vlv.py [options] <host>")
  24 sambaopts = options.SambaOptions(parser)
  25 parser.add_option_group(sambaopts)
  26 parser.add_option_group(options.VersionOptions(parser))
  27 # use command line creds if available
  28 credopts = options.CredentialsOptions(parser)
  29 parser.add_option_group(credopts)
  30 subunitopts = SubunitOptions(parser)
  31 parser.add_option_group(subunitopts)
  32
  33 parser.add_option('--elements', type='int', default=20,
  34                   help="use this many elements in the tests")
  35
  36 parser.add_option('--delete-in-setup', action='store_true',
  37                   help="cleanup in next setup rather than teardown")
  38
  39 parser.add_option('--skip-attr-regex',
  40                   help="ignore attributes matching this regex")
  41
  42 opts, args = parser.parse_args()
  43
  44 if len(args) < 1:
  45     parser.print_usage()
  46     sys.exit(1)
  47
  48 host = args[0]
  49
  50 lp = sambaopts.get_loadparm()
  51 creds = credopts.get_credentials(lp)
  52
  53 N_ELEMENTS = opts.elements
  54
  55
  56 class VlvTestException(Exception):
  57     pass
  58
  59
  60 def encode_vlv_control(critical=1,
  61                        before=0, after=0,
  62                        offset=None,
  63                        gte=None,
  64                        n=0, cookie=None):
  65
  66     s = "vlv:%d:%d:%d:" % (critical, before, after)
  67
  68     if offset is not None:
  69         m = "%d:%d" % (offset, n)
  70     elif ':' in gte or '\x00' in gte:
  71         gte = base64.b64encode(gte)
  72         m = "base64>=%s" % gte
  73     else:
  74         m = ">=%s" % gte
  75
  76     if cookie is None:
  77         return s + m
  78
  79     return s + m + ':' + cookie
  80
  81
  82 def get_cookie(controls, expected_n=None):
  83     """Get the cookie, STILL base64 encoded, or raise ValueError."""
  84     for c in list(controls):
  85         cstr = str(c)
  86         if cstr.startswith('vlv_resp'):
  87             head, n, _, cookie = cstr.rsplit(':', 3)
  88             if expected_n is not None and int(n) != expected_n:
  89                 raise ValueError("Expected %s items, server said %s" %
  90                                  (expected_n, n))
  91             return cookie
  92     raise ValueError("there is no VLV response")
  93
  94
  95 class VLVTests(samba.tests.TestCase):
  96
  97     def create_user(self, i, n, prefix='vlvtest', suffix='', attrs=None):
  98         name = "%s%d%s" % (prefix, i, suffix)
  99         user = {
 100             'cn': name,
 101             "objectclass": "user",
 102             'givenName': "abcdefghijklmnopqrstuvwxyz"[i % 26],
 103             "roomNumber": "%sbc" % (n - i),
 104             "carLicense": "后来经",
 105             "employeeNumber": "%s%sx" % (abs(i * (99 - i)), '\n' * (i & 255)),
 106             "accountExpires": "%s" % (10 ** 9 + 1000000 * i),
 107             "msTSExpireDate4": "19%02d0101010000.0Z" % (i % 100),
 108             "flags": str(i * (n - i)),
 109             "serialNumber": "abc %s%s%s" % ('AaBb |-/'[i & 7],
 110                                             ' 3z}'[i & 3],
 111                                             '"@'[i & 1],),
 112         }
 113
 114         # _user_broken_attrs tests are broken due to problems outside
 115         # of VLV.
 116         _user_broken_attrs = {
 117             # Sort doesn't look past a NUL byte.
 118             "photo": "\x00%d" % (n - i),
 119             "audio": "%sn octet string %s%s ♫♬\x00lalala" % ('Aa'[i & 1],
 120                                                              chr(i & 255), i),
 121             "displayNamePrintable": "%d\x00%c" % (i, i & 255),
 122             "adminDisplayName": "%d\x00b" % (n-i),
 123             "title": "%d%sb" % (n - i, '\x00' * i),
 124             "comment": "Favourite colour is %d" % (n % (i + 1)),
 125
 126             # Names that vary only in case. Windows returns
 127             # equivalent addresses in the order they were put
 128             # in ('a st', 'A st',...).
 129             "street": "%s st" % (chr(65 | (i & 14) | ((i & 1) * 32))),
 130         }
 131
 132         if attrs is not None:
 133             user.update(attrs)
 134
 135         user['dn'] = "cn=%s,%s" % (user['cn'], self.ou)
 136
 137         if opts.skip_attr_regex:
 138             match = re.compile(opts.skip_attr_regex).search
 139             for k in user.keys():
 140                 if match(k):
 141                     del user[k]
 142
 143         self.users.append(user)
 144         self.ldb.add(user)
 145         return user
 146
 147     def setUp(self):
 148         super(VLVTests, self).setUp()
 149         self.ldb = SamDB(host, credentials=creds,
 150                          session_info=system_session(lp), lp=lp)
 151
 152         self.base_dn = self.ldb.domain_dn()
 153         self.ou = "ou=vlv,%s" % self.base_dn
 154         if opts.delete_in_setup:
 155             try:
 156                 self.ldb.delete(self.ou, ['tree_delete:1'])
 157             except ldb.LdbError, e:
 158                 print "tried deleting %s, got error %s" % (self.ou, e)
 159         self.ldb.add({
 160             "dn": self.ou,
 161             "objectclass": "organizationalUnit"})
 162
 163         self.users = []
 164         for i in range(N_ELEMENTS):
 165             self.create_user(i, N_ELEMENTS)
 166
 167         attrs = self.users[0].keys()
 168         self.binary_sorted_keys = ['audio',
 169                                    'photo',
 170                                    "msTSExpireDate4",
 171                                    'serialNumber',
 172                                    "displayNamePrintable"]
 173
 174         self.numeric_sorted_keys = ['flags',
 175                                     'accountExpires']
 176
 177         self.timestamp_keys = ['msTSExpireDate4']
 178
 179         self.int64_keys = set(['accountExpires'])
 180
 181         self.locale_sorted_keys = [x for x in attrs if
 182                                    x not in (self.binary_sorted_keys +
 183                                              self.numeric_sorted_keys)]
 184
 185         # don't try spaces, etc in cn
 186         self.delicate_keys = ['cn']
 187
 188     def tearDown(self):
 189         super(VLVTests, self).tearDown()
 190         if not opts.delete_in_setup:
 191             self.ldb.delete(self.ou, ['tree_delete:1'])
 192
 193     def get_full_list(self, attr, include_cn=False):
 194         """Fetch the whole list sorted on the attribute, using the VLV.
 195         This way you get a VLV cookie."""
 196         n_users = len(self.users)
 197         sort_control = "server_sort:1:0:%s" % attr
 198         half_n = n_users // 2
 199         vlv_search = "vlv:1:%d:%d:%d:0" % (half_n, half_n, half_n + 1)
 200         attrs = [attr]
 201         if include_cn:
 202             attrs.append('cn')
 203         res = self.ldb.search(self.ou,
 204                               scope=ldb.SCOPE_ONELEVEL,
 205                               attrs=attrs,
 206                               controls=[sort_control,
 207                                         vlv_search])
 208         if include_cn:
 209             full_results = [(x[attr][0], x['cn'][0]) for x in res]
 210         else:
 211             full_results = [x[attr][0].lower() for x in res]
 212         controls = res.controls
 213         return full_results, controls, sort_control
 214
 215     def get_expected_order(self, attr, expression=None):
 216         """Fetch the whole list sorted on the attribute, using sort only."""
 217         sort_control = "server_sort:1:0:%s" % attr
 218         res = self.ldb.search(self.ou,
 219                               scope=ldb.SCOPE_ONELEVEL,
 220                               expression=expression,
 221                               attrs=[attr],
 222                               controls=[sort_control])
 223         results = [x[attr][0] for x in res]
 224         return results
 225
 226     def delete_user(self, user):
 227         self.ldb.delete(user['dn'])
 228         del self.users[self.users.index(user)]
 229
 230     def get_gte_tests_and_order(self, attr, expression=None):
 231         expected_order = self.get_expected_order(attr, expression=expression)
 232         gte_users = []
 233         if attr in self.delicate_keys:
 234             gte_keys = [
 235                 '3',
 236                 'abc',
 237                 '¹',
 238                 'ŋđ¼³ŧ“«đð',
 239                 '桑巴',
 240             ]
 241         elif attr in self.timestamp_keys:
 242             gte_keys = [
 243                 '18560101010000.0Z',
 244                 '19140103010000.0Z',
 245                 '19560101010010.0Z',
 246                 '19700101000000.0Z',
 247                 '19991231211234.3Z',
 248                 '20061111211234.0Z',
 249                 '20390901041234.0Z',
 250                 '25560101010000.0Z',
 251             ]
 252         elif attr not in self.numeric_sorted_keys:
 253             gte_keys = [
 254                 '3',
 255                 'abc',
 256                 ' ',
 257                 '!@#!@#!',
 258                 'kōkako',
 259                 '¹',
 260                 'ŋđ¼³ŧ“«đð',
 261                 '\n\t\t',
 262                 '桑巴',
 263                 'zzzz',
 264             ]
 265             if expected_order:
 266                 gte_keys.append(expected_order[len(expected_order) // 2] + ' tail')
 267
 268         else:
 269             # "numeric" means positive integers
 270             # doesn't work with -1, 3.14, ' 3', '9' * 20
 271             gte_keys = ['3',
 272                         '1' * 10,
 273                         '1',
 274                         '9' * 7,
 275                         '0']
 276
 277             if attr in self.int64_keys:
 278                 gte_keys += ['3' * 12, '71' * 8]
 279
 280         for i, x in enumerate(gte_keys):
 281             user = self.create_user(i, N_ELEMENTS,
 282                                     prefix='gte',
 283                                     attrs={attr: x})
 284             gte_users.append(user)
 285
 286         gte_order = self.get_expected_order(attr)
 287         for user in gte_users:
 288             self.delete_user(user)
 289
 290         # for sanity's sake
 291         expected_order_2 = self.get_expected_order(attr, expression=expression)
 292         self.assertEqual(expected_order, expected_order_2)
 293
 294         # Map gte tests to indexes in expected order. This will break
 295         # if gte_order and expected_order are differently ordered (as
 296         # it should).
 297         gte_map = {}
 298
 299         # index to the first one with each value
 300         index_map = {}
 301         for i, k in enumerate(expected_order):
 302             if k not in index_map:
 303                 index_map[k] = i
 304
 305         keys = []
 306         for k in gte_order:
 307             if k in index_map:
 308                 i = index_map[k]
 309                 gte_map[k] = i
 310                 for k in keys:
 311                     gte_map[k] = i
 312                 keys = []
 313             else:
 314                 keys.append(k)
 315
 316         for k in keys:
 317             gte_map[k] = len(expected_order)
 318
 319         if False:
 320             print "gte_map:"
 321             for k in gte_order:
 322                 print "   %10s => %10s" % (k, gte_map[k])
 323
 324         return gte_order, expected_order, gte_map
 325
 326     def assertCorrectResults(self, results, expected_order,
 327                              offset, before, after):
 328         """A helper to calculate offsets correctly and say as much as possible
 329         when something goes wrong."""
 330
 331         start = max(offset - before - 1, 0)
 332         end = offset + after
 333         expected_results = expected_order[start: end]
 334
 335         # if it is a tuple with the cn, drop the cn
 336         if expected_results and isinstance(expected_results[0], tuple):
 337             expected_results = [x[0] for x in expected_results]
 338
 339         if expected_results == results:
 340             return
 341
 342         if expected_order is not None:
 343             print "expected order: %s" % expected_order
 344
 345         print "offset %d before %d after %d" % (offset, before, after)
 346         print "start %d end %d" % (start, end)
 347         print "expected: %s" % expected_results
 348         print "got     : %s" % results
 349         self.assertEquals(expected_results, results)
 350
 351     def test_server_vlv_with_cookie(self):
 352         attrs = [x for x in self.users[0].keys() if x not in
 353                  ('dn', 'objectclass')]
 354         for attr in attrs:
 355             expected_order = self.get_expected_order(attr)
 356             sort_control = "server_sort:1:0:%s" % attr
 357             res = None
 358             n = len(self.users)
 359             for before in range(0, 9):
 360                 for after in range(0, 9):
 361                     for offset in range(max(1, before - 2),
 362                                         min(n - after + 2, n)):
 363                         if res is None:
 364                             vlv_search = "vlv:1:%d:%d:%d:0" % (before, after,
 365                                                                offset)
 366                         else:
 367                             cookie = get_cookie(res.controls, n)
 368                             vlv_search = ("vlv:1:%d:%d:%d:%s:%s" %
 369                                           (before, after, offset, n,
 370                                            cookie))
 371
 372                         res = self.ldb.search(self.ou,
 373                                               scope=ldb.SCOPE_ONELEVEL,
 374                                               attrs=[attr],
 375                                               controls=[sort_control,
 376                                                         vlv_search])
 377
 378                         results = [x[attr][0] for x in res]
 379
 380                         self.assertCorrectResults(results, expected_order,
 381                                                   offset, before, after)
 382
 383     def run_index_tests_with_expressions(self, expressions):
 384         # Here we don't test every before/after combination.
 385         attrs = [x for x in self.users[0].keys() if x not in
 386                  ('dn', 'objectclass')]
 387         for attr in attrs:
 388             for expression in expressions:
 389                 expected_order = self.get_expected_order(attr, expression)
 390                 sort_control = "server_sort:1:0:%s" % attr
 391                 res = None
 392                 n = len(expected_order)
 393                 for before in range(0, 11):
 394                     after = before
 395                     for offset in range(max(1, before - 2),
 396                                         min(n - after + 2, n)):
 397                         if res is None:
 398                             vlv_search = "vlv:1:%d:%d:%d:0" % (before, after,
 399                                                                offset)
 400                         else:
 401                             cookie = get_cookie(res.controls)
 402                             vlv_search = ("vlv:1:%d:%d:%d:%s:%s" %
 403                                           (before, after, offset, n,
 404                                            cookie))
 405
 406                         res = self.ldb.search(self.ou,
 407                                               expression=expression,
 408                                               scope=ldb.SCOPE_ONELEVEL,
 409                                               attrs=[attr],
 410                                               controls=[sort_control,
 411                                                         vlv_search])
 412
 413                         results = [x[attr][0] for x in res]
 414
 415                         self.assertCorrectResults(results, expected_order,
 416                                                   offset, before, after)
 417
 418     def test_server_vlv_with_expression(self):
 419         """What happens when we run the VLV with an expression?"""
 420         expressions = ["(objectClass=*)",
 421                        "(cn=%s)" % self.users[-1]['cn'],
 422                        "(roomNumber=%s)" % self.users[0]['roomNumber'],
 423                        ]
 424         self.run_index_tests_with_expressions(expressions)
 425
 426     def test_server_vlv_with_failing_expression(self):
 427         """What happens when we run the VLV on an expression that matches
 428         nothing?"""
 429         expressions = ["(samaccountname=testferf)",
 430                        "(cn=hefalump)",
 431                        ]
 432         self.run_index_tests_with_expressions(expressions)
 433
 434     def run_gte_tests_with_expressions(self, expressions):
 435         # Here we don't test every before/after combination.
 436         attrs = [x for x in self.users[0].keys() if x not in
 437                  ('dn', 'objectclass')]
 438         for expression in expressions:
 439             for attr in attrs:
 440                 gte_order, expected_order, gte_map = \
 441                     self.get_gte_tests_and_order(attr, expression)
 442                 # In case there is some order dependency, disorder tests
 443                 gte_tests = gte_order[:]
 444                 random.seed(2)
 445                 random.shuffle(gte_tests)
 446                 res = None
 447                 sort_control = "server_sort:1:0:%s" % attr
 448
 449                 expected_order = self.get_expected_order(attr, expression)
 450                 sort_control = "server_sort:1:0:%s" % attr
 451                 res = None
 452                 for before in range(0, 11):
 453                     after = before
 454                     for gte in gte_tests:
 455                         if res is not None:
 456                             cookie = get_cookie(res.controls)
 457                         else:
 458                             cookie = None
 459                         vlv_search = encode_vlv_control(before=before,
 460                                                         after=after,
 461                                                         gte=gte,
 462                                                         cookie=cookie)
 463
 464                         res = self.ldb.search(self.ou,
 465                                               scope=ldb.SCOPE_ONELEVEL,
 466                                               expression=expression,
 467                                               attrs=[attr],
 468                                               controls=[sort_control,
 469                                                         vlv_search])
 470
 471                         results = [x[attr][0] for x in res]
 472                         offset = gte_map.get(gte, len(expected_order))
 473
 474                         # here offset is 0-based
 475                         start = max(offset - before, 0)
 476                         end = offset + 1 + after
 477
 478                         expected_results = expected_order[start: end]
 479
 480                         self.assertEquals(expected_results, results)
 481
 482     def test_vlv_gte_with_expression(self):
 483         """What happens when we run the VLV with an expression?"""
 484         expressions = ["(objectClass=*)",
 485                        "(cn=%s)" % self.users[-1]['cn'],
 486                        "(roomNumber=%s)" % self.users[0]['roomNumber'],
 487                        ]
 488         self.run_gte_tests_with_expressions(expressions)
 489
 490     def test_vlv_gte_with_failing_expression(self):
 491         """What happens when we run the VLV on an expression that matches
 492         nothing?"""
 493         expressions = ["(samaccountname=testferf)",
 494                        "(cn=hefalump)",
 495                        ]
 496         self.run_gte_tests_with_expressions(expressions)
 497
 498     def test_server_vlv_with_cookie_while_adding_and_deleting(self):
 499         """What happens if we add or remove items in the middle of the VLV?
 500
 501         Nothing. The search and the sort is not repeated, and we only
 502         deal with ther objects originally found.
 503         """
 504         attrs = ['cn'] + [x for x in self.users[0].keys() if x not in
 505                  ('dn', 'objectclass')]
 506         user_number = 0
 507         iteration = 0
 508         for attr in attrs:
 509             full_results, controls, sort_control = \
 510                             self.get_full_list(attr, True)
 511             original_n = len(self.users)
 512
 513             expected_order = full_results
 514             random.seed(1)
 515
 516             for before in range(0, 3) + [6, 11, 19]:
 517                 for after in range(0, 3) + [6, 11, 19]:
 518                     start = max(before - 1, 1)
 519                     end = max(start + 4, original_n - after + 2)
 520                     for offset in range(start, end):
 521                         #if iteration > 2076:
 522                         #    return
 523                         cookie = get_cookie(controls, original_n)
 524                         vlv_search = encode_vlv_control(before=before,
 525                                                         after=after,
 526                                                         offset=offset,
 527                                                         n=original_n,
 528                                                         cookie=cookie)
 529
 530                         iteration += 1
 531                         res = self.ldb.search(self.ou,
 532                                               scope=ldb.SCOPE_ONELEVEL,
 533                                               attrs=[attr],
 534                                               controls=[sort_control,
 535                                                         vlv_search])
 536
 537                         controls = res.controls
 538                         results = [x[attr][0] for x in res]
 539                         real_offset = max(1, min(offset, len(expected_order)))
 540
 541                         expected_results = []
 542                         skipped = 0
 543                         begin_offset = max(real_offset - before - 1, 0)
 544                         real_before = min(before, real_offset - 1)
 545                         real_after = min(after,
 546                                          len(expected_order) - real_offset)
 547
 548                         for x in expected_order[begin_offset:]:
 549                             if x is not None:
 550                                 expected_results.append(x[0])
 551                                 if (len(expected_results) ==
 552                                     real_before + real_after + 1):
 553                                     break
 554                             else:
 555                                 skipped += 1
 556
 557                         if expected_results != results:
 558                             print ("attr %s before %d after %d offset %d" %
 559                                    (attr, before, after, offset))
 560                         self.assertEquals(expected_results, results)
 561
 562                         n = len(self.users)
 563                         if random.random() < 0.1 + (n < 5) * 0.05:
 564                             if n == 0:
 565                                 i = 0
 566                             else:
 567                                 i = random.randrange(n)
 568                             user = self.create_user(i, n, suffix='-%s' %
 569                                                     user_number)
 570                             user_number += 1
 571                         if random.random() < 0.1  + (n > 50) * 0.02 and n:
 572                             index = random.randrange(n)
 573                             user = self.users.pop(index)
 574
 575                             self.ldb.delete(user['dn'])
 576
 577                             replaced = (user[attr], user['cn'])
 578                             if replaced in expected_order:
 579                                 i = expected_order.index(replaced)
 580                                 expected_order[i] = None
 581
 582     def test_server_vlv_with_cookie_while_changing(self):
 583         """What happens if we modify items in the middle of the VLV?
 584
 585         The expected behaviour (as found on Windows) is the sort is
 586         not repeated, but the changes in attributes are reflected.
 587         """
 588         attrs = [x for x in self.users[0].keys() if x not in
 589                  ('dn', 'objectclass', 'cn')]
 590         for attr in attrs:
 591             n_users = len(self.users)
 592             expected_order = [x.upper() for x in self.get_expected_order(attr)]
 593             sort_control = "server_sort:1:0:%s" % attr
 594             res = None
 595             i = 0
 596
 597             # First we'll fetch the whole list so we know the original
 598             # sort order. This is necessary because we don't know how
 599             # the server will order equivalent items. We are using the
 600             # dn as a key.
 601             half_n = n_users // 2
 602             vlv_search = "vlv:1:%d:%d:%d:0" % (half_n, half_n, half_n + 1)
 603             res = self.ldb.search(self.ou,
 604                                   scope=ldb.SCOPE_ONELEVEL,
 605                                   attrs=['dn', attr],
 606                                   controls=[sort_control, vlv_search])
 607
 608             results = [x[attr][0].upper() for x in res]
 609             #self.assertEquals(expected_order, results)
 610
 611             dn_order = [str(x['dn']) for x in res]
 612             values = results[:]
 613
 614             for before in range(0, 3):
 615                 for after in range(0, 3):
 616                     for offset in range(1 + before, n_users - after):
 617                         cookie = get_cookie(res.controls, len(self.users))
 618                         vlv_search = ("vlv:1:%d:%d:%d:%s:%s" %
 619                                       (before, after, offset, len(self.users),
 620                                        cookie))
 621
 622                         res = self.ldb.search(self.ou,
 623                                               scope=ldb.SCOPE_ONELEVEL,
 624                                               attrs=['dn', attr],
 625                                               controls=[sort_control,
 626                                                         vlv_search])
 627
 628                         dn_results = [str(x['dn']) for x in res]
 629                         dn_expected = dn_order[offset - before - 1:
 630                                                offset + after]
 631
 632                         self.assertEquals(dn_expected, dn_results)
 633
 634                         results = [x[attr][0].upper() for x in res]
 635
 636                         self.assertCorrectResults(results, values,
 637                                                   offset, before, after)
 638
 639                         i += 1
 640                         if i % 3 == 2:
 641                             if (attr in self.locale_sorted_keys or
 642                                 attr in self.binary_sorted_keys):
 643                                 i1 = i % n_users
 644                                 i2 = (i ^ 255) % n_users
 645                                 dn1 = dn_order[i1]
 646                                 dn2 = dn_order[i2]
 647                                 v2 = values[i2]
 648
 649                                 if v2 in self.locale_sorted_keys:
 650                                     v2 += '-%d' % i
 651                                 cn1 = dn1.split(',', 1)[0][3:]
 652                                 cn2 = dn2.split(',', 1)[0][3:]
 653
 654                                 values[i1] = v2
 655
 656                                 m = ldb.Message()
 657                                 m.dn = ldb.Dn(self.ldb, dn1)
 658                                 m[attr] = ldb.MessageElement(v2,
 659                                                              ldb.FLAG_MOD_REPLACE,
 660                                                              attr)
 661
 662                                 self.ldb.modify(m)
 663
 664     def test_server_vlv_fractions_with_cookie(self):
 665         """What happens when the count is set to an arbitrary number?
 666
 667         In that case the offset and the count form a fraction, and the
 668         VLV should be centred at a point offset/count of the way
 669         through. For example, if offset is 3 and count is 6, the VLV
 670         should be looking around halfway. The actual algorithm is a
 671         bit fiddlier than that, because of the one-basedness of VLV.
 672         """
 673         attrs = [x for x in self.users[0].keys() if x not in
 674                  ('dn', 'objectclass')]
 675
 676         n_users = len(self.users)
 677
 678         random.seed(4)
 679
 680         for attr in attrs:
 681             full_results, controls, sort_control = self.get_full_list(attr)
 682             self.assertEqual(len(full_results), n_users)
 683             for before in range(0, 2):
 684                 for after in range(0, 2):
 685                     for denominator in range(1, 20):
 686                         for offset in range(1, denominator + 3):
 687                             cookie = get_cookie(controls, len(self.users))
 688                             vlv_search = ("vlv:1:%d:%d:%d:%s:%s" %
 689                                           (before, after, offset,
 690                                            denominator,
 691                                            cookie))
 692                             try:
 693                                 res = self.ldb.search(self.ou,
 694                                                       scope=ldb.SCOPE_ONELEVEL,
 695                                                       attrs=[attr],
 696                                                       controls=[sort_control,
 697                                                                 vlv_search])
 698                             except ldb.LdbError, e:
 699                                 if offset != 0:
 700                                     raise
 701                                 print ("offset %d denominator %d raised error "
 702                                        "expected error %s\n"
 703                                        "(offset zero is illegal unless "
 704                                        "content count is zero)" %
 705                                        (offset, denominator, e))
 706                                 continue
 707
 708                             results = [x[attr][0].lower() for x in res]
 709
 710                             if denominator == 0:
 711                                 denominator = n_users
 712                                 if offset == 0:
 713                                     offset = denominator
 714                             elif denominator == 1:
 715                                 # the offset can only be 1, but the 1/1 case
 716                                 # means something special
 717                                 if offset == 1:
 718                                     real_offset = n_users
 719                                 else:
 720                                     real_offset = 1
 721                             else:
 722                                 if offset > denominator:
 723                                     offset = denominator
 724                                 real_offset = (1 +
 725                                                int(round((n_users - 1) *
 726                                                          (offset - 1) /
 727                                                          (denominator - 1.0)))
 728                                 )
 729
 730                             self.assertCorrectResults(results, full_results,
 731                                                       real_offset, before,
 732                                                       after)
 733
 734                             controls = res.controls
 735                             if False:
 736                                 for c in list(controls):
 737                                     cstr = str(c)
 738                                     if cstr.startswith('vlv_resp'):
 739                                         bits = cstr.rsplit(':')
 740                                         print ("the answer is %s; we said %d" %
 741                                                (bits[2], real_offset))
 742                                         break
 743
 744     def test_server_vlv_no_cookie(self):
 745         attrs = [x for x in self.users[0].keys() if x not in
 746                  ('dn', 'objectclass')]
 747
 748         for attr in attrs:
 749             expected_order = self.get_expected_order(attr)
 750             sort_control = "server_sort:1:0:%s" % attr
 751             for before in range(0, 5):
 752                 for after in range(0, 7):
 753                     for offset in range(1 + before, len(self.users) - after):
 754                         res = self.ldb.search(self.ou,
 755                                               scope=ldb.SCOPE_ONELEVEL,
 756                                               attrs=[attr],
 757                                               controls=[sort_control,
 758                                                         "vlv:1:%d:%d:%d:0" %
 759                                                         (before, after,
 760                                                          offset)])
 761                         results = [x[attr][0] for x in res]
 762                         self.assertCorrectResults(results, expected_order,
 763                                                   offset, before, after)
 764
 765     def test_server_vlv_gte_with_cookie(self):
 766         attrs = [x for x in self.users[0].keys() if x not in
 767                  ('dn', 'objectclass')]
 768         for attr in attrs:
 769             gte_order, expected_order, gte_map = \
 770                                         self.get_gte_tests_and_order(attr)
 771             # In case there is some order dependency, disorder tests
 772             gte_tests = gte_order[:]
 773             random.seed(1)
 774             random.shuffle(gte_tests)
 775             res = None
 776             sort_control = "server_sort:1:0:%s" % attr
 777             for before in range(0, 5):
 778                 for after in range(0, 7):
 779                     for gte in gte_tests:
 780                         if res is not None:
 781                             cookie = get_cookie(res.controls, len(self.users))
 782                         else:
 783                             cookie = None
 784                         vlv_search = encode_vlv_control(before=before,
 785                                                         after=after,
 786                                                         gte=gte,
 787                                                         cookie=cookie)
 788
 789                         res = self.ldb.search(self.ou,
 790                                               scope=ldb.SCOPE_ONELEVEL,
 791                                               attrs=[attr],
 792                                               controls=[sort_control,
 793                                                         vlv_search])
 794
 795                         results = [x[attr][0] for x in res]
 796                         offset = gte_map.get(gte, len(expected_order))
 797
 798                         # here offset is 0-based
 799                         start = max(offset - before, 0)
 800                         end = offset + 1 + after
 801
 802                         expected_results = expected_order[start: end]
 803
 804                         self.assertEquals(expected_results, results)
 805
 806     def test_server_vlv_gte_no_cookie(self):
 807         attrs = [x for x in self.users[0].keys() if x not in
 808                  ('dn', 'objectclass')]
 809         iteration = 0
 810         for attr in attrs:
 811             gte_order, expected_order, gte_map = \
 812                                         self.get_gte_tests_and_order(attr)
 813             # In case there is some order dependency, disorder tests
 814             gte_tests = gte_order[:]
 815             random.seed(1)
 816             random.shuffle(gte_tests)
 817
 818             sort_control = "server_sort:1:0:%s" % attr
 819             for before in range(0, 4):
 820                 for after in range(0, 5):
 821                     for gte in gte_tests:
 822                         if attr == 'audio' and 0:
 823                             import pdb
 824                             pdb.set_trace()
 825                         vlv_search = encode_vlv_control(before=before,
 826                                                         after=after,
 827                                                         gte=gte)
 828
 829                         res = self.ldb.search(self.ou,
 830                                               scope=ldb.SCOPE_ONELEVEL,
 831                                               attrs=[attr],
 832                                               controls=[sort_control,
 833                                                         vlv_search])
 834                         results = [x[attr][0] for x in res]
 835
 836                         # here offset is 0-based
 837                         offset = gte_map.get(gte, len(expected_order))
 838                         start = max(offset - before, 0)
 839                         end = offset + after + 1
 840                         expected_results = expected_order[start: end]
 841                         iteration += 1
 842                         if expected_results != results:
 843                             middle = expected_order[len(expected_order) // 2]
 844                             print expected_results, results
 845                             print middle
 846                             print expected_order
 847                             print
 848                             print ("\nattr %s offset %d before %d "
 849                                    "after %d gte %s" %
 850                                    (attr, offset, before, after, gte))
 851                         self.assertEquals(expected_results, results)
 852
 853     def test_multiple_searches(self):
 854         """The maximum number of concurrent vlv searches per connection is
 855         currently set at 3. That means if you open 4 VLV searches the
 856         cookie on the first one should fail.
 857         """
 858         # Windows has a limit of 10 VLVs where there are low numbers
 859         # of objects in each search.
 860         attrs = ([x for x in self.users[0].keys() if x not in
 861                   ('dn', 'objectclass')] * 2)[:12]
 862
 863         vlv_cookies = []
 864         for attr in attrs:
 865             sort_control = "server_sort:1:0:%s" % attr
 866
 867             res = self.ldb.search(self.ou,
 868                                   scope=ldb.SCOPE_ONELEVEL,
 869                                   attrs=[attr],
 870                                   controls=[sort_control,
 871                                             "vlv:1:1:1:1:0"])
 872
 873             cookie = get_cookie(res.controls, len(self.users))
 874             vlv_cookies.append(cookie)
 875             time.sleep(0.2)
 876
 877         # now this one should fail
 878         self.assertRaises(ldb.LdbError,
 879                           self.ldb.search,
 880                           self.ou,
 881                           scope=ldb.SCOPE_ONELEVEL,
 882                           attrs=[attr],
 883                           controls=[sort_control,
 884                                     "vlv:1:1:1:1:0:%s" % vlv_cookies[0]])
 885
 886         # and this one should succeed
 887         res = self.ldb.search(self.ou,
 888                               scope=ldb.SCOPE_ONELEVEL,
 889                               attrs=[attr],
 890                               controls=[sort_control,
 891                                         "vlv:1:1:1:1:0:%s" % vlv_cookies[-1]])
 892
 893         # this one should fail because it is a new connection and
 894         # doesn't share cookies
 895         new_ldb = SamDB(host, credentials=creds,
 896                         session_info=system_session(lp), lp=lp)
 897
 898         self.assertRaises(ldb.LdbError,
 899                           new_ldb.search, self.ou,
 900                           scope=ldb.SCOPE_ONELEVEL,
 901                           attrs=[attr],
 902                           controls=[sort_control,
 903                                     "vlv:1:1:1:1:0:%s" % vlv_cookies[-1]])
 904
 905         # but now without the critical flag it just does no VLV.
 906         new_ldb.search(self.ou,
 907                        scope=ldb.SCOPE_ONELEVEL,
 908                        attrs=[attr],
 909                        controls=[sort_control,
 910                                  "vlv:0:1:1:1:0:%s" % vlv_cookies[-1]])
 911
 912
 913 if "://" not in host:
 914     if os.path.isfile(host):
 915         host = "tdb://%s" % host
 916     else:
 917         host = "ldap://%s" % host
 918
 919
 920 TestProgram(module=__name__, opts=subunitopts)