Handle skips when running on python2.6.

[samba.git] / python / samba / tests / __init__.py

diff --git a/python/samba/tests/__init__.py b/python/samba/tests/__init__.py
index 952b6eecdd147eb01ebbb0dec1d29ee8cfdaacf0..9c82abd66813d40df7495f922ca646e533c4ec4a 100644 (file)
--- a/python/samba/tests/__init__.py
+++ b/python/samba/tests/__init__.py
@@ -23,21 +23,20 @@ import samba
 import samba.auth
 from samba import param
 from samba.samdb import SamDB
+from samba import credentials
 import subprocess
+import sys
 import tempfile
+import unittest
 
-samba.ensure_external_module("mimeparse", "mimeparse")
-samba.ensure_external_module("extras", "extras")
-samba.ensure_external_module("testtools", "testtools")
+try:
+    from unittest import SkipTest
+except ImportError:
+    class SkipTest(Exception):
+        """Test skipped."""
 
-# Other modules import these two classes from here, for convenience:
-from testtools.testcase import (
-    TestCase as TesttoolsTestCase,
-    TestSkipped,
-    )
 
-
-class TestCase(TesttoolsTestCase):
+class TestCase(unittest.TestCase):
     """A Samba test case."""
 
     def setUp(self):
@@ -55,8 +54,83 @@ class TestCase(TesttoolsTestCase):
     def get_credentials(self):
         return cmdline_credentials
 
-
-class LdbTestCase(TesttoolsTestCase):
+    # These functions didn't exist before Python2.7:
+    if sys.version_info < (2, 7):
+        import warnings
+
+        def skipTest(self, reason):
+            raise SkipTest(reason)
+
+        def assertIs(self, a, b):
+            self.assertTrue(a is b)
+
+        def assertIsNot(self, a, b):
+            self.assertTrue(a is not b)
+
+        def assertIsInstance(self, a, b):
+            self.assertTrue(isinstance(a, b))
+
+        def addCleanup(self, fn, *args, **kwargs):
+            self._cleanups = getattr(self, "_cleanups", []) + [
+                (fn, args, kwargs)]
+
+        def _addSkip(self, result, reason):
+            addSkip = getattr(result, 'addSkip', None)
+            if addSkip is not None:
+                addSkip(self, reason)
+            else:
+                warnings.warn("TestResult has no addSkip method, skips not reported",
+                              RuntimeWarning, 2)
+                result.addSuccess(self)
+
+        def run(self, result=None):
+            if result is None: result = self.defaultTestResult()
+            result.startTest(self)
+            testMethod = getattr(self, self._testMethodName)
+            try:
+                try:
+                    self.setUp()
+                except SkipTest, e:
+                    self._addSkip(result, str(e))
+                    return
+                except KeyboardInterrupt:
+                    raise
+                except:
+                    result.addError(self, self._exc_info())
+                    return
+
+                ok = False
+                try:
+                    testMethod()
+                    ok = True
+                except SkipTest, e:
+                    self._addSkip(result, str(e))
+                    return
+                except self.failureException:
+                    result.addFailure(self, self._exc_info())
+                except KeyboardInterrupt:
+                    raise
+                except:
+                    result.addError(self, self._exc_info())
+
+                try:
+                    self.tearDown()
+                except SkipTest, e:
+                    self._addSkip(result, str(e))
+                except KeyboardInterrupt:
+                    raise
+                except:
+                    result.addError(self, self._exc_info())
+                    ok = False
+
+                for (fn, args, kwargs) in reversed(getattr(self, "_cleanups", [])):
+                    fn(*args, **kwargs)
+                if ok: result.addSuccess(self)
+            finally:
+                result.stopTest(self)
+
+
+class LdbTestCase(TestCase):
     """Trivial test case for running tests against a LDB."""
 
     def setUp(self):
@@ -234,6 +308,26 @@ def connect_samdb_ex(samdb_url, lp=None, session_info=None, credentials=None,
     return (sam_db, res[0])
 
 
+def connect_samdb_env(env_url, env_username, env_password, lp=None):
+    """Connect to SamDB by getting URL and Credentials from environment
+
+    :param env_url: Environment variable name to get lsb url from
+    :param env_username: Username environment variable
+    :param env_password: Password environment variable
+    :return: sam_db_connection
+    """
+    samdb_url = env_get_var_value(env_url)
+    creds = credentials.Credentials()
+    if lp is None:
+        # guess Credentials parameters here. Otherwise workstation
+        # and domain fields are NULL and gencache code segfalts
+        lp = param.LoadParm()
+        creds.guess(lp)
+    creds.set_username(env_get_var_value(env_username))
+    creds.set_password(env_get_var_value(env_password))
+    return connect_samdb(samdb_url, credentials=creds, lp=lp)
+
+
 def delete_force(samdb, dn):
     try:
         samdb.delete(dn)