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
56 # current Class or Attribute object id
60 super(DrsReplSchemaTestCase, self).setUp()
63 url_dc = samba.tests.env_get_var_value("DC1")
64 (self.ldb_dc1, self.info_dc1) = samba.tests.connect_samdb_ex(url_dc,
66 url_dc = samba.tests.env_get_var_value("DC2")
67 (self.ldb_dc2, self.info_dc2) = samba.tests.connect_samdb_ex(url_dc,
70 # initialize objects prefix if not done yet
71 if self.obj_prefix is None:
72 t = time.strftime("%s", time.gmtime())
73 DrsReplSchemaTestCase.obj_prefix = "DrsReplSchema-%s" % t
75 # cache some of RootDSE props
76 self.schema_dn = self.info_dc1["schemaNamingContext"][0]
77 self.domain_dn = self.info_dc1["defaultNamingContext"][0]
78 self.config_dn = self.info_dc1["configurationNamingContext"][0]
79 self.forest_level = int(self.info_dc1["forestFunctionality"][0])
81 # we will need DCs DNS names for 'samba-tool drs' command
82 self.dnsname_dc1 = self.info_dc1["dnsHostName"][0]
83 self.dnsname_dc2 = self.info_dc2["dnsHostName"][0]
86 super(DrsReplSchemaTestCase, self).tearDown()
88 def _net_drs_replicate(self, DC, fromDC, nc_dn):
89 """Triggers replication cycle on 'DC' to
90 replicate from 'fromDC'. Naming context to
91 be replicated is 'nc_dn' dn"""
92 # find out where is net command
93 samba_tool_cmd = os.path.abspath("./bin/samba-tool")
94 # make command line credentials string
95 creds = samba.tests.cmdline_credentials
96 cmd_line_auth = "-U%s/%s%%%s" % (creds.get_domain(),
97 creds.get_username(), creds.get_password())
98 # bin/samba-tool drs replicate <Dest_DC_NAME> <Src_DC_NAME> <Naming Context>
99 cmd_line = "%s drs replicate %s %s %s %s" % (samba_tool_cmd, DC, fromDC,
100 nc_dn, cmd_line_auth)
101 ret = os.system(cmd_line)
102 self.assertEquals(ret, 0, "Replicating %s from %s has failed!" % (DC, fromDC))
104 def _GUID_string(self, guid):
105 return self.ldb_dc1.schema_format_value("objectGUID", guid)
107 def _ldap_schemaUpdateNow(self, sam_db):
109 "schemaUpdateNow": "1"}
110 m = Message.from_dict(sam_db, rec, FLAG_MOD_REPLACE)
113 def _make_obj_names(self, base_name):
114 '''Try to create a unique name for an object
115 that is to be added to schema'''
117 obj_name = "%s-%d-%s" % (self.obj_prefix, self.obj_id, base_name)
118 obj_ldn = obj_name.replace("-", "")
119 obj_dn = "CN=%s,%s" % (obj_name, self.schema_dn)
120 return (obj_dn, obj_name, obj_ldn)
122 def _schema_new_class(self, ldb_ctx, base_name, attrs=None):
123 (class_dn, class_name, class_ldn) = self._make_obj_names(base_name)
124 rec = {"dn": class_dn,
125 "objectClass": ["top", "classSchema"],
127 "lDAPDisplayName": class_ldn,
128 "governsId": "1.2.840." + str(random.randint(1,100000)) + ".1.5.13",
130 "objectClassCategory": "1",
132 "systemOnly": "FALSE"}
133 # allow overriding/adding attributes
134 if not attrs is None:
136 # add it to the Schema
138 self._ldap_schemaUpdateNow(ldb_ctx)
139 return (rec["lDAPDisplayName"], rec["dn"])
141 def _schema_new_attr(self, ldb_ctx, base_name, attrs=None):
142 (attr_dn, attr_name, attr_ldn) = self._make_obj_names(base_name)
143 rec = {"dn": attr_dn,
144 "objectClass": ["top", "attributeSchema"],
146 "lDAPDisplayName": attr_ldn,
147 "attributeId": "1.2.841." + str(random.randint(1,100000)) + ".1.5.13",
148 "attributeSyntax": "2.5.5.12",
151 "isSingleValued": "TRUE",
152 "systemOnly": "FALSE"}
153 # allow overriding/adding attributes
154 if not attrs is None:
156 # add it to the Schema
158 self._ldap_schemaUpdateNow(ldb_ctx)
159 return (rec["lDAPDisplayName"], rec["dn"])
161 def _check_object(self, obj_dn):
162 '''Check if object obj_dn exists on both DCs'''
163 res_dc1 = self.ldb_dc1.search(base=obj_dn,
166 self.assertEquals(len(res_dc1), 1,
167 "%s doesn't exists on %s" % (obj_dn, self.dnsname_dc1))
169 res_dc2 = self.ldb_dc2.search(base=obj_dn,
172 except LdbError, (enum, estr):
173 if enum == ERR_NO_SUCH_OBJECT:
174 self.fail("%s doesn't exists on %s" % (obj_dn, self.dnsname_dc2))
176 self.assertEquals(len(res_dc2), 1,
177 "%s doesn't exists on %s" % (obj_dn, self.dnsname_dc2))
179 def test_class(self):
180 """Simple test for classSchema replication"""
181 # add new classSchema object
182 (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-S")
183 # force replication from DC1 to DC2
184 self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn)
185 # check object is replicated
186 self._check_object(c_dn)
188 def test_classInheritance(self):
189 """Test inheritance through subClassOf
190 I think 5 levels of inheritance is pretty decent for now."""
191 # add 5 levels deep hierarchy
194 for i in range(1, 6):
195 base_name = "cls-I-%02d" % i
196 (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, base_name)
197 c_dn_list.append(c_dn)
199 # inherit from last class added
200 m = Message.from_dict(self.ldb_dc1,
202 "subClassOf": c_ldn_last},
204 self.ldb_dc1.modify(m)
205 # store last class ldapDisplayName
207 # force replication from DC1 to DC2
208 self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn)
209 # check objects are replicated
210 for c_dn in c_dn_list:
211 self._check_object(c_dn)
213 def test_classWithCustomAttribute(self):
214 """Create new Attribute and a Class,
215 that has value for newly created attribute.
216 This should check code path that searches for
217 AttributeID_id in Schema cache"""
218 # add new attributeSchema object
219 (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-A")
220 # add a base classSchema class so we can use our new
221 # attribute in class definition in a sibling class
222 (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-A",
223 {"systemMayContain": a_ldn})
224 # add new classSchema object with value for a_ldb attribute
225 (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-B",
226 {"objectClass": ["top", "classSchema", c_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)