Gecko [ lorencats.com ]

Name	Size	Permission
__init__.py	3.3 KB	-rw-r--r--
__init__.pyc	3.25 KB	-rw-r--r--
graph.py	2.33 KB	-rw-r--r--
graph.pyc	2.23 KB	-rw-r--r--
graph_utils.py	5.27 KB	-rw-r--r--
graph_utils.pyc	4.99 KB	-rw-r--r--
kcc_utils.py	13.53 KB	-rw-r--r--
kcc_utils.pyc	9.86 KB	-rw-r--r--
ldif_import_export.py	9.68 KB	-rw-r--r--
ldif_import_export.pyc	10.78 KB	-rw-r--r--

Code Editor : ldif_import_export.py

# Unix SMB/CIFS implementation. Tests for samba.kcc.ldif_import_export.
# Copyright (C) Andrew Bartlett 2015
#
# Written by Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

"""Tests for samba.kcc.ldif_import_export"""

import samba
import os
import time
import shutil
import sys
import subprocess
import logging
import samba.tests
from samba.kcc import ldif_import_export, KCC
from samba import ldb
from samba.dcerpc import misc

from samba.param import LoadParm
from samba.credentials import Credentials
from samba.samdb import SamDB

unix_now = int(time.time())

MULTISITE_LDIF = os.path.join(os.environ['SRCDIR_ABS'],
                              "testdata/ldif-utils-test-multisite.ldif")

# UNCONNECTED_LDIF is a single site, unconnected 5DC database that was
# created using samba-tool domain join in testenv.
UNCONNECTED_LDIF = os.path.join(os.environ['SRCDIR_ABS'],
        "testdata/unconnected-intrasite.ldif")

MULTISITE_LDIF_DSAS = (
    ("CN=WIN08,CN=Servers,CN=Site-4,CN=Sites,CN=Configuration,DC=ad,DC=samba,DC=example,DC=com",
     "Site-4"),
    ("CN=WIN07,CN=Servers,CN=Site-4,CN=Sites,CN=Configuration,DC=ad,DC=samba,DC=example,DC=com",
     "Site-4"),
    ("CN=WIN06,CN=Servers,CN=Site-3,CN=Sites,CN=Configuration,DC=ad,DC=samba,DC=example,DC=com",
     "Site-3"),
    ("CN=WIN09,CN=Servers,CN=Site-5,CN=Sites,CN=Configuration,DC=ad,DC=samba,DC=example,DC=com",
     "Site-5"),
    ("CN=WIN10,CN=Servers,CN=Site-5,CN=Sites,CN=Configuration,DC=ad,DC=samba,DC=example,DC=com",
     "Site-5"),
    ("CN=WIN02,CN=Servers,CN=Site-2,CN=Sites,CN=Configuration,DC=ad,DC=samba,DC=example,DC=com",
     "Site-2"),
    ("CN=WIN04,CN=Servers,CN=Site-2,CN=Sites,CN=Configuration,DC=ad,DC=samba,DC=example,DC=com",
     "Site-2"),
    ("CN=WIN03,CN=Servers,CN=Site-2,CN=Sites,CN=Configuration,DC=ad,DC=samba,DC=example,DC=com",
     "Site-2"),
    ("CN=WIN05,CN=Servers,CN=Site-2,CN=Sites,CN=Configuration,DC=ad,DC=samba,DC=example,DC=com",
     "Site-2"),
    ("CN=WIN01,CN=Servers,CN=Default-First-Site-Name,CN=Sites,CN=Configuration,DC=ad,DC=samba,DC=example,DC=com",
     "Default-First-Site-Name"),
)

class LdifImportExportTests(samba.tests.TestCaseInTempDir):
    def setUp(self):
        super(LdifImportExportTests, self).setUp()
        self.lp = LoadParm()
        self.creds = Credentials()
        self.creds.guess(self.lp)

def remove_files(self, *files):
        for f in files:
            assert(f.startswith(self.tempdir))
            os.unlink(f)

def test_write_search_url(self):
        pass

def test_ldif_to_samdb(self):
        dburl = os.path.join(self.tempdir, "ldap")
        samdb = ldif_import_export.ldif_to_samdb(dburl, self.lp,
                                                 MULTISITE_LDIF)
        self.assertIsInstance(samdb, SamDB)

dsa = ("CN=WIN01,CN=Servers,CN=Default-First-Site-Name,CN=Sites,"
               "CN=Configuration,DC=ad,DC=samba,DC=example,DC=com")
        res = samdb.search(ldb.Dn(samdb, "CN=NTDS Settings," + dsa),
                           scope=ldb.SCOPE_BASE, attrs=["objectGUID"])

ntds_guid = misc.GUID(samdb.get_ntds_GUID())
        self.assertEqual(misc.GUID(res[0]["objectGUID"][0]), ntds_guid)

service_name_res = samdb.search(base="",
                                        scope=ldb.SCOPE_BASE,
                                        attrs=["dsServiceName"])
        dn = ldb.Dn(samdb,
                    service_name_res[0]["dsServiceName"][0].decode('utf8'))
        self.assertEqual(dn, ldb.Dn(samdb, "CN=NTDS Settings," + dsa))
        self.remove_files(dburl)

def test_ldif_to_samdb_forced_local_dsa(self):
        for dsa, site in MULTISITE_LDIF_DSAS:
            dburl = os.path.join(self.tempdir, "ldif-to-samba-forced-local-dsa"
                                 "-%s" % dsa)
            samdb = ldif_import_export.ldif_to_samdb(dburl, self.lp,
                                                     MULTISITE_LDIF,
                                                     forced_local_dsa=dsa)
            self.assertIsInstance(samdb, SamDB)
            self.assertEqual(samdb.server_site_name(), site)

res = samdb.search(ldb.Dn(samdb, "CN=NTDS Settings," + dsa),
                               scope=ldb.SCOPE_BASE, attrs=["objectGUID"])

ntds_guid = misc.GUID(samdb.get_ntds_GUID())
            self.assertEqual(misc.GUID(res[0]["objectGUID"][0]), ntds_guid)

service_name_res = samdb.search(base="",
                                            scope=ldb.SCOPE_BASE,
                                            attrs=["dsServiceName"])
            dn = ldb.Dn(samdb,
                        service_name_res[0]["dsServiceName"][0].decode('utf8'))
            self.assertEqual(dn, ldb.Dn(samdb, "CN=NTDS Settings," + dsa))
            self.remove_files(dburl)

def test_samdb_to_ldif_file(self):
        dburl = os.path.join(self.tempdir, "ldap")
        dburl2 = os.path.join(self.tempdir, "ldap_roundtrip")
        ldif_file = os.path.join(self.tempdir, "ldif")
        samdb = ldif_import_export.ldif_to_samdb(dburl, self.lp,
                                                 MULTISITE_LDIF)
        self.assertIsInstance(samdb, SamDB)
        ldif_import_export.samdb_to_ldif_file(samdb, dburl,
                                              lp=self.lp, creds=None,
                                              ldif_file=ldif_file)
        self.assertGreater(os.path.getsize(ldif_file), 1000,
                           "LDIF should be larger than 1000 bytes")
        samdb = ldif_import_export.ldif_to_samdb(dburl2, self.lp,
                                                 ldif_file)
        self.assertIsInstance(samdb, SamDB)
        dsa = ("CN=WIN01,CN=Servers,CN=Default-First-Site-Name,CN=Sites,"
               "CN=Configuration,DC=ad,DC=samba,DC=example,DC=com")
        res = samdb.search(ldb.Dn(samdb, "CN=NTDS Settings," + dsa),
                           scope=ldb.SCOPE_BASE, attrs=["objectGUID"])
        self.remove_files(dburl)
        self.remove_files(dburl2)
        self.remove_files(ldif_file)

class KCCMultisiteLdifTests(samba.tests.TestCaseInTempDir):
    def setUp(self):
        super(KCCMultisiteLdifTests, self).setUp()
        self.lp = LoadParm()
        self.creds = Credentials()
        self.creds.guess(self.lp)

def remove_files(self, *files):
        for f in files:
            assert(f.startswith(self.tempdir))
            os.unlink(f)

def _get_kcc(self, name, readonly=False, verify=False, dot_file_dir=None):
        # Note that setting read-only to False won't affect the ldif,
        # only the temporary database that is created from it.
        my_kcc = KCC(unix_now, readonly=readonly, verify=verify,
                     dot_file_dir=dot_file_dir)
        tmpdb = os.path.join(self.tempdir, 'tmpdb')
        my_kcc.import_ldif(tmpdb, self.lp, MULTISITE_LDIF)
        self.remove_files(tmpdb)
        return my_kcc

def test_list_dsas(self):
        my_kcc = self._get_kcc('test-list')
        dsas = set(my_kcc.list_dsas())
        expected_dsas = set(x[0] for x in MULTISITE_LDIF_DSAS)
        self.assertEqual(dsas, expected_dsas)

def test_verify(self):
        """Check that the KCC generates graphs that pass its own verify
        option.
        """
        my_kcc = self._get_kcc('test-verify', verify=True)
        tmpdb = os.path.join(self.tempdir, 'verify-tmpdb')
        my_kcc.import_ldif(tmpdb, self.lp, MULTISITE_LDIF)

my_kcc.run(None,
                   self.lp, self.creds,
                   attempt_live_connections=False)
        self.remove_files(tmpdb)

def test_unconnected_db(self):
        """Check that the KCC generates errors on a unconnected db
        """
        my_kcc = self._get_kcc('test-verify', verify=True)
        tmpdb = os.path.join(self.tempdir, 'verify-tmpdb')
        my_kcc.import_ldif(tmpdb, self.lp, UNCONNECTED_LDIF)

try:
            my_kcc.run(None,
                       self.lp, self.creds,
                       attempt_live_connections=False)
        except samba.kcc.graph_utils.GraphError:
            pass
        except Exception:
            self.fail("Did not expect this error.")
        finally:
            self.remove_files(tmpdb)

def test_dotfiles(self):
        """Check that KCC writes dot_files when asked.
        """
        my_kcc = self._get_kcc('test-dotfiles', dot_file_dir=self.tempdir)
        tmpdb = os.path.join(self.tempdir, 'dotfile-tmpdb')
        files = [tmpdb]
        my_kcc.import_ldif(tmpdb, self.lp, MULTISITE_LDIF)
        my_kcc.run(None,
                   self.lp, self.creds,
                   attempt_live_connections=False)

dot = '/usr/bin/dot'
        for fn in os.listdir(self.tempdir):
            if fn.endswith('.dot'):
                ffn = os.path.join(self.tempdir, fn)
                if os.path.exists(dot) and subprocess.call([dot, '-?']) == 0:
                    r = subprocess.call([dot, '-Tcanon', ffn])
                    self.assertEqual(r, 0)

#even if dot is not there, at least check the file is non-empty
                size = os.stat(ffn).st_size
                self.assertNotEqual(size, 0)
                files.append(ffn)

self.remove_files(*files)

${this.title}

Code Editor : ldif_import_export.py