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
  80     def tearDown(self):
  81         super(TestCaseInTempDir, self).tearDown()
  82         self.assertEquals([], os.listdir(self.tempdir))
  83         os.rmdir(self.tempdir)
  84
  85
  86 def env_loadparm():
  87     lp = param.LoadParm()
  88     try:
  89         lp.load(os.environ["SMB_CONF_PATH"])
  90     except KeyError:
  91         raise Exception("SMB_CONF_PATH not set")
  92     return lp
  93
  94
  95 def env_get_var_value(var_name):
  96     """Returns value for variable in os.environ
  97
  98     Function throws AssertionError if variable is defined.
  99     Unit-test based python tests require certain input params
 100     to be set in environment, otherwise they can't be run
 101     """
 102     assert var_name in os.environ.keys(), "Please supply %s in environment" % var_name
 103     return os.environ[var_name]
 104
 105
 106 cmdline_credentials = None
 107
 108 class RpcInterfaceTestCase(TestCase):
 109     """DCE/RPC Test case."""
 110
 111
 112 class ValidNetbiosNameTests(TestCase):
 113
 114     def test_valid(self):
 115         self.assertTrue(samba.valid_netbios_name("FOO"))
 116
 117     def test_too_long(self):
 118         self.assertFalse(samba.valid_netbios_name("FOO"*10))
 119
 120     def test_invalid_characters(self):
 121         self.assertFalse(samba.valid_netbios_name("*BLA"))
 122
 123
 124 class BlackboxProcessError(Exception):
 125     """This is raised when check_output() process returns a non-zero exit status
 126
 127     Exception instance should contain the exact exit code (S.returncode),
 128     command line (S.cmd), process output (S.stdout) and process error stream
 129     (S.stderr)
 130     """
 131
 132     def __init__(self, returncode, cmd, stdout, stderr):
 133         self.returncode = returncode
 134         self.cmd = cmd
 135         self.stdout = stdout
 136         self.stderr = stderr
 137
 138     def __str__(self):
 139         return "Command '%s'; exit status %d; stdout: '%s'; stderr: '%s'" % (self.cmd, self.returncode,
 140                                                                              self.stdout, self.stderr)
 141
 142 class BlackboxTestCase(TestCase):
 143     """Base test case for blackbox tests."""
 144
 145     def _make_cmdline(self, line):
 146         bindir = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../bin"))
 147         parts = line.split(" ")
 148         if os.path.exists(os.path.join(bindir, parts[0])):
 149             parts[0] = os.path.join(bindir, parts[0])
 150         line = " ".join(parts)
 151         return line
 152
 153     def check_run(self, line):
 154         line = self._make_cmdline(line)
 155         p = subprocess.Popen(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
 156         retcode = p.wait()
 157         if retcode:
 158             raise BlackboxProcessError(retcode, line, p.stdout.read(), p.stderr.read())
 159
 160     def check_output(self, line):
 161         line = self._make_cmdline(line)
 162         p = subprocess.Popen(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, close_fds=True)
 163         retcode = p.wait()
 164         if retcode:
 165             raise BlackboxProcessError(retcode, line, p.stdout.read(), p.stderr.read())
 166         return p.stdout.read()
 167
 168 def connect_samdb(samdb_url, lp=None, session_info=None, credentials=None,
 169                   flags=0, ldb_options=None, ldap_only=False):
 170     """Create SamDB instance and connects to samdb_url database.
 171
 172     :param samdb_url: Url for database to connect to.
 173     :param lp: Optional loadparm object
 174     :param session_info: Optional session information
 175     :param credentials: Optional credentials, defaults to anonymous.
 176     :param flags: Optional LDB flags
 177     :param ldap_only: If set, only remote LDAP connection will be created.
 178
 179     Added value for tests is that we have a shorthand function
 180     to make proper URL for ldb.connect() while using default
 181     parameters for connection based on test environment
 182     """
 183     samdb_url = samdb_url.lower()
 184     if not "://" in samdb_url:
 185         if not ldap_only and os.path.isfile(samdb_url):
 186             samdb_url = "tdb://%s" % samdb_url
 187         else:
 188             samdb_url = "ldap://%s" % samdb_url
 189     # use 'paged_search' module when connecting remotely
 190     if samdb_url.startswith("ldap://"):
 191         ldb_options = ["modules:paged_searches"]
 192     elif ldap_only:
 193         raise AssertionError("Trying to connect to %s while remote "
 194                              "connection is required" % samdb_url)
 195
 196     # set defaults for test environment
 197     if lp is None:
 198         lp = env_loadparm()
 199     if session_info is None:
 200         session_info = samba.auth.system_session(lp)
 201     if credentials is None:
 202         credentials = cmdline_credentials
 203
 204     return SamDB(url=samdb_url,
 205                  lp=lp,
 206                  session_info=session_info,
 207                  credentials=credentials,
 208                  flags=flags,
 209                  options=ldb_options)
 210
 211
 212 def connect_samdb_ex(samdb_url, lp=None, session_info=None, credentials=None,
 213                      flags=0, ldb_options=None, ldap_only=False):
 214     """Connects to samdb_url database
 215
 216     :param samdb_url: Url for database to connect to.
 217     :param lp: Optional loadparm object
 218     :param session_info: Optional session information
 219     :param credentials: Optional credentials, defaults to anonymous.
 220     :param flags: Optional LDB flags
 221     :param ldap_only: If set, only remote LDAP connection will be created.
 222     :return: (sam_db_connection, rootDse_record) tuple
 223     """
 224     sam_db = connect_samdb(samdb_url, lp, session_info, credentials,
 225                            flags, ldb_options, ldap_only)
 226     # fetch RootDse
 227     res = sam_db.search(base="", expression="", scope=ldb.SCOPE_BASE,
 228                         attrs=["*"])
 229     return (sam_db, res[0])
 230
 231
 232 def delete_force(samdb, dn):
 233     try:
 234         samdb.delete(dn)
 235     except ldb.LdbError, (num, _):
 236         assert(num == ldb.ERR_NO_SUCH_OBJECT)