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 from samba import credentials
  27 import subprocess
  28 import tempfile
  29 import unittest
  30
  31 try:
  32     from unittest import SkipTest as TestSkipped
  33 except ImportError:
  34     class TestSkipped(Exception):
  35         """Test skipped."""
  36
  37
  38 class TestCase(unittest.TestCase):
  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(unittest.TestCase):
  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 KeyError("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     if not "://" in samdb_url:
 187         if not ldap_only and os.path.isfile(samdb_url):
 188             samdb_url = "tdb://%s" % samdb_url
 189         else:
 190             samdb_url = "ldap://%s" % samdb_url
 191     # use 'paged_search' module when connecting remotely
 192     if samdb_url.startswith("ldap://"):
 193         ldb_options = ["modules:paged_searches"]
 194     elif ldap_only:
 195         raise AssertionError("Trying to connect to %s while remote "
 196                              "connection is required" % samdb_url)
 197
 198     # set defaults for test environment
 199     if lp is None:
 200         lp = env_loadparm()
 201     if session_info is None:
 202         session_info = samba.auth.system_session(lp)
 203     if credentials is None:
 204         credentials = cmdline_credentials
 205
 206     return SamDB(url=samdb_url,
 207                  lp=lp,
 208                  session_info=session_info,
 209                  credentials=credentials,
 210                  flags=flags,
 211                  options=ldb_options,
 212                  global_schema=global_schema)
 213
 214
 215 def connect_samdb_ex(samdb_url, lp=None, session_info=None, credentials=None,
 216                      flags=0, ldb_options=None, ldap_only=False):
 217     """Connects to samdb_url database
 218
 219     :param samdb_url: Url for database to connect to.
 220     :param lp: Optional loadparm object
 221     :param session_info: Optional session information
 222     :param credentials: Optional credentials, defaults to anonymous.
 223     :param flags: Optional LDB flags
 224     :param ldap_only: If set, only remote LDAP connection will be created.
 225     :return: (sam_db_connection, rootDse_record) tuple
 226     """
 227     sam_db = connect_samdb(samdb_url, lp, session_info, credentials,
 228                            flags, ldb_options, ldap_only)
 229     # fetch RootDse
 230     res = sam_db.search(base="", expression="", scope=ldb.SCOPE_BASE,
 231                         attrs=["*"])
 232     return (sam_db, res[0])
 233
 234
 235 def connect_samdb_env(env_url, env_username, env_password, lp=None):
 236     """Connect to SamDB by getting URL and Credentials from environment
 237
 238     :param env_url: Environment variable name to get lsb url from
 239     :param env_username: Username environment variable
 240     :param env_password: Password environment variable
 241     :return: sam_db_connection
 242     """
 243     samdb_url = env_get_var_value(env_url)
 244     creds = credentials.Credentials()
 245     if lp is None:
 246         # guess Credentials parameters here. Otherwise workstation
 247         # and domain fields are NULL and gencache code segfalts
 248         lp = param.LoadParm()
 249         creds.guess(lp)
 250     creds.set_username(env_get_var_value(env_username))
 251     creds.set_password(env_get_var_value(env_password))
 252     return connect_samdb(samdb_url, credentials=creds, lp=lp)
 253
 254
 255 def delete_force(samdb, dn):
 256     try:
 257         samdb.delete(dn)
 258     except ldb.LdbError, (num, errstr):
 259         assert num == ldb.ERR_NO_SUCH_OBJECT, "ldb.delete() failed: %s" % errstr