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 Exception("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     samdb_url = samdb_url.lower()
 189     if not "://" in samdb_url:
 190         if not ldap_only and os.path.isfile(samdb_url):
 191             samdb_url = "tdb://%s" % samdb_url
 192         else:
 193             samdb_url = "ldap://%s" % samdb_url
 194     # use 'paged_search' module when connecting remotely
 195     if samdb_url.startswith("ldap://"):
 196         ldb_options = ["modules:paged_searches"]
 197     elif ldap_only:
 198         raise AssertionError("Trying to connect to %s while remote "
 199                              "connection is required" % samdb_url)
 200
 201     # set defaults for test environment
 202     if lp is None:
 203         lp = env_loadparm()
 204     if session_info is None:
 205         session_info = samba.auth.system_session(lp)
 206     if credentials is None:
 207         credentials = cmdline_credentials
 208
 209     return SamDB(url=samdb_url,
 210                  lp=lp,
 211                  session_info=session_info,
 212                  credentials=credentials,
 213                  flags=flags,
 214                  options=ldb_options,
 215                  global_schema=global_schema)
 216
 217
 218 def connect_samdb_ex(samdb_url, lp=None, session_info=None, credentials=None,
 219                      flags=0, ldb_options=None, ldap_only=False):
 220     """Connects to samdb_url database
 221
 222     :param samdb_url: Url for database to connect to.
 223     :param lp: Optional loadparm object
 224     :param session_info: Optional session information
 225     :param credentials: Optional credentials, defaults to anonymous.
 226     :param flags: Optional LDB flags
 227     :param ldap_only: If set, only remote LDAP connection will be created.
 228     :return: (sam_db_connection, rootDse_record) tuple
 229     """
 230     sam_db = connect_samdb(samdb_url, lp, session_info, credentials,
 231                            flags, ldb_options, ldap_only)
 232     # fetch RootDse
 233     res = sam_db.search(base="", expression="", scope=ldb.SCOPE_BASE,
 234                         attrs=["*"])
 235     return (sam_db, res[0])
 236
 237
 238 def delete_force(samdb, dn):
 239     try:
 240         samdb.delete(dn)
 241     except ldb.LdbError, (num, _):
 242         assert(num == ldb.ERR_NO_SUCH_OBJECT)