source4/scripting/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 def connect_samdb(samdb_url, lp=None, session_info=None, credentials=None,
 170                   flags=0, ldb_options=None, ldap_only=False):
 171     """Create SamDB instance and connects to samdb_url database.
 172
 173     :param samdb_url: Url for database to connect to.
 174     :param lp: Optional loadparm object
 175     :param session_info: Optional session information
 176     :param credentials: Optional credentials, defaults to anonymous.
 177     :param flags: Optional LDB flags
 178     :param ldap_only: If set, only remote LDAP connection will be created.
 179
 180     Added value for tests is that we have a shorthand function
 181     to make proper URL for ldb.connect() while using default
 182     parameters for connection based on test environment
 183     """
 184     samdb_url = samdb_url.lower()
 185     if not "://" in samdb_url:
 186         if not ldap_only and os.path.isfile(samdb_url):
 187             samdb_url = "tdb://%s" % samdb_url
 188         else:
 189             samdb_url = "ldap://%s" % samdb_url
 190     # use 'paged_search' module when connecting remotely
 191     if samdb_url.startswith("ldap://"):
 192         ldb_options = ["modules:paged_searches"]
 193     elif ldap_only:
 194         raise AssertionError("Trying to connect to %s while remote "
 195                              "connection is required" % samdb_url)
 196
 197     # set defaults for test environment
 198     if lp is None:
 199         lp = env_loadparm()
 200     if session_info is None:
 201         session_info = samba.auth.system_session(lp)
 202     if credentials is None:
 203         credentials = cmdline_credentials
 204
 205     return SamDB(url=samdb_url,
 206                  lp=lp,
 207                  session_info=session_info,
 208                  credentials=credentials,
 209                  flags=flags,
 210                  options=ldb_options)
 211
 212
 213 def connect_samdb_ex(samdb_url, lp=None, session_info=None, credentials=None,
 214                      flags=0, ldb_options=None, ldap_only=False):
 215     """Connects to samdb_url database
 216
 217     :param samdb_url: Url for database to connect to.
 218     :param lp: Optional loadparm object
 219     :param session_info: Optional session information
 220     :param credentials: Optional credentials, defaults to anonymous.
 221     :param flags: Optional LDB flags
 222     :param ldap_only: If set, only remote LDAP connection will be created.
 223     :return: (sam_db_connection, rootDse_record) tuple
 224     """
 225     sam_db = connect_samdb(samdb_url, lp, session_info, credentials,
 226                            flags, ldb_options, ldap_only)
 227     # fetch RootDse
 228     res = sam_db.search(base="", expression="", scope=ldb.SCOPE_BASE,
 229                         attrs=["*"])
 230     return (sam_db, res[0])
 231
 232
 233 def delete_force(samdb, dn):
 234     try:
 235         samdb.delete(dn)
 236     except ldb.LdbError, (num, _):
 237         assert(num == ldb.ERR_NO_SUCH_OBJECT)