2 # -*- coding: utf-8 -*-
4 # Tests various schema replication scenarios
6 # Copyright (C) Kamen Mazdrashki <kamenim@samba.org> 2010
8 # This program is free software; you can redistribute it and/or modify
9 # it under the terms of the GNU General Public License as published by
10 # the Free Software Foundation; either version 3 of the License, or
11 # (at your option) any later version.
13 # This program is distributed in the hope that it will be useful,
14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 # GNU General Public License for more details.
18 # You should have received a copy of the GNU General Public License
19 # along with this program. If not, see <http://www.gnu.org/licenses/>.
24 # export DC1=dc1_dns_name
25 # export DC2=dc2_dns_name
26 # export SUBUNITRUN=$samba4srcdir/scripting/bin/subunitrun
27 # PYTHONPATH="$PYTHONPATH:$samba4srcdir/torture/drs/python" $SUBUNITRUN repl_schema -U"$DOMAIN/$DC_USERNAME"%"$DC_PASSWORD"
35 sys.path.append("bin/python")
37 samba.ensure_external_module("testtools", "testtools")
38 samba.ensure_external_module("subunit", "subunit/python")
52 class DrsReplSchemaTestCase(samba.tests.TestCase):
54 # prefix for all objects created
58 super(DrsReplSchemaTestCase, self).setUp()
61 url_dc = samba.tests.env_get_var_value("DC1")
62 (self.ldb_dc1, self.info_dc1) = samba.tests.connect_samdb_ex(url_dc,
64 url_dc = samba.tests.env_get_var_value("DC2")
65 (self.ldb_dc2, self.info_dc2) = samba.tests.connect_samdb_ex(url_dc,
68 # initialize objects prefix if not done yet
69 if self.obj_prefix is None:
70 t = time.strftime("%s", time.gmtime())
71 DrsReplSchemaTestCase.obj_prefix = "DrsReplSchema-%s-" % t
73 # cache some of RootDSE props
74 self.schema_dn = self.info_dc1["schemaNamingContext"][0]
75 self.domain_dn = self.info_dc1["defaultNamingContext"][0]
76 self.config_dn = self.info_dc1["configurationNamingContext"][0]
77 self.forest_level = int(self.info_dc1["forestFunctionality"][0])
79 # we will need DCs DNS names for 'samba-tool drs' command
80 self.dnsname_dc1 = self.info_dc1["dnsHostName"][0]
81 self.dnsname_dc2 = self.info_dc2["dnsHostName"][0]
84 super(DrsReplSchemaTestCase, self).tearDown()
86 def _net_drs_replicate(self, DC, fromDC, nc_dn):
87 """Triggers replication cycle on 'DC' to
88 replicate from 'fromDC'. Naming context to
89 be replicated is 'nc_dn' dn"""
90 # find out where is net command
91 samba_tool_cmd = os.path.abspath("./bin/samba-tool")
92 # make command line credentials string
93 creds = samba.tests.cmdline_credentials
94 cmd_line_auth = "-U%s/%s%%%s" % (creds.get_domain(),
95 creds.get_username(), creds.get_password())
96 # bin/samba-tool drs replicate <Dest_DC_NAME> <Src_DC_NAME> <Naming Context>
97 cmd_line = "%s drs replicate %s %s %s %s" % (samba_tool_cmd, DC, fromDC,
99 ret = os.system(cmd_line)
100 self.assertEquals(ret, 0, "Replicating %s from %s has failed!" % (DC, fromDC))
102 def _GUID_string(self, guid):
103 return self.ldb_dc1.schema_format_value("objectGUID", guid)
105 def _ldap_schemaUpdateNow(self, sam_db):
107 "schemaUpdateNow": "1"}
108 m = Message.from_dict(sam_db, rec, FLAG_MOD_REPLACE)
111 def _make_obj_names(self, base_name):
112 '''Try to create a unique name for an object
113 that is to be added to schema'''
114 obj_name = self.obj_prefix + base_name
115 obj_ldn = obj_name.replace("-", "")
116 obj_dn = "CN=%s,%s" % (obj_name, self.schema_dn)
117 return (obj_dn, obj_name, obj_ldn)
119 def _schema_new_class(self, ldb_ctx, base_name, attrs=None):
120 (class_dn, class_name, class_ldn) = self._make_obj_names(base_name)
121 rec = {"dn": class_dn,
122 "objectClass": ["top", "classSchema"],
124 "lDAPDisplayName": class_ldn,
125 "governsId": "1.2.840." + str(random.randint(1,100000)) + ".1.5.13",
127 "objectClassCategory": "1",
128 "subClassOf": "organizationalPerson",
129 "systemOnly": "FALSE"}
130 # allow overriding/adding attributes
131 if not attrs is None:
133 # add it to the Schema
135 self._ldap_schemaUpdateNow(ldb_ctx)
136 return (rec["lDAPDisplayName"], rec["dn"])
138 def _schema_new_attr(self, ldb_ctx, base_name, attrs=None):
139 (attr_dn, attr_name, attr_ldn) = self._make_obj_names(base_name)
140 rec = {"dn": attr_dn,
141 "objectClass": ["top", "attributeSchema"],
143 "lDAPDisplayName": attr_ldn,
144 "attributeId": "1.2.841." + str(random.randint(1,100000)) + ".1.5.13",
145 "attributeSyntax": "2.5.5.12",
148 "isSingleValued": "TRUE",
149 "systemOnly": "FALSE"}
150 # allow overriding/adding attributes
151 if not attrs is None:
153 # add it to the Schema
155 self._ldap_schemaUpdateNow(ldb_ctx)
156 return (rec["lDAPDisplayName"], rec["dn"])
158 def _check_object(self, obj_dn):
159 '''Check if object obj_dn exists on both DCs'''
160 res_dc1 = self.ldb_dc1.search(base=obj_dn,
163 self.assertEquals(len(res_dc1), 1,
164 "%s doesn't exists on %s" % (obj_dn, self.dnsname_dc1))
166 res_dc2 = self.ldb_dc2.search(base=obj_dn,
169 except LdbError, (enum, estr):
170 if enum == ERR_NO_SUCH_OBJECT:
171 self.fail("%s doesn't exists on %s" % (obj_dn, self.dnsname_dc2))
173 self.assertEquals(len(res_dc2), 1,
174 "%s doesn't exists on %s" % (obj_dn, self.dnsname_dc2))
176 def test_class(self):
177 """Simple test for classSchema replication"""
178 # add new classSchema object
179 (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-S")
180 # force replication from DC1 to DC2
181 self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn)
182 # check object is replicated
183 self._check_object(c_dn)
185 def test_classInheritance(self):
186 """Test inheritance through subClassOf
187 I think 5 levels of inheritance is pretty decent for now."""
188 # add 5 levels deep hierarchy
191 for i in range(1, 6):
192 base_name = "cls-I-%02d" % i
193 (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, base_name)
194 c_dn_list.append(c_dn)
196 # inherit from last class added
197 m = Message.from_dict(self.ldb_dc1,
199 "subClassOf": c_ldn_last},
201 self.ldb_dc1.modify(m)
202 # store last class ldapDisplayName
204 # force replication from DC1 to DC2
205 self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn)
206 # check objects are replicated
207 for c_dn in c_dn_list:
208 self._check_object(c_dn)
210 def test_classWithCustomAttribute(self):
211 """Create new Attribute and a Class,
212 that has value for newly created attribute.
213 This should check code path that searches for
214 AttributeID_id in Schema cache"""
215 # add new attributeSchema object
216 (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-A")
217 # add a base classSchema class so we can use our new
218 # attribute in class definition in a sibling class
219 (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-A",
220 {"systemMayContain": a_ldn})
221 # add new classSchema object with value for a_ldb attribute
222 (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-B",
223 {"objectClass": ["top", "classSchema", c_ldn],
224 a_ldn: "test_classWithCustomAttribute"})
225 #(c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-B",
226 # {"systemMayContain": a_ldn,
227 # a_ldn: "test_classWithCustomAttribute"})
228 # force replication from DC1 to DC2
229 self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn)
230 # check objects are replicated
231 self._check_object(c_dn)
232 self._check_object(a_dn)
234 def test_attribute(self):
235 """Simple test for attributeSchema replication"""
236 # add new attributeSchema object
237 (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-S")
238 # force replication from DC1 to DC2
239 self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn)
240 # check object is replicated
241 self._check_object(a_dn)
244 """Basic plan is to create bunch of classSchema
245 and attributeSchema objects, replicate Schema NC
246 and then check all objects are replicated correctly"""
248 # add new classSchema object
249 (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-A")
250 # add new attributeSchema object
251 (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-A")
253 # add attribute to the class we have
254 m = Message.from_dict(self.ldb_dc1,
256 "mayContain": a_ldn},
258 self.ldb_dc1.modify(m)
260 # force replication from DC1 to DC2
261 self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn)
263 # check objects are replicated
264 self._check_object(c_dn)
265 self._check_object(a_dn)