s4:python tests __init__.py - do not depend on "subprocess.check_call()"
[ira/wip.git] / source4 / scripting / python / samba / tests / __init__.py
1 #!/usr/bin/env python
2
3 # Unix SMB/CIFS implementation.
4 # Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2007-2010
5 #
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 3 of the License, or
9 # (at your option) any later version.
10 #
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14 # GNU General Public License for more details.
15 #
16 # You should have received a copy of the GNU General Public License
17 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
18 #
19
20 """Samba Python tests."""
21
22 import os
23 import ldb
24 import samba
25 import samba.auth
26 from samba import param
27 from samba.samdb import SamDB
28 import subprocess
29 import tempfile
30
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
82     def tearDown(self):
83         super(TestCaseInTempDir, self).tearDown()
84         self.assertEquals([], os.listdir(self.tempdir))
85         os.rmdir(self.tempdir)
86
87
88 def env_loadparm():
89     lp = param.LoadParm()
90     try:
91         lp.load(os.environ["SMB_CONF_PATH"])
92     except KeyError:
93         raise Exception("SMB_CONF_PATH not set")
94     return lp
95
96
97 def env_get_var_value(var_name):
98     """Returns value for variable in os.environ
99
100     Function throws AssertionError if variable is defined.
101     Unit-test based python tests require certain input params
102     to be set in environment, otherwise they can't be run
103     """
104     assert var_name in os.environ.keys(), "Please supply %s in environment" % var_name
105     return os.environ[var_name]
106
107
108 cmdline_credentials = None
109
110 class RpcInterfaceTestCase(TestCase):
111     """DCE/RPC Test case."""
112
113
114 class ValidNetbiosNameTests(TestCase):
115
116     def test_valid(self):
117         self.assertTrue(samba.valid_netbios_name("FOO"))
118
119     def test_too_long(self):
120         self.assertFalse(samba.valid_netbios_name("FOO"*10))
121
122     def test_invalid_characters(self):
123         self.assertFalse(samba.valid_netbios_name("*BLA"))
124
125
126 class BlackboxProcessError(Exception):
127     """This is raised when check_output() process returns a non-zero exit status
128
129     Exception instance should contain the exact exit code (S.returncode),
130     command line (S.cmd), process output (S.stdout) and process error stream
131     (S.stderr)
132     """
133
134     def __init__(self, returncode, cmd, stdout, stderr):
135         self.returncode = returncode
136         self.cmd = cmd
137         self.stdout = stdout
138         self.stderr = stderr
139
140     def __str__(self):
141         return "Command '%s'; exit status %d; stdout: '%s'; stderr: '%s'" % (self.cmd, self.returncode,
142                                                                              self.stdout, self.stderr)
143
144 class BlackboxTestCase(TestCase):
145     """Base test case for blackbox tests."""
146
147     def _make_cmdline(self, line):
148         bindir = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../bin"))
149         parts = line.split(" ")
150         if os.path.exists(os.path.join(bindir, parts[0])):
151             parts[0] = os.path.join(bindir, parts[0])
152         line = " ".join(parts)
153         return line
154
155     def check_run(self, line):
156         line = self._make_cmdline(line)
157         p = subprocess.Popen(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
158         retcode = p.wait()
159         if retcode:
160             raise BlackboxProcessError(retcode, line, p.stdout.read(), p.stderr.read())
161
162     def check_output(self, line):
163         line = self._make_cmdline(line)
164         p = subprocess.Popen(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, close_fds=True)
165         retcode = p.wait()
166         if retcode:
167             raise BlackboxProcessError(retcode, line, p.stdout.read(), p.stderr.read())
168         return p.stdout.read()
169
170 def connect_samdb(samdb_url, lp=None, session_info=None, credentials=None,
171                   flags=0, ldb_options=None, ldap_only=False):
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
181     Added value for tests is that we have a shorthand function
182     to make proper URL for ldb.connect() while using default
183     parameters for connection based on test environment
184     """
185     samdb_url = samdb_url.lower()
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
213
214 def connect_samdb_ex(samdb_url, lp=None, session_info=None, credentials=None,
215                      flags=0, ldb_options=None, ldap_only=False):
216     """Connects to samdb_url database
217
218     :param samdb_url: Url for database to connect to.
219     :param lp: Optional loadparm object
220     :param session_info: Optional session information
221     :param credentials: Optional credentials, defaults to anonymous.
222     :param flags: Optional LDB flags
223     :param ldap_only: If set, only remote LDAP connection will be created.
224     :return: (sam_db_connection, rootDse_record) tuple
225     """
226     sam_db = connect_samdb(samdb_url, lp, session_info, credentials,
227                            flags, ldb_options, ldap_only)
228     # fetch RootDse
229     res = sam_db.search(base="", expression="", scope=ldb.SCOPE_BASE,
230                         attrs=["*"])
231     return (sam_db, res[0])
232
233
234 def delete_force(samdb, dn):
235     try:
236         samdb.delete(dn)
237     except ldb.LdbError, (num, _):
238         assert(num == ldb.ERR_NO_SUCH_OBJECT)