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"
45 class DrsReplSchemaTestCase(drs_base
.DrsBaseTestCase
):
47 # prefix for all objects created
49 # current Class or Attribute object id
53 super(DrsReplSchemaTestCase
, self
).setUp()
55 # initialize objects prefix if not done yet
56 if self
.obj_prefix
is None:
57 t
= time
.strftime("%s", time
.gmtime())
58 DrsReplSchemaTestCase
.obj_prefix
= "DrsReplSchema-%s" % t
61 super(DrsReplSchemaTestCase
, self
).tearDown()
63 def _make_obj_names(self
, base_name
):
64 '''Try to create a unique name for an object
65 that is to be added to schema'''
67 obj_name
= "%s-%d-%s" % (self
.obj_prefix
, self
.obj_id
, base_name
)
68 obj_ldn
= obj_name
.replace("-", "")
69 obj_dn
= "CN=%s,%s" % (obj_name
, self
.schema_dn
)
70 return (obj_dn
, obj_name
, obj_ldn
)
72 def _schema_new_class(self
, ldb_ctx
, base_name
, attrs
=None):
73 (class_dn
, class_name
, class_ldn
) = self
._make
_obj
_names
(base_name
)
74 rec
= {"dn": class_dn
,
75 "objectClass": ["top", "classSchema"],
77 "lDAPDisplayName": class_ldn
,
78 "governsId": "1.2.840." + str(random
.randint(1,100000)) + ".1.5.13",
80 "objectClassCategory": "1",
82 "systemOnly": "FALSE"}
83 # allow overriding/adding attributes
86 # add it to the Schema
89 except LdbError
, (enum
, estr
):
90 self
.fail("Adding record failed with %d/%s" % (enum
, estr
))
92 self
._ldap
_schemaUpdateNow
(ldb_ctx
)
93 return (rec
["lDAPDisplayName"], rec
["dn"])
95 def _schema_new_attr(self
, ldb_ctx
, base_name
, attrs
=None):
96 (attr_dn
, attr_name
, attr_ldn
) = self
._make
_obj
_names
(base_name
)
98 "objectClass": ["top", "attributeSchema"],
100 "lDAPDisplayName": attr_ldn
,
101 "attributeId": "1.2.841." + str(random
.randint(1,100000)) + ".1.5.13",
102 "attributeSyntax": "2.5.5.12",
105 "isSingleValued": "TRUE",
106 "systemOnly": "FALSE"}
107 # allow overriding/adding attributes
108 if not attrs
is None:
110 # add it to the Schema
112 self
._ldap
_schemaUpdateNow
(ldb_ctx
)
113 return (rec
["lDAPDisplayName"], rec
["dn"])
115 def _check_object(self
, obj_dn
):
116 '''Check if object obj_dn exists on both DCs'''
117 res_dc1
= self
.ldb_dc1
.search(base
=obj_dn
,
120 self
.assertEquals(len(res_dc1
), 1,
121 "%s doesn't exists on %s" % (obj_dn
, self
.dnsname_dc1
))
123 res_dc2
= self
.ldb_dc2
.search(base
=obj_dn
,
126 except LdbError
, (enum
, estr
):
127 if enum
== ERR_NO_SUCH_OBJECT
:
128 self
.fail("%s doesn't exists on %s" % (obj_dn
, self
.dnsname_dc2
))
130 self
.assertEquals(len(res_dc2
), 1,
131 "%s doesn't exists on %s" % (obj_dn
, self
.dnsname_dc2
))
133 def test_class(self
):
134 """Simple test for classSchema replication"""
135 # add new classSchema object
136 (c_ldn
, c_dn
) = self
._schema
_new
_class
(self
.ldb_dc1
, "cls-S")
137 # force replication from DC1 to DC2
138 self
._net
_drs
_replicate
(DC
=self
.dnsname_dc2
, fromDC
=self
.dnsname_dc1
, nc_dn
=self
.schema_dn
)
139 # check object is replicated
140 self
._check
_object
(c_dn
)
142 def test_classInheritance(self
):
143 """Test inheritance through subClassOf
144 I think 5 levels of inheritance is pretty decent for now."""
145 # add 5 levels deep hierarchy
148 for i
in range(1, 6):
149 base_name
= "cls-I-%02d" % i
150 (c_ldn
, c_dn
) = self
._schema
_new
_class
(self
.ldb_dc1
, base_name
)
151 c_dn_list
.append(c_dn
)
153 # inherit from last class added
154 m
= Message
.from_dict(self
.ldb_dc1
,
156 "subClassOf": c_ldn_last
},
158 self
.ldb_dc1
.modify(m
)
159 # store last class ldapDisplayName
161 # force replication from DC1 to DC2
162 self
._net
_drs
_replicate
(DC
=self
.dnsname_dc2
, fromDC
=self
.dnsname_dc1
, nc_dn
=self
.schema_dn
)
163 # check objects are replicated
164 for c_dn
in c_dn_list
:
165 self
._check
_object
(c_dn
)
167 def test_classWithCustomAttribute(self
):
168 """Create new Attribute and a Class,
169 that has value for newly created attribute.
170 This should check code path that searches for
171 AttributeID_id in Schema cache"""
172 # add new attributeSchema object
173 (a_ldn
, a_dn
) = self
._schema
_new
_attr
(self
.ldb_dc1
, "attr-A")
174 # add a base classSchema class so we can use our new
175 # attribute in class definition in a sibling class
176 (c_ldn
, c_dn
) = self
._schema
_new
_class
(self
.ldb_dc1
, "cls-A",
177 {"systemMayContain": a_ldn
})
178 # add new classSchema object with value for a_ldb attribute
179 (c_ldn
, c_dn
) = self
._schema
_new
_class
(self
.ldb_dc1
, "cls-B",
180 {"objectClass": ["top", "classSchema", c_ldn
],
181 a_ldn
: "test_classWithCustomAttribute"})
182 # force replication from DC1 to DC2
183 self
._net
_drs
_replicate
(DC
=self
.dnsname_dc2
, fromDC
=self
.dnsname_dc1
, nc_dn
=self
.schema_dn
)
184 # check objects are replicated
185 self
._check
_object
(c_dn
)
186 self
._check
_object
(a_dn
)
188 def test_attribute(self
):
189 """Simple test for attributeSchema replication"""
190 # add new attributeSchema object
191 (a_ldn
, a_dn
) = self
._schema
_new
_attr
(self
.ldb_dc1
, "attr-S")
192 # force replication from DC1 to DC2
193 self
._net
_drs
_replicate
(DC
=self
.dnsname_dc2
, fromDC
=self
.dnsname_dc1
, nc_dn
=self
.schema_dn
)
194 # check object is replicated
195 self
._check
_object
(a_dn
)
198 """Basic plan is to create bunch of classSchema
199 and attributeSchema objects, replicate Schema NC
200 and then check all objects are replicated correctly"""
202 # add new classSchema object
203 (c_ldn
, c_dn
) = self
._schema
_new
_class
(self
.ldb_dc1
, "cls-A")
204 # add new attributeSchema object
205 (a_ldn
, a_dn
) = self
._schema
_new
_attr
(self
.ldb_dc1
, "attr-A")
207 # add attribute to the class we have
208 m
= Message
.from_dict(self
.ldb_dc1
,
210 "mayContain": a_ldn
},
212 self
.ldb_dc1
.modify(m
)
214 # force replication from DC1 to DC2
215 self
._net
_drs
_replicate
(DC
=self
.dnsname_dc2
, fromDC
=self
.dnsname_dc1
, nc_dn
=self
.schema_dn
)
217 # check objects are replicated
218 self
._check
_object
(c_dn
)
219 self
._check
_object
(a_dn
)