source4/scripting/python/samba/tests/__init__.py

   1 #!/usr/bin/env python
   2
   3 # Unix SMB/CIFS implementation.
   4 # Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2007-2010
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 3 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful,
  12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 # GNU General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18 #
  19
  20 """Samba Python tests."""
  21
  22 import os
  23 import ldb
  24 import samba
  25 import samba.auth
  26 from samba import param
  27 from samba.samdb import SamDB
  28 import subprocess
  29 import tempfile
  30
  31 samba.ensure_external_module("testtools", "testtools")
  32
  33 # Other modules import these two classes from here, for convenience:
  34 from testtools.testcase import (
  35     TestCase as TesttoolsTestCase,
  36     TestSkipped,
  37     )
  38
  39
  40 class TestCase(TesttoolsTestCase):
  41     """A Samba test case."""
  42
  43     def setUp(self):
  44         super(TestCase, self).setUp()
  45         test_debug_level = os.getenv("TEST_DEBUG_LEVEL")
  46         if test_debug_level is not None:
  47             test_debug_level = int(test_debug_level)
  48             self._old_debug_level = samba.get_debug_level()
  49             samba.set_debug_level(test_debug_level)
  50             self.addCleanup(samba.set_debug_level, test_debug_level)
  51
  52     def get_loadparm(self):
  53         return env_loadparm()
  54
  55     def get_credentials(self):
  56         return cmdline_credentials
  57
  58
  59 class LdbTestCase(TesttoolsTestCase):
  60     """Trivial test case for running tests against a LDB."""
  61
  62     def setUp(self):
  63         super(LdbTestCase, self).setUp()
  64         self.filename = os.tempnam()
  65         self.ldb = samba.Ldb(self.filename)
  66
  67     def set_modules(self, modules=[]):
  68         """Change the modules for this Ldb."""
  69         m = ldb.Message()
  70         m.dn = ldb.Dn(self.ldb, "@MODULES")
  71         m["@LIST"] = ",".join(modules)
  72         self.ldb.add(m)
  73         self.ldb = samba.Ldb(self.filename)
  74
  75
  76 class TestCaseInTempDir(TestCase):
  77
  78     def setUp(self):
  79         super(TestCaseInTempDir, self).setUp()
  80         self.tempdir = tempfile.mkdtemp()
  81
  82     def tearDown(self):
  83         super(TestCaseInTempDir, self).tearDown()
  84         self.assertEquals([], os.listdir(self.tempdir))
  85         os.rmdir(self.tempdir)
  86
  87
  88 def env_loadparm():
  89     lp = param.LoadParm()
  90     try:
  91         lp.load(os.environ["SMB_CONF_PATH"])
  92     except KeyError:
  93         raise Exception("SMB_CONF_PATH not set")
  94     return lp
  95
  96
  97 def env_get_var_value(var_name):
  98     """Returns value for variable in os.environ
  99
 100     Function throws AssertionError if variable is defined.
 101     Unit-test based python tests require certain input params
 102     to be set in environment, otherwise they can't be run
 103     """
 104     assert var_name in os.environ.keys(), "Please supply %s in environment" % var_name
 105     return os.environ[var_name]
 106
 107
 108 cmdline_credentials = None
 109
 110 class RpcInterfaceTestCase(TestCase):
 111     """DCE/RPC Test case."""
 112
 113
 114 class ValidNetbiosNameTests(TestCase):
 115
 116     def test_valid(self):
 117         self.assertTrue(samba.valid_netbios_name("FOO"))
 118
 119     def test_too_long(self):
 120         self.assertFalse(samba.valid_netbios_name("FOO"*10))
 121
 122     def test_invalid_characters(self):
 123         self.assertFalse(samba.valid_netbios_name("*BLA"))
 124
 125
 126 class BlackboxProcessError(subprocess.CalledProcessError):
 127     """This exception is raised when a process run by check_output() returns
 128     a non-zero exit status. Exception instance should contain
 129     the exact exit code (S.returncode), command line (S.cmd),
 130     process output (S.stdout) and process error stream (S.stderr)"""
 131     def __init__(self, returncode, cmd, stdout, stderr):
 132         super(BlackboxProcessError, self).__init__(returncode, cmd)
 133         self.stdout = stdout
 134         self.stderr = stderr
 135     def __str__(self):
 136         return "Command '%s'; exit status %d; stdout: '%s'; stderr: '%s'" % (self.cmd, self.returncode,
 137                                                                              self.stdout, self.stderr)
 138
 139 class BlackboxTestCase(TestCase):
 140     """Base test case for blackbox tests."""
 141
 142     def _make_cmdline(self, line):
 143         bindir = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../bin"))
 144         parts = line.split(" ")
 145         if os.path.exists(os.path.join(bindir, parts[0])):
 146             parts[0] = os.path.join(bindir, parts[0])
 147         line = " ".join(parts)
 148         return line
 149
 150     def check_run(self, line):
 151         line = self._make_cmdline(line)
 152         subprocess.check_call(line, shell=True)
 153
 154     def check_output(self, line):
 155         line = self._make_cmdline(line)
 156         p = subprocess.Popen(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, close_fds=True)
 157         retcode = p.wait()
 158         if retcode:
 159             raise BlackboxProcessError(retcode, line, p.stdout.read(), p.stderr.read())
 160         return p.stdout.read()
 161
 162 def connect_samdb(samdb_url, lp=None, session_info=None, credentials=None,
 163                   flags=0, ldb_options=None, ldap_only=False):
 164     """Create SamDB instance and connects to samdb_url database.
 165
 166     :param samdb_url: Url for database to connect to.
 167     :param lp: Optional loadparm object
 168     :param session_info: Optional session information
 169     :param credentials: Optional credentials, defaults to anonymous.
 170     :param flags: Optional LDB flags
 171     :param ldap_only: If set, only remote LDAP connection will be created.
 172
 173     Added value for tests is that we have a shorthand function
 174     to make proper URL for ldb.connect() while using default
 175     parameters for connection based on test environment
 176     """
 177     samdb_url = samdb_url.lower()
 178     if not "://" in samdb_url:
 179         if not ldap_only and os.path.isfile(samdb_url):
 180             samdb_url = "tdb://%s" % samdb_url
 181         else:
 182             samdb_url = "ldap://%s" % samdb_url
 183     # use 'paged_search' module when connecting remotely
 184     if samdb_url.startswith("ldap://"):
 185         ldb_options = ["modules:paged_searches"]
 186     elif ldap_only:
 187         raise AssertionError("Trying to connect to %s while remote "
 188                              "connection is required" % samdb_url)
 189
 190     # set defaults for test environment
 191     if lp is None:
 192         lp = env_loadparm()
 193     if session_info is None:
 194         session_info = samba.auth.system_session(lp)
 195     if credentials is None:
 196         credentials = cmdline_credentials
 197
 198     return SamDB(url=samdb_url,
 199                  lp=lp,
 200                  session_info=session_info,
 201                  credentials=credentials,
 202                  flags=flags,
 203                  options=ldb_options)
 204
 205
 206 def connect_samdb_ex(samdb_url, lp=None, session_info=None, credentials=None,
 207                      flags=0, ldb_options=None, ldap_only=False):
 208     """Connects to samdb_url database
 209
 210     :param samdb_url: Url for database to connect to.
 211     :param lp: Optional loadparm object
 212     :param session_info: Optional session information
 213     :param credentials: Optional credentials, defaults to anonymous.
 214     :param flags: Optional LDB flags
 215     :param ldap_only: If set, only remote LDAP connection will be created.
 216     :return: (sam_db_connection, rootDse_record) tuple
 217     """
 218     sam_db = connect_samdb(samdb_url, lp, session_info, credentials,
 219                            flags, ldb_options, ldap_only)
 220     # fetch RootDse
 221     res = sam_db.search(base="", expression="", scope=ldb.SCOPE_BASE,
 222                         attrs=["*"])
 223     return (sam_db, res[0])
 224
 225
 226 def delete_force(samdb, dn):
 227     try:
 228         samdb.delete(dn)
 229     except ldb.LdbError, (num, _):
 230         assert(num == ldb.ERR_NO_SUCH_OBJECT)