1 # Unix SMB/CIFS implementation.
2 # Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2007-2010
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.
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.
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/>.
18 """Samba Python tests."""
24 from samba import param
25 from samba.samdb import SamDB
26 from samba import credentials
30 samba.ensure_external_module("mimeparse", "mimeparse")
31 samba.ensure_external_module("extras", "extras")
32 samba.ensure_external_module("testtools", "testtools")
34 # Other modules import these two classes from here, for convenience:
35 from testtools.testcase import (
36 TestCase as TesttoolsTestCase,
41 class TestCase(TesttoolsTestCase):
42 """A Samba test case."""
45 super(TestCase, self).setUp()
46 test_debug_level = os.getenv("TEST_DEBUG_LEVEL")
47 if test_debug_level is not None:
48 test_debug_level = int(test_debug_level)
49 self._old_debug_level = samba.get_debug_level()
50 samba.set_debug_level(test_debug_level)
51 self.addCleanup(samba.set_debug_level, test_debug_level)
53 def get_loadparm(self):
56 def get_credentials(self):
57 return cmdline_credentials
60 class LdbTestCase(TesttoolsTestCase):
61 """Trivial test case for running tests against a LDB."""
64 super(LdbTestCase, self).setUp()
65 self.filename = os.tempnam()
66 self.ldb = samba.Ldb(self.filename)
68 def set_modules(self, modules=[]):
69 """Change the modules for this Ldb."""
71 m.dn = ldb.Dn(self.ldb, "@MODULES")
72 m["@LIST"] = ",".join(modules)
74 self.ldb = samba.Ldb(self.filename)
77 class TestCaseInTempDir(TestCase):
80 super(TestCaseInTempDir, self).setUp()
81 self.tempdir = tempfile.mkdtemp()
82 self.addCleanup(self._remove_tempdir)
84 def _remove_tempdir(self):
85 self.assertEquals([], os.listdir(self.tempdir))
86 os.rmdir(self.tempdir)
93 lp.load(os.environ["SMB_CONF_PATH"])
95 raise KeyError("SMB_CONF_PATH not set")
99 def env_get_var_value(var_name):
100 """Returns value for variable in os.environ
102 Function throws AssertionError if variable is defined.
103 Unit-test based python tests require certain input params
104 to be set in environment, otherwise they can't be run
106 assert var_name in os.environ.keys(), "Please supply %s in environment" % var_name
107 return os.environ[var_name]
110 cmdline_credentials = None
112 class RpcInterfaceTestCase(TestCase):
113 """DCE/RPC Test case."""
116 class ValidNetbiosNameTests(TestCase):
118 def test_valid(self):
119 self.assertTrue(samba.valid_netbios_name("FOO"))
121 def test_too_long(self):
122 self.assertFalse(samba.valid_netbios_name("FOO"*10))
124 def test_invalid_characters(self):
125 self.assertFalse(samba.valid_netbios_name("*BLA"))
128 class BlackboxProcessError(Exception):
129 """This is raised when check_output() process returns a non-zero exit status
131 Exception instance should contain the exact exit code (S.returncode),
132 command line (S.cmd), process output (S.stdout) and process error stream
136 def __init__(self, returncode, cmd, stdout, stderr):
137 self.returncode = returncode
143 return "Command '%s'; exit status %d; stdout: '%s'; stderr: '%s'" % (self.cmd, self.returncode,
144 self.stdout, self.stderr)
146 class BlackboxTestCase(TestCase):
147 """Base test case for blackbox tests."""
149 def _make_cmdline(self, line):
150 bindir = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../bin"))
151 parts = line.split(" ")
152 if os.path.exists(os.path.join(bindir, parts[0])):
153 parts[0] = os.path.join(bindir, parts[0])
154 line = " ".join(parts)
157 def check_run(self, line):
158 line = self._make_cmdline(line)
159 p = subprocess.Popen(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
162 raise BlackboxProcessError(retcode, line, p.stdout.read(), p.stderr.read())
164 def check_output(self, line):
165 line = self._make_cmdline(line)
166 p = subprocess.Popen(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, close_fds=True)
169 raise BlackboxProcessError(retcode, line, p.stdout.read(), p.stderr.read())
170 return p.stdout.read()
173 def connect_samdb(samdb_url, lp=None, session_info=None, credentials=None,
174 flags=0, ldb_options=None, ldap_only=False, global_schema=True):
175 """Create SamDB instance and connects to samdb_url database.
177 :param samdb_url: Url for database to connect to.
178 :param lp: Optional loadparm object
179 :param session_info: Optional session information
180 :param credentials: Optional credentials, defaults to anonymous.
181 :param flags: Optional LDB flags
182 :param ldap_only: If set, only remote LDAP connection will be created.
183 :param global_schema: Whether to use global schema.
185 Added value for tests is that we have a shorthand function
186 to make proper URL for ldb.connect() while using default
187 parameters for connection based on test environment
189 if not "://" in samdb_url:
190 if not ldap_only and os.path.isfile(samdb_url):
191 samdb_url = "tdb://%s" % samdb_url
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"]
198 raise AssertionError("Trying to connect to %s while remote "
199 "connection is required" % samdb_url)
201 # set defaults for test environment
204 if session_info is None:
205 session_info = samba.auth.system_session(lp)
206 if credentials is None:
207 credentials = cmdline_credentials
209 return SamDB(url=samdb_url,
211 session_info=session_info,
212 credentials=credentials,
215 global_schema=global_schema)
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
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
230 sam_db = connect_samdb(samdb_url, lp, session_info, credentials,
231 flags, ldb_options, ldap_only)
233 res = sam_db.search(base="", expression="", scope=ldb.SCOPE_BASE,
235 return (sam_db, res[0])
238 def connect_samdb_env(env_url, env_username, env_password, lp=None):
239 """Connect to SamDB by getting URL and Credentials from environment
241 :param env_url: Environment variable name to get lsb url from
242 :param env_username: Username environment variable
243 :param env_password: Password environment variable
244 :return: sam_db_connection
246 samdb_url = env_get_var_value(env_url)
247 creds = credentials.Credentials()
249 # guess Credentials parameters here. Otherwise workstation
250 # and domain fields are NULL and gencache code segfalts
251 lp = param.LoadParm()
253 creds.set_username(env_get_var_value(env_username))
254 creds.set_password(env_get_var_value(env_password))
255 return connect_samdb(samdb_url, credentials=creds, lp=lp)
258 def delete_force(samdb, dn):
261 except ldb.LdbError, (num, errstr):
262 assert num == ldb.ERR_NO_SUCH_OBJECT, "ldb.delete() failed: %s" % errstr