lib/ldb-samba/tests/index.py

   1 #!/usr/bin/env python3
   2 #
   3 # Tests for comparison expressions on indexed keys
   4 #
   5 #   Copyright (C) Andrew Bartlett <abartlet@samba.org> 2019
   6 #
   7 # This program is free software; you can redistribute it and/or modify
   8 # it under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation; either version 3 of the License, or
  10 # (at your option) any later version.
  11 #
  12 # This program is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19 #
  20 """Tests for expressions containing comparisons on indexed attributes.
  21    Copied from ldb's index.py"""
  22
  23 import os
  24 from unittest import TestCase
  25 import sys
  26 from samba import _ldb
  27 import shutil
  28 from ldb import SCOPE_SUBTREE
  29 from samba.tests.subunitrun import TestProgram
  30
  31
  32 TDB_PREFIX = "tdb://"
  33 MDB_PREFIX = "mdb://"
  34
  35 def tempdir():
  36     import tempfile
  37     try:
  38         dir_prefix = os.path.join(os.environ["SELFTEST_PREFIX"], "tmp")
  39     except KeyError:
  40         dir_prefix = None
  41     return tempfile.mkdtemp(dir=dir_prefix)
  42
  43 class LdbBaseTest(TestCase):
  44     def setUp(self):
  45         super(LdbBaseTest, self).setUp()
  46         try:
  47             if self.prefix is None:
  48                 self.prefix = TDB_PREFIX
  49         except AttributeError:
  50             self.prefix = TDB_PREFIX
  51
  52     def tearDown(self):
  53         super(LdbBaseTest, self).tearDown()
  54
  55     def url(self):
  56         return self.prefix + self.filename
  57
  58     def flags(self):
  59         if self.prefix == MDB_PREFIX:
  60             return ldb.FLG_NOSYNC
  61         else:
  62             return 0
  63
  64     def options(self):
  65         if self.prefix == MDB_PREFIX:
  66             return ['disable_full_db_scan_for_self_test:1']
  67         else:
  68             return None
  69
  70 class LdbTDBIndexedComparisonExpressions(LdbBaseTest):
  71     def tearDown(self):
  72         shutil.rmtree(self.testdir)
  73         super(LdbTDBIndexedComparisonExpressions, self).tearDown()
  74
  75         # Ensure the LDB is closed now, so we close the FD
  76         del(self.l)
  77
  78     def setUp(self):
  79         super(LdbTDBIndexedComparisonExpressions, self).setUp()
  80         self.testdir = tempdir()
  81         self.filename = os.path.join(self.testdir, "indexedcomptest.ldb")
  82         # Note that the maximum key length is set to 54
  83         # This accounts for the 4 bytes added by the dn formatting
  84         # a leading dn=, and a trailing zero terminator
  85         #
  86         self.l = _ldb.Ldb(self.url(), options=self.options())
  87         self.l.add({"dn": "@ATTRIBUTES"})
  88         self.l.add({"dn": "@INDEXLIST",
  89                     "@IDXATTR": [b"int32attr"],
  90                     "@IDXONE": [b"1"],
  91                     "@IDXGUID": [b"objectUUID"],
  92                     "@IDX_DN_GUID": [b"GUID"]})
  93
  94     def test_comparison_expression(self):
  95         self.l.samba_schema_attribute_add("int32attr", 0,
  96                                           _ldb.SYNTAX_SAMBA_INT32)
  97
  98         int32_max = 2**31-1
  99         int32_min = -2**31
 100         test_nums = list(range(-5, 5))
 101         test_nums += list(range(int32_max-5, int32_max+1))
 102         test_nums += list(range(int32_min, int32_min+5))
 103         test_nums = sorted(test_nums)
 104
 105         for i in test_nums:
 106             ouuid = 0x0123456789abcdef + i
 107             ouuid_s = bytes(('0' + hex(ouuid)[2:]).encode())
 108             self.l.add({"dn": "OU=COMPTESTOU{},DC=SAMBA,DC=ORG".format(i),
 109                         "objectUUID": ouuid_s,
 110                         "int32attr": str(i)})
 111
 112         def assert_int32_expr(expr, py_expr=None):
 113             res = self.l.search(base="DC=SAMBA,DC=ORG",
 114                                 scope=SCOPE_SUBTREE,
 115                                 expression="(int32attr%s)" % (expr))
 116
 117             if not py_expr:
 118                 py_expr = expr
 119             expect = [n for n in test_nums if eval(str(n) + py_expr)]
 120             vals = sorted([int(r.get("int32attr")[0]) for r in res])
 121             self.assertEqual(len(res), len(expect))
 122             self.assertEqual(set(vals), set(expect))
 123             self.assertEqual(expect, vals)
 124
 125         assert_int32_expr(">=-2")
 126         assert_int32_expr("<=2")
 127         assert_int32_expr(">=" + str(int32_min))
 128         assert_int32_expr("<=" + str(int32_min))
 129         assert_int32_expr("<=" + str(int32_min+1))
 130         assert_int32_expr("<=" + str(int32_max))
 131         assert_int32_expr(">=" + str(int32_max))
 132         assert_int32_expr(">=" + str(int32_max-1))
 133         assert_int32_expr("=10", "==10")
 134
 135     def test_comparison_expression_duplicates(self):
 136         self.l.samba_schema_attribute_add("int32attr", 0,
 137                                           _ldb.SYNTAX_SAMBA_INT32)
 138
 139         int32_max = 2**31-1
 140         int32_min = -2**31
 141
 142         test_nums = list(range(-5, 5)) * 3
 143         test_nums += list(range(-20, 20, 5)) * 2
 144         test_nums += list(range(-50, 50, 15))
 145         test_nums = sorted(test_nums)
 146
 147         for i, n in enumerate(test_nums):
 148             ouuid = 0x0123456789abcdef + i
 149             ouuid_s = bytes(('0' + hex(ouuid)[2:]).encode())
 150             self.l.add({"dn": "OU=COMPTESTOU{},DC=SAMBA,DC=ORG".format(i),
 151                         "objectUUID": ouuid_s,
 152                         "int32attr": str(n)})
 153
 154         def assert_int32_expr(expr, py_expr=None):
 155             res = self.l.search(base="DC=SAMBA,DC=ORG",
 156                                 scope=SCOPE_SUBTREE,
 157                                 expression="(int32attr%s)" % (expr))
 158
 159             if not py_expr:
 160                 py_expr = expr
 161             expect = [n for n in test_nums if eval(str(n) + py_expr)]
 162             vals = sorted([int(r.get("int32attr")[0]) for r in res])
 163             self.assertEqual(len(res), len(expect))
 164             self.assertEqual(set(vals), set(expect))
 165             self.assertEqual(expect, vals)
 166
 167         assert_int32_expr(">=-2")
 168         assert_int32_expr("<=2")
 169         assert_int32_expr(">=" + str(int32_min))
 170         assert_int32_expr("<=" + str(int32_min))
 171         assert_int32_expr("<=" + str(int32_min+1))
 172         assert_int32_expr("<=" + str(int32_max))
 173         assert_int32_expr(">=" + str(int32_max))
 174         assert_int32_expr(">=" + str(int32_max-1))
 175         assert_int32_expr("=-5", "==-5")
 176         assert_int32_expr("=5", "==5")
 177
 178 # Run the same tests against an lmdb backend
 179 class LdbLMDBIndexedComparisonExpressions(LdbTDBIndexedComparisonExpressions):
 180
 181     def setUp(self):
 182         if os.environ.get('HAVE_LMDB', '1') == '0':
 183             self.skipTest("No lmdb backend")
 184         self.prefix = MDB_PREFIX
 185         super(LdbLMDBIndexedComparisonExpressions, self).setUp()
 186
 187     def tearDown(self):
 188         super(LdbLMDBIndexedComparisonExpressions, self).tearDown()
 189
 190
 191 TestProgram(module=__name__, opts=[])