2 # -*- coding: utf-8 -*-
4 # Unix SMB/CIFS implementation.
5 # Copyright (C) Kamen Mazdrashki <kamenim@samba.org> 2011
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/>.
26 sys.path.insert(0, "bin/python")
28 from samba import dsdb
37 class DrsBaseTestCase(samba.tests.BlackboxTestCase):
38 """Base class implementation for all DRS python tests.
39 It is intended to provide common initialization and
40 and functionality used by all DRS tests in drs/python
41 test package. For instance, DC1 and DC2 are always used
42 to pass URLs for DCs to test against"""
45 super(DrsBaseTestCase, self).setUp()
48 url_dc = samba.tests.env_get_var_value("DC1")
49 (self.ldb_dc1, self.info_dc1) = samba.tests.connect_samdb_ex(url_dc,
51 url_dc = samba.tests.env_get_var_value("DC2")
52 (self.ldb_dc2, self.info_dc2) = samba.tests.connect_samdb_ex(url_dc,
55 # cache some of RootDSE props
56 self.schema_dn = self.info_dc1["schemaNamingContext"][0]
57 self.domain_dn = self.info_dc1["defaultNamingContext"][0]
58 self.config_dn = self.info_dc1["configurationNamingContext"][0]
59 self.forest_level = int(self.info_dc1["forestFunctionality"][0])
61 # we will need DCs DNS names for 'samba-tool drs' command
62 self.dnsname_dc1 = self.info_dc1["dnsHostName"][0]
63 self.dnsname_dc2 = self.info_dc2["dnsHostName"][0]
66 super(DrsBaseTestCase, self).tearDown()
68 def _GUID_string(self, guid):
69 return self.ldb_dc1.schema_format_value("objectGUID", guid)
71 def _ldap_schemaUpdateNow(self, sam_db):
73 "schemaUpdateNow": "1"}
74 m = Message.from_dict(sam_db, rec, FLAG_MOD_REPLACE)
77 def _deleted_objects_dn(self, sam_ldb):
78 wkdn = "<WKGUID=18E2EA80684F11D2B9AA00C04F79F805,%s>" % self.domain_dn
79 res = sam_ldb.search(base=wkdn,
81 controls=["show_deleted:1"])
82 self.assertEquals(len(res), 1)
83 return str(res[0]["dn"])
85 def _lost_and_found_dn(self, sam_ldb, nc):
86 wkdn = "<WKGUID=%s,%s>" % (dsdb.DS_GUID_LOSTANDFOUND_CONTAINER, nc)
87 res = sam_ldb.search(base=wkdn,
89 self.assertEquals(len(res), 1)
90 return str(res[0]["dn"])
92 def _make_obj_name(self, prefix):
93 return prefix + time.strftime("%s", time.gmtime())
95 def _samba_tool_cmdline(self, drs_command):
96 # find out where is net command
97 samba_tool_cmd = os.path.abspath("./bin/samba-tool")
98 # make command line credentials string
99 creds = self.get_credentials()
100 cmdline_auth = "-U%s/%s%%%s" % (creds.get_domain(),
101 creds.get_username(), creds.get_password())
102 # bin/samba-tool drs <drs_command> <cmdline_auth>
103 return "%s drs %s %s" % (samba_tool_cmd, drs_command, cmdline_auth)
105 def _net_drs_replicate(self, DC, fromDC, nc_dn=None, forced=True, local=False, full_sync=False):
107 nc_dn = self.domain_dn
108 # make base command line
109 samba_tool_cmdline = self._samba_tool_cmdline("replicate")
111 samba_tool_cmdline += " --sync-forced"
113 samba_tool_cmdline += " --local"
115 samba_tool_cmdline += " --full-sync"
116 # bin/samba-tool drs replicate <Dest_DC_NAME> <Src_DC_NAME> <Naming Context>
117 cmd_line = "%s %s %s %s" % (samba_tool_cmdline, DC, fromDC, nc_dn)
118 return self.check_output(cmd_line)
120 def _enable_inbound_repl(self, DC):
121 # make base command line
122 samba_tool_cmd = self._samba_tool_cmdline("options")
123 # disable replication
124 self.check_run("%s %s --dsa-option=-DISABLE_INBOUND_REPL" %(samba_tool_cmd, DC))
126 def _disable_inbound_repl(self, DC):
127 # make base command line
128 samba_tool_cmd = self._samba_tool_cmdline("options")
129 # disable replication
130 self.check_run("%s %s --dsa-option=+DISABLE_INBOUND_REPL" %(samba_tool_cmd, DC))