tests: Add new tests for backup-rename command

Extend the existing 'backup online' tests to also test the domain
rename case. This mostly involves some extra assertions that the
restored DB has been modified appropriatelt (i.e. domain NetBIOS
name is updated, etc).

I've also added an extra test case that creates a few objects and
links and specifically asserts that they get renamed appropriately.

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
Reviewed-by: Gary Lockyer <gary@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>

diff --git a/python/samba/tests/domain_backup.py b/python/samba/tests/domain_backup.py
index 845ae3bf643f32557f067be9c8cf8a5823265b38..dabe56f1cc545485e8d595f365734557153d91ff 100644 (file)
--- a/python/samba/tests/domain_backup.py
+++ b/python/samba/tests/domain_backup.py
@@ -19,12 +19,13 @@ import tarfile
 import os
 import shutil
 from samba.tests.samba_tool.base import SambaToolCmdTest
-from samba.tests import TestCaseInTempDir, env_loadparm
+from samba.tests import TestCaseInTempDir, env_loadparm, create_test_ou
 import ldb
 from samba.samdb import SamDB
 from samba.auth import system_session
 from samba import Ldb, dn_from_dns_name
 from samba.netcmd.fsmo import get_fsmo_roleowner
+import re
 
 
 def get_prim_dom(secrets_path, lp):
@@ -332,3 +333,122 @@ class DomainBackupOnline(DomainBackupBase):
 
     def test_backup_restore_with_conf(self):
         self._test_backup_restore_with_conf()
+
+
+class DomainBackupRename(DomainBackupBase):
+
+    # run the above test cases using a rename backup
+    def setUp(self):
+        super(DomainBackupRename, self).setUp()
+        self.new_server = "RENAMESERV"
+        self.restore_domain = "NEWDOMAIN"
+        self.restore_realm = "rename.test.net"
+        self.new_basedn = "DC=rename,DC=test,DC=net"
+        self.base_cmd = ["domain", "backup", "rename", self.restore_domain,
+                         self.restore_realm]
+        self.backup_markers += ['backupRename']
+
+    # run the common test case code for backup-renames
+    def test_backup_untar(self):
+        self._test_backup_untar()
+
+    def test_backup_restore(self):
+        self._test_backup_restore()
+
+    def test_backup_restore_with_conf(self):
+        self._test_backup_restore_with_conf()
+
+    def add_link(self, attr, source, target):
+        m = ldb.Message()
+        m.dn = ldb.Dn(self.ldb, source)
+        m[attr] = ldb.MessageElement(target, ldb.FLAG_MOD_ADD, attr)
+        self.ldb.modify(m)
+
+    def test_one_way_links(self):
+        """Sanity-check that a rename handles one-way links correctly"""
+
+        # Do some initial setup on the DC before back it up:
+        # create an OU to hold the test objects we'll create
+        test_ou = create_test_ou(self.ldb, "rename_test")
+        self.addCleanup(self.ldb.delete, test_ou, ["tree_delete:1"])
+
+        # create the source and target objects and link them together.
+        # We use addressBookRoots2 here because it's a one-way link
+        src_dn = "CN=link_src,%s" % test_ou
+        self.ldb.add({"dn": src_dn,
+                      "objectclass": "msExchConfigurationContainer"})
+        target_dn = "OU=link_tgt,%s" % test_ou
+        self.ldb.add({"dn": target_dn, "objectclass": "organizationalunit"})
+        link_attr = "addressBookRoots2"
+        self.add_link(link_attr, src_dn, target_dn)
+
+        # add a second link target that's in a different partition
+        server_dn = ("CN=testrename,CN=Servers,CN=Default-First-Site-Name,"
+                     "CN=Sites,%s" % str(self.ldb.get_config_basedn()))
+        self.ldb.add({"dn": server_dn, "objectclass": "server"})
+        self.addCleanup(self.ldb.delete, server_dn)
+        self.add_link(link_attr, src_dn, server_dn)
+
+        # do the backup/restore
+        backup_file = self.create_backup()
+        self.restore_backup(backup_file)
+        lp = self.check_restored_smbconf()
+        restored_ldb = self.check_restored_database(lp)
+
+        # work out what the new DNs should be
+        old_basedn = str(self.ldb.get_default_basedn())
+        new_target_dn = re.sub(old_basedn + '$', self.new_basedn, target_dn)
+        new_src_dn = re.sub(old_basedn + '$', self.new_basedn, src_dn)
+        new_server_dn = re.sub(old_basedn + '$', self.new_basedn, server_dn)
+
+        # check the links exist in the renamed DB with the correct DNs
+        res = restored_ldb.search(base=new_src_dn, scope=ldb.SCOPE_BASE,
+                                  attrs=[link_attr])
+        self.assertEqual(len(res), 1,
+                         "Failed to find renamed link source object")
+        self.assertTrue(link_attr in res[0], "Missing link attribute")
+        self.assertTrue(new_target_dn in res[0][link_attr])
+        self.assertTrue(new_server_dn in res[0][link_attr])
+
+    # extra checks we run on the restored DB in the rename case
+    def check_restored_database(self, lp):
+        # run the common checks over the restored DB
+        samdb = super(DomainBackupRename, self).check_restored_database(lp)
+
+        # check we have actually renamed the DNs
+        basedn = str(samdb.get_default_basedn())
+        self.assertEqual(basedn, self.new_basedn)
+
+        # check the partition and netBIOS name match the new domain
+        partitions_dn = samdb.get_partitions_dn()
+        nc_name = ldb.binary_encode(str(basedn))
+        res = samdb.search(base=partitions_dn, scope=ldb.SCOPE_ONELEVEL,
+                           attrs=["nETBIOSName", "cn"],
+                           expression='ncName=%s' % nc_name)
+        self.assertEqual(len(res), 1,
+                         "Looking up partition's NetBIOS name failed")
+        self.assertEqual(str(res[0].get("nETBIOSName")), self.restore_domain)
+        self.assertEqual(str(res[0].get("cn")), self.restore_domain)
+
+        # check the DC has the correct dnsHostname
+        realm = self.restore_realm
+        dn = "CN=%s,OU=Domain Controllers,%s" % (self.new_server,
+                                                 self.new_basedn)
+        res = samdb.search(base=dn, scope=ldb.SCOPE_BASE,
+                           attrs=["dNSHostName"])
+        self.assertEqual(len(res), 1,
+                         "Looking up new DC's dnsHostname failed")
+        expected_val = "%s.%s" % (self.new_server.lower(), realm)
+        self.assertEqual(str(res[0].get("dNSHostName")), expected_val)
+
+        # check the DNS zones for the new realm are present
+        dn = "DC=%s,CN=MicrosoftDNS,DC=DomainDnsZones,%s" % (realm, basedn)
+        res = samdb.search(base=dn, scope=ldb.SCOPE_BASE)
+        self.assertEqual(len(res), 1, "Lookup of new domain's DNS zone failed")
+
+        forestdn = samdb.get_root_basedn().get_linearized()
+        dn = "DC=_msdcs.%s,CN=MicrosoftDNS,DC=ForestDnsZones,%s" % (realm,
+                                                                    forestdn)
+        res = samdb.search(base=dn, scope=ldb.SCOPE_BASE)
+        self.assertEqual(len(res), 1, "Lookup of new domain's DNS zone failed")
+        return samdb