python/samba/tests/__init__.py

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