* callback error */
if ( ! ctx->request_terminated && ret != LDB_SUCCESS) {
/* Not indexed, so we need to do a full scan */
- if (ltdb->warn_unindexed) {
+ if (ltdb->warn_unindexed || ltdb->disable_full_db_scan) {
/* useful for debugging when slow performance
* is caused by unindexed searches */
char *expression = ldb_filter_from_tree(ctx, ctx->tree);
talloc_free(expression);
}
+
if (match_count != 0) {
/* the indexing code gave an error
* after having returned at least one
ltdb->kv_ops->unlock_read(module);
return LDB_ERR_OPERATIONS_ERROR;
}
+
+ if (ltdb->disable_full_db_scan) {
+ ldb_set_errstring(ldb,
+ "ldb FULL SEARCH disabled");
+ ltdb->kv_ops->unlock_read(module);
+ return LDB_ERR_INAPPROPRIATE_MATCHING;
+ }
+
ret = ltdb_search_full(ctx);
if (ret != LDB_SUCCESS) {
ldb_set_errstring(ldb, "Indexed and full searches both failed!\n");
super(SearchTests, self).setUp()
self.testdir = tempdir()
self.filename = os.path.join(self.testdir, "search_test.ldb")
+ options = ["modules:rdn_name"]
+ if hasattr(self, 'IDXCHECK'):
+ options.append("disable_full_db_scan_for_self_test:1")
self.l = ldb.Ldb(self.url(),
flags=self.flags(),
- options=["modules:rdn_name"])
+ options=options)
try:
self.l.add(self.index)
except AttributeError:
expression="(|(x=y)(y=b))")
self.assertEqual(len(res11), 20)
+ def test_one_unindexable(self):
+ """Testing a search"""
+
+ try:
+ res11 = self.l.search(base="DC=samba,DC=org",
+ scope=ldb.SCOPE_ONELEVEL,
+ expression="(y=b*)")
+ if hasattr(self, 'IDX') and \
+ not hasattr(self, 'IDXONE') and \
+ hasattr(self, 'IDXCHECK'):
+ self.fail("Should have failed as un-indexed search")
+
+ self.assertEqual(len(res11), 9)
+
+ except ldb.LdbError as err:
+ enum = err.args[0]
+ estr = err.args[1]
+ self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
+ self.assertIn(estr, "ldb FULL SEARCH disabled")
+
+ def test_one_unindexable_presence(self):
+ """Testing a search"""
+
+ try:
+ res11 = self.l.search(base="DC=samba,DC=org",
+ scope=ldb.SCOPE_ONELEVEL,
+ expression="(y=*)")
+ if hasattr(self, 'IDX') and \
+ not hasattr(self, 'IDXONE') and \
+ hasattr(self, 'IDXCHECK'):
+ self.fail("Should have failed as un-indexed search")
+
+ self.assertEqual(len(res11), 24)
+
+ except ldb.LdbError as err:
+ enum = err.args[0]
+ estr = err.args[1]
+ self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
+ self.assertIn(estr, "ldb FULL SEARCH disabled")
+
+
def test_subtree_and_or(self):
"""Testing a search"""
expression="(@IDXONE=DC=SAMBA,DC=ORG)")
self.assertEqual(len(res11), 0)
+ def test_subtree_unindexable(self):
+ """Testing a search"""
+
+ try:
+ res11 = self.l.search(base="DC=samba,DC=org",
+ scope=ldb.SCOPE_SUBTREE,
+ expression="(y=b*)")
+ if hasattr(self, 'IDX') and \
+ hasattr(self, 'IDXCHECK'):
+ self.fail("Should have failed as un-indexed search")
+
+ self.assertEqual(len(res11), 9)
+
+ except ldb.LdbError as err:
+ enum = err.args[0]
+ estr = err.args[1]
+ self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
+ self.assertIn(estr, "ldb FULL SEARCH disabled")
+
+ def test_subtree_unindexable_presence(self):
+ """Testing a search"""
+
+ try:
+ res11 = self.l.search(base="DC=samba,DC=org",
+ scope=ldb.SCOPE_SUBTREE,
+ expression="(y=*)")
+ if hasattr(self, 'IDX') and \
+ hasattr(self, 'IDXCHECK'):
+ self.fail("Should have failed as un-indexed search")
+
+ self.assertEqual(len(res11), 24)
+
+ except ldb.LdbError as err:
+ enum = err.args[0]
+ estr = err.args[1]
+ self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
+ self.assertIn(estr, "ldb FULL SEARCH disabled")
+
+
def test_dn_filter_one(self):
"""Testing that a dn= filter succeeds
(or fails with disallowDNFilter
"@IDXATTR": [b"x", b"y", b"ou"]})
self.IDX = True
+class IndexedCheckSearchTests(IndexedSearchTests):
+ """Test searches using the index, to ensure the index doesn't
+ break things (full scan disabled)"""
+ def setUp(self):
+ self.IDXCHECK = True
+ super(IndexedCheckSearchTests, self).setUp()
+
class IndexedSearchDnFilterTests(SearchTests):
"""Test searches using the index, to ensure the index doesn't
break things"""
"@IDXATTR": [b"x", b"y", b"ou"],
"@IDXONE": [b"1"]})
self.IDX = True
+ self.IDXONE = True
+
+class IndexedCheckedAndOneLevelSearchTests(IndexedAndOneLevelSearchTests):
+ """Test searches using the index including @IDXONE, to ensure
+ the index doesn't break things (full scan disabled)"""
+ def setUp(self):
+ self.IDXCHECK = True
+ super(IndexedCheckedAndOneLevelSearchTests, self).setUp()
class IndexedAndOneLevelDNFilterSearchTests(SearchTests):
"""Test searches using the index including @IDXONE, to ensure