[conduit: 124/138] Shuffle things around so environment loader fires before any code to be tested is imported



commit c9e7da4b9b61a4be48409bd26a485a7d5bc71812
Author: John Carr <john carr unrouted co uk>
Date:   Wed May 6 13:22:13 2009 -0700

    Shuffle things around so environment loader fires before any code to be tested is imported
---
 test/soup/__init__.py             |  229 +------------------------------------
 test/soup/modules/google.py       |    2 +-
 test/soup/modules/ipod.py         |    2 +-
 test/soup/soup                    |    2 +-
 test/soup/test_dataprovider.py    |    2 +-
 test/soup/test_datatypes.py       |    2 +-
 test/soup/test_synchronization.py |    2 +-
 test/soup/utils/test.py           |  229 ++++++++++++++++++++++++++++++++++++-
 8 files changed, 234 insertions(+), 236 deletions(-)

diff --git a/test/soup/__init__.py b/test/soup/__init__.py
index d7c54c4..6a87326 100644
--- a/test/soup/__init__.py
+++ b/test/soup/__init__.py
@@ -12,236 +12,9 @@ def get_root():
         parts.pop()
     raise NotImplementedError(get_root)
 
-sys.path.insert(0, get_root())
-
-import env
-import data
-import modules
-
-import conduit
-import conduit.utils as Utils
-import conduit.MappingDB as MappingDB
-import conduit.Module as Module
-import conduit.TypeConverter as TypeConverter
-import conduit.Synchronization as Synchronization
-import conduit.ModuleWrapper as ModuleWrapper
-import conduit.Conduit as Conduit
-import conduit.Settings as Settings
-
-conduit.IS_INSTALLED =              False
-conduit.IS_DEVELOPMENT_VERSION =    True
-conduit.SHARED_DATA_DIR =           os.path.join(get_root(),"data")
-conduit.SHARED_MODULE_DIR =         os.path.join(get_root(),"conduit","modules")
-conduit.FILE_IMPL =                 os.environ.get("CONDUIT_FILE_IMPL","GIO")
-conduit.BROWSER_IMPL =              os.environ.get("CONDUIT_BROWSER_IMPL","system")
-conduit.SETTINGS_IMPL =             os.environ.get("CONDUIT_SETTINGS_IMPL","GConf")
-conduit.GLOBALS.settings =          Settings.Settings()
-
-CHANGE_ADD = 1
-CHANGE_REPLACE = 2
-CHANGE_DELETE = 3
-
-
 def get_module(name):
     """ This is just to avoid importing sys everywhere and i want my tests to be pretty! """
     return sys.modules[name]
 
-class TestCase(unittest.TestCase):
-
-    def __init__(self, methodName='runTest'):
-        super(TestCase, self).__init__(methodName)
-        self.testMethodName = methodName
-        self.testMethod = getattr(self, methodName)
-
-    @classmethod
-    def name(cls):
-        """ Returns the name of the class. We need to override this on generated classes """
-        return cls.__name__
-
-    def id(self):
-        """ Returns the name of the class and the test this particular instance will run """
-        return self.name() + "." + self.testMethodName
-
-    def requires(self):
-        """ Yields feature objects that we depend on to run this test, such as an internet connection or python-gpod """
-        return []
-
-    def shortDescription(self):
-        """ Describe the test that is currently running
-            Returns something like TestClass.test_function: Tests how good Conduit is """
-        return "%s.%s: %s" % (self.name(), self.testMethodName, super(TestCase, self).shortDescription())
-
-    def setUp(self):
-        for feature in self.requires():
-            feature.require()
-
-    def setUpSync(self):
-        #Set up our own mapping DB so we dont pollute the global one
-        dbFile = os.path.join(os.environ['TEST_DIRECTORY'],Utils.random_string()+".db")
-        conduit.GLOBALS.mappingDB = MappingDB.MappingDB(dbFile)
-
-        self.modules = Module.ModuleManager([])
-        conduit.GLOBALS.moduleManager = self.modules
-        self.modules.load_all(whitelist=None, blacklist=None)
-
-        self.type_converter = conduit.TypeConverter.TypeConverter(self.modules)
-        conduit.GLOBALS.typeConverter = self.type_converter
-        self.sync_manager = conduit.Synchronization.SyncManager(self.type_converter)
-        conduit.GLOBALS.syncManager = self.sync_manager
-
-    def get_dataprovider(self, name):
-        wrapper = None
-        for dp in self.modules.get_all_modules():
-            if dp.classname == name:
-                wrapper = self.modules.get_module_wrapper_with_instance(dp.get_key())
-        assert wrapper != None
-        return wrapper
-
-    def get_dataprovider_factory(self, className, die=True):
-        factory = None
-        for f in self.model.dataproviderFactories:
-            if f.__class__.__name__ == className:
-                factory = f
-        assert factory != None
-        return factory
-
-    def wrap_dataprovider(self, dp):
-        wrapper = ModuleWrapper.ModuleWrapper(
-                         klass=dp.__class__,
-                         initargs=(),
-                         category=None
-                         )
-        wrapper.module = dp
-        return wrapper
-
-    def networked_dataprovider(self, dp):
-        """
-        Dirty evil cludge so we can test networked sync...
-        """
-        factory = self.get_dataprovider_factory("NetworkServerFactory")
-        server = factory.share_dataprovider(dp)
-        assert server != None
-
-        conduit = Conduit.Conduit(self.sync_manager)
-        time.sleep(1)
-
-        factory = self.get_dataprovider_factory("NetworkClientFactory")
-        newdp = factory.dataprovider_create("http://localhost";, conduit.uid, server.get_info())
-        assert newdp != None
-        return self.wrap_dataprovider( newdp() )
-
-    def create_conduit(self):
-        return Conduit.Conduit(self.sync_manager)
-
-    def create_syncset(self):
-        return SyncSet.SyncSet(
-            moduleManager=self.modules,
-            syncManager=self.sync_manager
-        )
-
-
-#
-# Custom exceptions
-#
-
-class TestSkipped(Exception):
-    """ Indicate a test was intentionally skipped rather than failed """
-
-
-class UnavailableFeature(Exception):
-    """ A feature required for this test was unavailable """
-
-
-#
-# 'Features'
-# Some tests need things that might not be available, like python gpod, so provide an interface
-# to fail gracefully.
-#
-
-class Feature(object):
-
-    def __init__(self):
-        self._cached = None
-
-    def probe(self):
-        raise NotImplementedError
-
-    def available(self):
-        if self._cached == None:
-            self._cached = self.probe()
-        return self._cached
-
-    def require(self):
-        if not self.available():
-            raise UnavailableFeature(self)
-
-    @classmethod
-    def name(cls):
-        return cls.__name__
-
-    def __str__(self):
-        return self.name()
-
-
-class _HumanInteractivity(Feature):
-
-    def probe(self):
-        try:
-            return os.environ["CONDUIT_INTERACTIVE"] == "TRUE"
-        except:
-            return False
-
-HumanInteractivity = _HumanInteractivity()
-
-
-class _Online(Feature):
-
-    def probe(self):
-        try:
-            return os.environ["CONDUIT_ONLINE"] == "TRUE"
-        except:
-            return False
-
-Online = _Online()
-
-
-#
-# Custom test loader
-#
-
-class TestLoader(unittest.TestLoader):
-
-    def __init__(self, include=None, exclude=None):
-        self.include = include or []
-        self.exclude = exclude or []
-
-    def _flatten(self, tests):
-        if isinstance(tests, unittest.TestSuite):
-            for test in tests:
-                for subtest in self._flatten(test):
-                    yield subtest
-        else:
-            yield tests
-
-    def loadTestsFromModule(self, module):
-        for test in self._flatten(super(TestLoader, self).loadTestsFromModule(module)):
-            if len(self.include) > 0:
-                is_included = False
-                for i in self.include:
-                    if i in test.id():
-                        is_included = True
-                if not is_included:
-                    continue
-            if len(self.exclude) > 0:
-                is_excluded = False
-                for x in self.exclude:
-                    if x in test.id():
-                        is_excluded = True
-                if is_excluded:
-                    continue
-            yield test
-
-    def loadTestsFromMain(self):
-        """ Load all tests that can be found starting from __main__ """
-        return self.loadTestsFromModule(__import__('__main__'))
+sys.path.insert(0, get_root())
 
diff --git a/test/soup/modules/google.py b/test/soup/modules/google.py
index 208e8e6..575f4f7 100644
--- a/test/soup/modules/google.py
+++ b/test/soup/modules/google.py
@@ -2,7 +2,7 @@
 import soup
 import soup.modules
 
-from soup import Online
+from soup.utils.test import Online
 from soup.data.contact import ContactWrapper
 from soup.data.event import EventWrapper
 
diff --git a/test/soup/modules/ipod.py b/test/soup/modules/ipod.py
index 54b4985..fd4d5f2 100644
--- a/test/soup/modules/ipod.py
+++ b/test/soup/modules/ipod.py
@@ -15,7 +15,7 @@ import uuid
 import shutil
 
 
-class _GpodModule(soup.Feature):
+class _GpodModule(soup.utils.test.Feature):
 
     def probe(self):
         try:
diff --git a/test/soup/soup b/test/soup/soup
index a5f2b95..4e97be7 100755
--- a/test/soup/soup
+++ b/test/soup/soup
@@ -70,7 +70,7 @@ from test_dataprovider import *
 from test_synchronization import *
 
 # Figure out which tests to run
-tests = soup.TestLoader(include=args, exclude=opts.exclude).loadTestsFromMain()
+tests = test.TestLoader(include=args, exclude=opts.exclude).loadTestsFromMain()
 
 if opts.randomize:
    import random
diff --git a/test/soup/test_dataprovider.py b/test/soup/test_dataprovider.py
index 0ca4d38..c9ad29d 100644
--- a/test/soup/test_dataprovider.py
+++ b/test/soup/test_dataprovider.py
@@ -2,7 +2,7 @@ import soup
 from soup.modules import ModuleLoader
 
 def make_testcase(wrp):
-    class TestDataprovider(soup.TestCase):
+    class TestDataprovider(soup.utils.test.TestCase):
         wrapperclass = wrp
 
         @classmethod
diff --git a/test/soup/test_datatypes.py b/test/soup/test_datatypes.py
index 3a8410f..e31bf51 100644
--- a/test/soup/test_datatypes.py
+++ b/test/soup/test_datatypes.py
@@ -4,7 +4,7 @@ from soup.data import DataLoader
 import pickle
 
 def make_testcase(wrp):
-    class TestDatatype(soup.TestCase):
+    class TestDatatype(soup.utils.test.TestCase):
         wrapperclass = wrp
 
         @classmethod
diff --git a/test/soup/test_synchronization.py b/test/soup/test_synchronization.py
index d992184..88952a5 100644
--- a/test/soup/test_synchronization.py
+++ b/test/soup/test_synchronization.py
@@ -5,7 +5,7 @@ from soup.modules import ModuleLoader
 import conduit
 
 def make_testcase(src, src_data, snk, snk_data):
-    class TestSynchronization(soup.TestCase):
+    class TestSynchronization(soup.utils.test.TestCase):
         source_class = src
         source_data_class = src_data
         sink_class = snk
diff --git a/test/soup/utils/test.py b/test/soup/utils/test.py
index ccebbb2..dec2200 100644
--- a/test/soup/utils/test.py
+++ b/test/soup/utils/test.py
@@ -1,6 +1,5 @@
 
-import soup.env
-from soup import UnavailableFeature
+import soup
 from soup.utils import progressbar
 
 import sys
@@ -9,6 +8,232 @@ import cgitb
 import traceback
 import time
 
+# FIXME: These only need to be here because i'm mixing unittest tweaks with conduit helpers
+import os
+import conduit
+import conduit.utils as Utils
+import conduit.MappingDB as MappingDB
+import conduit.Module as Module
+import conduit.TypeConverter as TypeConverter
+import conduit.Synchronization as Synchronization
+import conduit.ModuleWrapper as ModuleWrapper
+import conduit.Conduit as Conduit
+import conduit.Settings as Settings
+
+CHANGE_ADD = 1
+CHANGE_REPLACE = 2
+CHANGE_DELETE = 3
+
+
+class TestCase(unittest.TestCase):
+
+    def __init__(self, methodName='runTest'):
+        super(TestCase, self).__init__(methodName)
+        self.testMethodName = methodName
+        self.testMethod = getattr(self, methodName)
+
+    @classmethod
+    def name(cls):
+        """ Returns the name of the class. We need to override this on generated classes """
+        return cls.__name__
+
+    def id(self):
+        """ Returns the name of the class and the test this particular instance will run """
+        return self.name() + "." + self.testMethodName
+
+    def requires(self):
+        """ Yields feature objects that we depend on to run this test, such as an internet connection or python-gpod """
+        return []
+
+    def shortDescription(self):
+        """ Describe the test that is currently running
+            Returns something like TestClass.test_function: Tests how good Conduit is """
+        return "%s.%s: %s" % (self.name(), self.testMethodName, super(TestCase, self).shortDescription())
+
+    def setUp(self):
+        for feature in self.requires():
+            feature.require()
+
+    def setUpSync(self):
+	# FIXME: I'd put this in an EnvironmentWrapper, but i need priorities before i can do that :/
+	conduit.IS_INSTALLED =              False
+	conduit.IS_DEVELOPMENT_VERSION =    True
+	conduit.SHARED_DATA_DIR =           os.path.join(soup.get_root(),"data")
+	conduit.SHARED_MODULE_DIR =         os.path.join(soup.get_root(),"conduit","modules")
+	conduit.FILE_IMPL =                 os.environ.get("CONDUIT_FILE_IMPL","GIO")
+	conduit.BROWSER_IMPL =              os.environ.get("CONDUIT_BROWSER_IMPL","system")
+	conduit.SETTINGS_IMPL =             os.environ.get("CONDUIT_SETTINGS_IMPL","GConf")
+	conduit.GLOBALS.settings =          Settings.Settings()
+
+        #Set up our own mapping DB so we dont pollute the global one
+        dbFile = os.path.join(os.environ['TEST_DIRECTORY'],Utils.random_string()+".db")
+        conduit.GLOBALS.mappingDB = MappingDB.MappingDB(dbFile)
+
+        self.modules = Module.ModuleManager([])
+        conduit.GLOBALS.moduleManager = self.modules
+        self.modules.load_all(whitelist=None, blacklist=None)
+
+        self.type_converter = conduit.TypeConverter.TypeConverter(self.modules)
+        conduit.GLOBALS.typeConverter = self.type_converter
+        self.sync_manager = conduit.Synchronization.SyncManager(self.type_converter)
+        conduit.GLOBALS.syncManager = self.sync_manager
+
+    def get_dataprovider(self, name):
+        wrapper = None
+        for dp in self.modules.get_all_modules():
+            if dp.classname == name:
+                wrapper = self.modules.get_module_wrapper_with_instance(dp.get_key())
+        assert wrapper != None
+        return wrapper
+
+    def get_dataprovider_factory(self, className, die=True):
+        factory = None
+        for f in self.model.dataproviderFactories:
+            if f.__class__.__name__ == className:
+                factory = f
+        assert factory != None
+        return factory
+
+    def wrap_dataprovider(self, dp):
+        wrapper = ModuleWrapper.ModuleWrapper(
+                         klass=dp.__class__,
+                         initargs=(),
+                         category=None
+                         )
+        wrapper.module = dp
+        return wrapper
+
+    def networked_dataprovider(self, dp):
+        """
+        Dirty evil cludge so we can test networked sync...
+        """
+        factory = self.get_dataprovider_factory("NetworkServerFactory")
+        server = factory.share_dataprovider(dp)
+        assert server != None
+
+        conduit = Conduit.Conduit(self.sync_manager)
+        time.sleep(1)
+
+        factory = self.get_dataprovider_factory("NetworkClientFactory")
+        newdp = factory.dataprovider_create("http://localhost";, conduit.uid, server.get_info())
+        assert newdp != None
+        return self.wrap_dataprovider( newdp() )
+
+    def create_conduit(self):
+        return Conduit.Conduit(self.sync_manager)
+
+    def create_syncset(self):
+        return SyncSet.SyncSet(
+            moduleManager=self.modules,
+            syncManager=self.sync_manager
+        )
+
+
+#
+# Custom exceptions
+#
+
+class TestSkipped(Exception):
+    """ Indicate a test was intentionally skipped rather than failed """
+
+
+class UnavailableFeature(Exception):
+    """ A feature required for this test was unavailable """
+
+
+#
+# 'Features'
+# Some tests need things that might not be available, like python gpod, so provide an interface
+# to fail gracefully.
+#
+
+class Feature(object):
+
+    def __init__(self):
+        self._cached = None
+
+    def probe(self):
+        raise NotImplementedError
+
+    def available(self):
+        if self._cached == None:
+            self._cached = self.probe()
+        return self._cached
+
+    def require(self):
+        if not self.available():
+            raise UnavailableFeature(self)
+
+    @classmethod
+    def name(cls):
+        return cls.__name__
+
+    def __str__(self):
+        return self.name()
+
+
+class _HumanInteractivity(Feature):
+
+    def probe(self):
+        try:
+            return os.environ["CONDUIT_INTERACTIVE"] == "TRUE"
+        except:
+            return False
+
+HumanInteractivity = _HumanInteractivity()
+
+
+class _Online(Feature):
+
+    def probe(self):
+        try:
+            return os.environ["CONDUIT_ONLINE"] == "TRUE"
+        except:
+            return False
+
+Online = _Online()
+
+
+#
+# Custom test loader
+#
+
+class TestLoader(unittest.TestLoader):
+
+    def __init__(self, include=None, exclude=None):
+        self.include = include or []
+        self.exclude = exclude or []
+
+    def _flatten(self, tests):
+        if isinstance(tests, unittest.TestSuite):
+            for test in tests:
+                for subtest in self._flatten(test):
+                    yield subtest
+        else:
+            yield tests
+
+    def loadTestsFromModule(self, module):
+        for test in self._flatten(super(TestLoader, self).loadTestsFromModule(module)):
+            if len(self.include) > 0:
+                is_included = False
+                for i in self.include:
+                    if i in test.id():
+                        is_included = True
+                if not is_included:
+                    continue
+            if len(self.exclude) > 0:
+                is_excluded = False
+                for x in self.exclude:
+                    if x in test.id():
+                        is_excluded = True
+                if is_excluded:
+                    continue
+            yield test
+
+    def loadTestsFromMain(self):
+        """ Load all tests that can be found starting from __main__ """
+        return self.loadTestsFromModule(__import__('__main__'))
+
 
 class TestResult(unittest.TestResult):
 



[Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]