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("testtools", "testtools")
  30
  31 # Other modules import these two classes from here, for convenience:
  32 from testtools.testcase import (
  33     TestCase as TesttoolsTestCase,
  34     TestSkipped,
  35     )
  36
  37
  38 class TestCase(TesttoolsTestCase):
  39     """A Samba test case."""
  40
  41     def setUp(self):
  42         super(TestCase, self).setUp()
  43         test_debug_level = os.getenv("TEST_DEBUG_LEVEL")
  44         if test_debug_level is not None:
  45             test_debug_level = int(test_debug_level)
  46             self._old_debug_level = samba.get_debug_level()
  47             samba.set_debug_level(test_debug_level)
  48             self.addCleanup(samba.set_debug_level, test_debug_level)
  49
  50     def get_loadparm(self):
  51         return env_loadparm()
  52
  53     def get_credentials(self):
  54         return cmdline_credentials
  55
  56
  57 class LdbTestCase(TesttoolsTestCase):
  58     """Trivial test case for running tests against a LDB."""
  59
  60     def setUp(self):
  61         super(LdbTestCase, self).setUp()
  62         self.filename = os.tempnam()
  63         self.ldb = samba.Ldb(self.filename)
  64
  65     def set_modules(self, modules=[]):
  66         """Change the modules for this Ldb."""
  67         m = ldb.Message()
  68         m.dn = ldb.Dn(self.ldb, "@MODULES")
  69         m["@LIST"] = ",".join(modules)
  70         self.ldb.add(m)
  71         self.ldb = samba.Ldb(self.filename)
  72
  73
  74 class TestCaseInTempDir(TestCase):
  75
  76     def setUp(self):
  77         super(TestCaseInTempDir, self).setUp()
  78         self.tempdir = tempfile.mkdtemp()
  79         self.addCleanup(self._remove_tempdir)
  80
  81     def _remove_tempdir(self):
  82         self.assertEquals([], os.listdir(self.tempdir))
  83         os.rmdir(self.tempdir)
  84         self.tempdir = None
  85
  86
  87 def env_loadparm():
  88     lp = param.LoadParm()
  89     try:
  90         lp.load(os.environ["SMB_CONF_PATH"])
  91     except KeyError:
  92         raise Exception("SMB_CONF_PATH not set")
  93     return lp
  94
  95
  96 def env_get_var_value(var_name):
  97     """Returns value for variable in os.environ
  98
  99     Function throws AssertionError if variable is defined.
 100     Unit-test based python tests require certain input params
 101     to be set in environment, otherwise they can't be run
 102     """
 103     assert var_name in os.environ.keys(), "Please supply %s in environment" % var_name
 104     return os.environ[var_name]
 105
 106
 107 cmdline_credentials = None
 108
 109 class RpcInterfaceTestCase(TestCase):
 110     """DCE/RPC Test case."""
 111
 112
 113 class ValidNetbiosNameTests(TestCase):
 114
 115     def test_valid(self):
 116         self.assertTrue(samba.valid_netbios_name("FOO"))
 117
 118     def test_too_long(self):
 119         self.assertFalse(samba.valid_netbios_name("FOO"*10))
 120
 121     def test_invalid_characters(self):
 122         self.assertFalse(samba.valid_netbios_name("*BLA"))
 123
 124
 125 class BlackboxProcessError(Exception):
 126     """This is raised when check_output() process returns a non-zero exit status
 127
 128     Exception instance should contain the exact exit code (S.returncode),
 129     command line (S.cmd), process output (S.stdout) and process error stream
 130     (S.stderr)
 131     """
 132
 133     def __init__(self, returncode, cmd, stdout, stderr):
 134         self.returncode = returncode
 135         self.cmd = cmd
 136         self.stdout = stdout
 137         self.stderr = stderr
 138
 139     def __str__(self):
 140         return "Command '%s'; exit status %d; stdout: '%s'; stderr: '%s'" % (self.cmd, self.returncode,
 141                                                                              self.stdout, self.stderr)
 142
 143 class BlackboxTestCase(TestCase):
 144     """Base test case for blackbox tests."""
 145
 146     def _make_cmdline(self, line):
 147         bindir = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../bin"))
 148         parts = line.split(" ")
 149         if os.path.exists(os.path.join(bindir, parts[0])):
 150             parts[0] = os.path.join(bindir, parts[0])
 151         line = " ".join(parts)
 152         return line
 153
 154     def check_run(self, line):
 155         line = self._make_cmdline(line)
 156         p = subprocess.Popen(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
 157         retcode = p.wait()
 158         if retcode:
 159             raise BlackboxProcessError(retcode, line, p.stdout.read(), p.stderr.read())
 160
 161     def check_output(self, line):
 162         line = self._make_cmdline(line)
 163         p = subprocess.Popen(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, close_fds=True)
 164         retcode = p.wait()
 165         if retcode:
 166             raise BlackboxProcessError(retcode, line, p.stdout.read(), p.stderr.read())
 167         return p.stdout.read()
 168
 169
 170 def connect_samdb(samdb_url, lp=None, session_info=None, credentials=None,
 171                   flags=0, ldb_options=None, ldap_only=False, global_schema=True):
 172     """Create SamDB instance and connects to samdb_url database.
 173
 174     :param samdb_url: Url for database to connect to.
 175     :param lp: Optional loadparm object
 176     :param session_info: Optional session information
 177     :param credentials: Optional credentials, defaults to anonymous.
 178     :param flags: Optional LDB flags
 179     :param ldap_only: If set, only remote LDAP connection will be created.
 180     :param global_schema: Whether to use global schema.
 181
 182     Added value for tests is that we have a shorthand function
 183     to make proper URL for ldb.connect() while using default
 184     parameters for connection based on test environment
 185     """
 186     samdb_url = samdb_url.lower()
 187     if not "://" in samdb_url:
 188         if not ldap_only and os.path.isfile(samdb_url):
 189             samdb_url = "tdb://%s" % samdb_url
 190         else:
 191             samdb_url = "ldap://%s" % samdb_url
 192     # use 'paged_search' module when connecting remotely
 193     if samdb_url.startswith("ldap://"):
 194         ldb_options = ["modules:paged_searches"]
 195     elif ldap_only:
 196         raise AssertionError("Trying to connect to %s while remote "
 197                              "connection is required" % samdb_url)
 198
 199     # set defaults for test environment
 200     if lp is None:
 201         lp = env_loadparm()
 202     if session_info is None:
 203         session_info = samba.auth.system_session(lp)
 204     if credentials is None:
 205         credentials = cmdline_credentials
 206
 207     return SamDB(url=samdb_url,
 208                  lp=lp,
 209                  session_info=session_info,
 210                  credentials=credentials,
 211                  flags=flags,
 212                  options=ldb_options,
 213                  global_schema=global_schema)
 214
 215
 216 def connect_samdb_ex(samdb_url, lp=None, session_info=None, credentials=None,
 217                      flags=0, ldb_options=None, ldap_only=False):
 218     """Connects to samdb_url database
 219
 220     :param samdb_url: Url for database to connect to.
 221     :param lp: Optional loadparm object
 222     :param session_info: Optional session information
 223     :param credentials: Optional credentials, defaults to anonymous.
 224     :param flags: Optional LDB flags
 225     :param ldap_only: If set, only remote LDAP connection will be created.
 226     :return: (sam_db_connection, rootDse_record) tuple
 227     """
 228     sam_db = connect_samdb(samdb_url, lp, session_info, credentials,
 229                            flags, ldb_options, ldap_only)
 230     # fetch RootDse
 231     res = sam_db.search(base="", expression="", scope=ldb.SCOPE_BASE,
 232                         attrs=["*"])
 233     return (sam_db, res[0])
 234
 235
 236 def delete_force(samdb, dn):
 237     try:
 238         samdb.delete(dn)
 239     except ldb.LdbError, (num, _):
 240         assert(num == ldb.ERR_NO_SUCH_OBJECT)