2 # -*- coding: utf-8 -*-
4 # Unix SMB/CIFS implementation.
5 # Copyright (C) Kamen Mazdrashki <kamenim@samba.org> 2010
7 # This program is free software; you can redistribute it and/or modify
8 # it under the terms of the GNU General Public License as published by
9 # the Free Software Foundation; either version 3 of the License, or
10 # (at your option) any later version.
12 # This program is distributed in the hope that it will be useful,
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 # GNU General Public License for more details.
17 # You should have received a copy of the GNU General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
23 # export DC_SERVER=target_dc_or_local_samdb_url
24 # export SUBUNITRUN=$samba4srcdir/scripting/bin/subunitrun
25 # PYTHONPATH="$PYTHONPATH:$samba4srcdir/lib/ldb/tests/python" $SUBUNITRUN dsdb_schema_info -U"$DOMAIN/$DC_USERNAME"%"$DC_PASSWORD"
33 sys.path.append("bin/python")
35 from samba.auth import system_session
36 from ldb import SCOPE_BASE, LdbError
37 from samba.samdb import SamDB
40 import samba.dcerpc.drsuapi
41 from samba.dcerpc.drsblobs import schemaInfoBlob
42 from samba.ndr import ndr_unpack
43 from samba.dcerpc.misc import GUID
46 class SchemaInfoTestCase(samba.tests.TestCase):
49 super(SchemaInfoTestCase, self).setUp()
53 res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])
54 self.assertEquals(len(res), 1)
55 self.schema_dn = res[0]["schemaNamingContext"][0]
56 self.base_dn = res[0]["defaultNamingContext"][0]
57 self.forest_level = int(res[0]["forestFunctionality"][0])
59 # get DC invocation_id
60 self.invocation_id = GUID(ldb.get_invocation_id())
63 super(SchemaInfoTestCase, self).tearDown()
65 def _getSchemaInfo(self):
67 schema_info_data = ldb.searchone(attribute="schemaInfo",
68 basedn=self.schema_dn,
69 expression="(objectClass=*)",
71 self.assertEqual(len(schema_info_data), 21)
72 schema_info = ndr_unpack(schemaInfoBlob, schema_info_data)
73 self.assertEqual(schema_info.marker, 0xFF)
75 # create default schemaInfo if
76 # attribute value is not created yet
77 schema_info = schemaInfoBlob()
78 schema_info.revision = 0
79 schema_info.invocation_id = self.invocation_id
82 def _checkSchemaInfo(self, schi_before, schi_after):
83 self.assertEqual(schi_before.revision + 1, schi_after.revision)
84 self.assertEqual(schi_before.invocation_id, schi_after.invocation_id)
85 self.assertEqual(schi_after.invocation_id, self.invocation_id)
87 def _ldap_schemaUpdateNow(self):
94 self.ldb.modify_ldif(ldif)
96 def _make_obj_names(self, prefix):
97 obj_name = prefix + time.strftime("%s", time.gmtime())
98 obj_ldap_name = obj_name.replace("-", "")
99 obj_dn = "CN=%s,%s" % (obj_name, self.schema_dn)
100 return (obj_name, obj_ldap_name, obj_dn)
102 def _make_attr_ldif(self, attr_name, attr_dn):
104 dn: """ + attr_dn + """
106 objectClass: attributeSchema
107 adminDescription: """ + attr_name + """
108 adminDisplayName: """ + attr_name + """
109 cn: """ + attr_name + """
110 attributeId: 1.2.840.""" + str(random.randint(1,100000)) + """.1.5.9940
111 attributeSyntax: 2.5.5.12
119 def test_AddModifyAttribute(self):
120 # get initial schemaInfo
121 schi_before = self._getSchemaInfo()
123 # create names for an attribute to add
124 (attr_name, attr_ldap_name, attr_dn) = self._make_obj_names("schemaInfo-Attr-")
125 ldif = self._make_attr_ldif(attr_name, attr_dn)
127 # add the new attribute
128 self.ldb.add_ldif(ldif)
129 self._ldap_schemaUpdateNow()
130 # compare resulting schemaInfo
131 schi_after = self._getSchemaInfo()
132 self._checkSchemaInfo(schi_before, schi_after)
134 # rename the Attribute
135 attr_dn_new = attr_dn.replace(attr_name, attr_name + "-NEW")
137 self.ldb.rename(attr_dn, attr_dn_new)
138 except LdbError, (num, _):
139 self.fail("failed to change lDAPDisplayName for %s: %s" % (attr_name, _))
141 # compare resulting schemaInfo
142 schi_after = self._getSchemaInfo()
143 self._checkSchemaInfo(schi_before, schi_after)
147 def _make_class_ldif(self, class_name, class_dn):
149 dn: """ + class_dn + """
151 objectClass: classSchema
152 adminDescription: """ + class_name + """
153 adminDisplayName: """ + class_name + """
154 cn: """ + class_name + """
155 governsId: 1.2.840.""" + str(random.randint(1,100000)) + """.1.5.9939
157 objectClassCategory: 1
158 subClassOf: organizationalPerson
160 systemMustContain: cn
165 def test_AddModifyClass(self):
166 # get initial schemaInfo
167 schi_before = self._getSchemaInfo()
169 # create names for a Class to add
170 (class_name, class_ldap_name, class_dn) = self._make_obj_names("schemaInfo-Class-")
171 ldif = self._make_class_ldif(class_name, class_dn)
174 self.ldb.add_ldif(ldif)
175 self._ldap_schemaUpdateNow()
176 # compare resulting schemaInfo
177 schi_after = self._getSchemaInfo()
178 self._checkSchemaInfo(schi_before, schi_after)
181 class_dn_new = class_dn.replace(class_name, class_name + "-NEW")
183 self.ldb.rename(class_dn, class_dn_new)
184 except LdbError, (num, _):
185 self.fail("failed to change lDAPDisplayName for %s: %s" % (class_name, _))
187 # compare resulting schemaInfo
188 schi_after = self._getSchemaInfo()
189 self._checkSchemaInfo(schi_before, schi_after)
192 ########################################################################################
193 if not "DC_SERVER" in os.environ.keys():
194 raise AssertionError("Please supply TARGET_DC in environment")
195 ldb_url = os.environ["DC_SERVER"]
198 if not "://" in ldb_url:
199 if os.path.isfile(ldb_url):
200 ldb_url = "tdb://%s" % ldb_url
202 ldb_url = "ldap://%s" % ldb_url
203 # user 'paged_search' module when connecting remotely
204 ldb_options = ["modules:paged_searches"]
206 ldb = SamDB(url=ldb_url,
207 lp=samba.tests.env_loadparm(),
208 session_info=system_session(),
209 credentials=samba.tests.cmdline_credentials,