2 # -*- coding: utf-8 -*-
4 # Unix SMB/CIFS implementation.
5 # This speed test aims to show difference in execution time for bulk
6 # creation of user objects. This will help us compare
7 # Samba4 vs MS Active Directory performance.
9 # Copyright (C) Zahari Zahariev <zahari.zahariev@postpath.com> 2010
11 # This program is free software; you can redistribute it and/or modify
12 # it under the terms of the GNU General Public License as published by
13 # the Free Software Foundation; either version 3 of the License, or
14 # (at your option) any later version.
16 # This program is distributed in the hope that it will be useful,
17 # but WITHOUT ANY WARRANTY; without even the implied warranty of
18 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19 # GNU General Public License for more details.
21 # You should have received a copy of the GNU General Public License
22 # along with this program. If not, see <http://www.gnu.org/licenses/>.
25 from __future__ import print_function
30 from decimal import Decimal
32 sys.path.insert(0, "bin/python")
34 from samba.tests.subunitrun import TestProgram, SubunitOptions
36 import samba.getopt as options
38 from ldb import SCOPE_BASE, SCOPE_SUBTREE
39 from samba.ndr import ndr_unpack
40 from samba.dcerpc import security
42 from samba.auth import system_session
43 from samba import gensec, sd_utils
44 from samba.samdb import SamDB
45 from samba.credentials import Credentials
47 from samba.tests import delete_force
49 parser = optparse.OptionParser("speedtest.py [options] <host>")
50 sambaopts = options.SambaOptions(parser)
51 parser.add_option_group(sambaopts)
52 parser.add_option_group(options.VersionOptions(parser))
54 # use command line creds if available
55 credopts = options.CredentialsOptions(parser)
56 parser.add_option_group(credopts)
57 subunitopts = SubunitOptions(parser)
58 parser.add_option_group(subunitopts)
59 opts, args = parser.parse_args()
67 lp = sambaopts.get_loadparm()
68 creds = credopts.get_credentials(lp)
69 creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)
76 class SpeedTest(samba.tests.TestCase):
78 def find_domain_sid(self, ldb):
79 res = ldb.search(base=self.base_dn, expression="(objectClass=*)", scope=SCOPE_BASE)
80 return ndr_unpack(security.dom_sid, res[0]["objectSid"][0])
83 super(SpeedTest, self).setUp()
85 self.base_dn = ldb.domain_dn()
86 self.domain_sid = security.dom_sid(ldb.get_domain_sid())
87 self.user_pass = "samba123@"
88 print("baseDN: %s" % self.base_dn)
90 def create_user(self, user_dn):
92 dn: """ + user_dn + """
93 sAMAccountName: """ + user_dn.split(",")[0][3:] + """
95 unicodePwd:: """ + base64.b64encode(("\"%s\"" % self.user_pass).encode('utf-16-le')).decode('utf8') + """
98 self.ldb_admin.add_ldif(ldif)
100 def create_group(self, group_dn, desc=None):
102 dn: """ + group_dn + """
104 sAMAccountName: """ + group_dn.split(",")[0][3:] + """
108 self.ldb_admin.add_ldif(ldif)
110 def create_bundle(self, count):
111 for i in range(count):
112 self.create_user("cn=speedtestuser%d,cn=Users,%s" % (i + 1, self.base_dn))
114 def remove_bundle(self, count):
115 for i in range(count):
116 delete_force(self.ldb_admin, "cn=speedtestuser%d,cn=Users,%s" % (i + 1, self.base_dn))
118 def remove_test_users(self):
119 res = ldb.search(base="cn=Users,%s" % self.base_dn, expression="(objectClass=user)", scope=SCOPE_SUBTREE)
120 dn_list = [item.dn for item in res if "speedtestuser" in str(item.dn)]
122 delete_force(self.ldb_admin, dn)
125 class SpeedTestAddDel(SpeedTest):
128 super(SpeedTestAddDel, self).setUp()
130 def run_bundle(self, num):
131 print("\n=== Test ADD/DEL %s user objects ===\n" % num)
132 avg_add = Decimal("0.0")
133 avg_del = Decimal("0.0")
136 self.create_bundle(num)
137 res_add = Decimal(str(time.time() - start))
139 print(" Attempt %s ADD: %.3fs" % (x, float(res_add)))
142 self.remove_bundle(num)
143 res_del = Decimal(str(time.time() - start))
145 print(" Attempt %s DEL: %.3fs" % (x, float(res_del)))
146 print("Average ADD: %.3fs" % float(Decimal(avg_add) / Decimal("3.0")))
147 print("Average DEL: %.3fs" % float(Decimal(avg_del) / Decimal("3.0")))
150 def test_00000(self):
151 """ Remove possibly undeleted test users from previous test
153 self.remove_test_users()
155 def test_00010(self):
158 def test_00100(self):
161 def test_01000(self):
162 self.run_bundle(1000)
164 def _test_10000(self):
165 """ This test should be enabled preferably against MS Active Directory.
166 It takes quite the time against Samba4 (1-2 days).
168 self.run_bundle(10000)
171 class AclSearchSpeedTest(SpeedTest):
174 super(AclSearchSpeedTest, self).setUp()
175 self.ldb_admin.newuser("acltestuser", "samba123@")
176 self.sd_utils = sd_utils.SDUtils(self.ldb_admin)
177 self.ldb_user = self.get_ldb_connection("acltestuser", "samba123@")
178 self.user_sid = self.sd_utils.get_object_sid(self.get_user_dn("acltestuser"))
181 super(AclSearchSpeedTest, self).tearDown()
182 delete_force(self.ldb_admin, self.get_user_dn("acltestuser"))
184 def run_search_bundle(self, num, _ldb):
185 print("\n=== Creating %s user objects ===\n" % num)
186 self.create_bundle(num)
187 mod = "(A;;LC;;;%s)(D;;RP;;;%s)" % (str(self.user_sid), str(self.user_sid))
189 self.sd_utils.dacl_add_ace("cn=speedtestuser%d,cn=Users,%s" %
190 (i + 1, self.base_dn), mod)
191 print("\n=== %s user objects created ===\n" % num)
192 print("\n=== Test search on %s user objects ===\n" % num)
193 avg_search = Decimal("0.0")
196 res = _ldb.search(base=self.base_dn, expression="(objectClass=*)", scope=SCOPE_SUBTREE)
197 res_search = Decimal(str(time.time() - start))
198 avg_search += res_search
199 print(" Attempt %s SEARCH: %.3fs" % (x, float(res_search)))
200 print("Average Search: %.3fs" % float(Decimal(avg_search) / Decimal("3.0")))
201 self.remove_bundle(num)
203 def get_user_dn(self, name):
204 return "CN=%s,CN=Users,%s" % (name, self.base_dn)
206 def get_ldb_connection(self, target_username, target_password):
207 creds_tmp = Credentials()
208 creds_tmp.set_username(target_username)
209 creds_tmp.set_password(target_password)
210 creds_tmp.set_domain(creds.get_domain())
211 creds_tmp.set_realm(creds.get_realm())
212 creds_tmp.set_workstation(creds.get_workstation())
213 creds_tmp.set_gensec_features(creds_tmp.get_gensec_features()
214 | gensec.FEATURE_SEAL)
215 ldb_target = SamDB(url=host, credentials=creds_tmp, lp=lp)
218 def test_search_01000(self):
219 self.run_search_bundle(1000, self.ldb_admin)
221 def test_search2_01000(self):
222 # allow the user to see objects but not attributes, all attributes will be filtered out
223 mod = "(A;;LC;;;%s)(D;;RP;;;%s)" % (str(self.user_sid), str(self.user_sid))
224 self.sd_utils.dacl_add_ace("CN=Users,%s" % self.base_dn, mod)
225 self.run_search_bundle(1000, self.ldb_user)
227 # Important unit running information
230 if "://" not in host:
231 host = "ldap://%s" % host
233 ldb_options = ["modules:paged_searches"]
234 ldb = SamDB(host, credentials=creds, session_info=system_session(), lp=lp, options=ldb_options)
236 TestProgram(module=__name__, opts=subunitopts)