tests: use shared base class in provider tests

Done for: Account, Journald and LogicalFile

5 files changed, 15 insertions, 179 deletions

diff --git a/src/account/test/TestAccount.py b/src/account/test/TestAccount.py
index 53cb429..a66ef45 100644
--- a/src/account/test/TestAccount.py
+++ b/src/account/test/TestAccount.py
@@ -49,7 +49,7 @@ class TestAccount(AccountBase):
         # make sure the account will not exist
         clean_account(self.user_name)
         computer_system = self.wbemconnection.ExecQuery('WQL',
-            'select * from PG_ComputerSystem')[0]
+            'select * from %s' % self.system_cs_name)[0]
         lams = self.wbemconnection.ExecQuery('WQL',
             'select * from LMI_AccountManagementService')[0]
         self.wbemconnection.InvokeMethod("CreateAccount", lams.path,
diff --git a/src/account/test/TestGroup.py b/src/account/test/TestGroup.py
index 1ad104d..d0bec65 100644
--- a/src/account/test/TestGroup.py
+++ b/src/account/test/TestGroup.py
@@ -47,7 +47,7 @@ class TestGroup(AccountBase):
         # make sure the group will not exist
         clean_group(self.group_name)
         computer_system = self.wbemconnection.ExecQuery('WQL',
-            'select * from PG_ComputerSystem')[0]
+            'select * from %s' % self.system_cs_name)[0]
         lams = self.wbemconnection.ExecQuery('WQL',
             'select * from LMI_AccountManagementService')[0]
         self.wbemconnection.InvokeMethod("CreateGroup", lams.path,
diff --git a/src/account/test/TestIndications.py b/src/account/test/TestIndications.py
index 7b34b2e..5f74548 100644
--- a/src/account/test/TestIndications.py
+++ b/src/account/test/TestIndications.py
@@ -25,6 +25,8 @@ class TestIndications(AccountBase):
     """
     Class for testing LMI_Account indications
     """
+    NEEDS_INDICATIONS = True
+
     def test_check_good_filter(self):
         """
         Account: Test good indication filter
diff --git a/src/account/test/TestService.py b/src/account/test/TestService.py
index 3450ccf..1b34d69 100644
--- a/src/account/test/TestService.py
+++ b/src/account/test/TestService.py
@@ -34,7 +34,7 @@ class TestService(AccountBase):
         # make sure the account will not exist
         clean_account(self.user_name)
         computer_system = self.wbemconnection.ExecQuery('WQL',
-            'select * from PG_ComputerSystem')[0]
+            'select * from %s' % self.system_cs_name)[0]
         lams = self.wbemconnection.ExecQuery('WQL',
             'select * from LMI_AccountManagementService')[0]
 
diff --git a/src/account/test/common.py b/src/account/test/common.py
index 57975be..8e1c20d 100644
--- a/src/account/test/common.py
+++ b/src/account/test/common.py
@@ -17,188 +17,22 @@
 #
 # Authors: Roman Rakus <rrakus@redhat.com>
 #
-
-import pywbem
-import os
-import unittest
-import Queue
-import random
-import BaseHTTPServer
-import socket
-import threading
-
 """
-Base class for all tests
+Base class and utilities for all OpenLMI Account tests.
 """
 
-class CIMListener(object):
-    """ CIM Listener
-    """
-    class CIMHandler(BaseHTTPServer.BaseHTTPRequestHandler):
-        def do_POST(self):
-            data = self.rfile.read(int(self.headers['Content-Length']))
-            tt = pywbem.parse_cim(pywbem.xml_to_tupletree(data))
-            # Get the instance from CIM-XML, copied from
-            # http://sf.net/apps/mediawiki/pywbem/?title=Indications_Tutorial
-            insts = [x[1] for x in tt[2][2][0][2][2]]
-            for inst in insts:
-                self.callback(inst)
-            self.send_response(200)
-            self.end_headers()
-
-        def log_message(self, format, *p):
-            # suppress log messages
-            pass
-
-    def __init__(self, callback, http_port=5988):
-        self.address = ('', http_port)
-        self.CIMHandler.callback = callback
-        self.thread = None
-        self.server = None
-
-    def start(self):
-        BaseHTTPServer.HTTPServer.allow_reuse_address = True
-        self.server = BaseHTTPServer.HTTPServer(self.address, self.CIMHandler)
-        self.thread = threading.Thread(target=self.server.serve_forever)
-        self.thread.start()
-
-    def stop(self):
-        if self.server is not None:
-            self.server.shutdown()
-            self.server.socket.close()
-        if self.thread is not None:
-            self.thread.join()
-
-    def running(self):
-        return self.thread is not None
+import os
 
+from lmi.test import base
 
-class AccountBase(unittest.TestCase):
+class AccountBase(base.LmiTestCase):
     """
-    Base class for all LMI Account tests
+    Base class for all LMI Account tests.
     """
-    def setUp(self):
-        """
-        Connnect to server
-        """
-        self.url = os.environ.get("LMI_CIMOM_URL", "https://localhost:5989")
-        self.username = os.environ.get("LMI_CIMOM_USERNAME", "root")
-        self.password = os.environ.get("LMI_CIMOM_PASSWORD", "")
-        self.user_name = os.environ.get("LMI_ACCOUNT_USER")
-        self.group_name = os.environ.get("LMI_ACCOUNT_GROUP")
-        self.wbemconnection = pywbem.WBEMConnection(self.url,
-            (self.username, self.password))
-
-        # for indications
-        self.indication_port = random.randint(12000, 13000)
-        self.indication_queue = Queue.Queue()
-        self.listener = CIMListener(
-                callback=self._process_indication,
-                http_port=self.indication_port)
-
-        self.subscribed = {}
-
-    def tearDown(self):
-        self.listener.stop()
-        if self.subscribed:
-            for name in self.subscribed.keys():
-                self.unsubscribe(name)
-
-    def get_indication(self, timeout):
-        """ Wait for an indication for given nr. of seconds and return it."""
-        try:
-            indication = self.indication_queue.get(timeout=timeout)
-        except Queue.Empty:
-            raise AssertionError("Timeout when waiting for indicaiton")
-        self.indication_queue.task_done()
-        return indication
-
-    def subscribe(self, filter_name, query=None, querylang="DMTF:CQL"):
-        """
-        Create indication subscription for given filter name.
-        """
-        namespace = "root/interop"
-        hostname = socket.gethostname()
-
-        if query is not None:
-            # Create filter first
-            filterinst = pywbem.CIMInstance('CIM_IndicationFilter')
-            filterinst['CreationClassName'] = 'CIM_IndicationFilter'
-            filterinst['SystemCreationClassName'] = 'CIM_ComputerSystem'
-            filterinst['SystemName'] = hostname
-            filterinst['Name'] = filter_name
-            filterinst['Query'] = query
-            filterinst['QueryLanguage'] = querylang
-            filterinst['SourceNamespace'] = "root/cimv2"#namespace
-            cop = pywbem.CIMInstanceName('CIM_IndicationFilter')
-            cop.keybindings = { 'CreationClassName': 'CIM_IndicationFilter',
-                'SystemClassName': 'CIM_ComputerSystem',
-                'SystemName': hostname,
-                'Name': filter_name
-            }
-            cop.namespace=namespace
-            filterinst.path = cop
-            indfilter = self.wbemconnection.CreateInstance(filterinst)
-        else:
-            # the filter is already created, assemble its name
-            indfilter = pywbem.CIMInstanceName(
-                    classname="CIM_IndicationFilter",
-                    namespace=namespace,
-                    keybindings={
-                        'CreationClassName': 'CIM_IndicationFilter',
-                        'SystemClassName': 'CIM_ComputerSystem',
-                        'SystemName': hostname,
-                        'Name': filter_name})
-
-        # create destination
-        destinst = pywbem.CIMInstance('CIM_ListenerDestinationCIMXML')
-        destinst['CreationClassName'] = 'CIM_ListenerDestinationCIMXML'
-        destinst['SystemCreationClassName'] = 'CIM_ComputerSystem'
-        destinst['SystemName'] = hostname
-        destinst['Name'] = filter_name
-        destinst['Destination'] = "http://localhost:%d" % (self.indication_port)
-        destinst['PersistenceType'] = pywbem.Uint16(3) # Transient
-        cop = pywbem.CIMInstanceName('CIM_ListenerDestinationCIMXML')
-        cop.keybindings = { 'CreationClassName':'CIM_ListenerDestinationCIMXML',
-                'SystemClassName':'CIM_ComputerSystem',
-                'SystemName':hostname,
-                'Name':filter_name }
-        cop.namespace = namespace
-        destinst.path = cop
-        destname = self.wbemconnection.CreateInstance(destinst)
-
-        # create the subscription
-        subinst = pywbem.CIMInstance('CIM_IndicationSubscription')
-        subinst['Filter'] = indfilter
-        subinst['Handler'] = destname
-        cop = pywbem.CIMInstanceName('CIM_IndicationSubscription')
-        cop.keybindings = { 'Filter': indfilter,
-                'Handler': destname }
-        cop.namespace = namespace
-        subinst.path = cop
-        subscription = self.wbemconnection.CreateInstance(subinst)
-
-        self.subscribed[filter_name] = [subscription, destname]
-
-        # start listening
-        if not self.listener.running():
-            self._start_listening()
-        return subscription
-
-    def unsubscribe(self, filter_name):
-        """
-        Unsubscribe fron given filter.
-        """
-        _list = self.subscribed.pop(filter_name)
-        for instance in _list:
-            self.wbemconnection.DeleteInstance(instance)
-
-    def _start_listening(self):
-        """ Start listening for incoming indications. """
-        self.listener.start()
-
-    def _process_indication(self, indication):
-        """ Callback to process one indication."""
-        self.indication_queue.put(indication)
 
+    @classmethod
+    def setUpClass(cls):
+        base.LmiTestCase.setUpClass.im_func(cls)
+        cls.user_name = os.environ.get("LMI_ACCOUNT_USER")
+        cls.group_name = os.environ.get("LMI_ACCOUNT_GROUP")