Menu
▾
▴
[Omc-commit] SF.net SVN: omc:[1095] cmpi-bindings/trunk/test/python/UpcallAtomTest.py
|
From: <np...@us...> - 2008-10-14 20:49:39
|
Revision: 1095
http://omc.svn.sourceforge.net/omc/?rev=1095&view=rev
Author: npaxton
Date: 2008-10-14 20:49:31 +0000 (Tue, 14 Oct 2008)
Log Message:
-----------
pull indication helper functions outside of class for independent use
Modified Paths:
--------------
cmpi-bindings/trunk/test/python/UpcallAtomTest.py
Modified: cmpi-bindings/trunk/test/python/UpcallAtomTest.py
===================================================================
--- cmpi-bindings/trunk/test/python/UpcallAtomTest.py 2008-10-14 20:29:02 UTC (rev 1094)
+++ cmpi-bindings/trunk/test/python/UpcallAtomTest.py 2008-10-14 20:49:31 UTC (rev 1095)
@@ -15,89 +15,91 @@
_g_opts = None
_g_args = None
-################################################################################
-class UpcallAtomTest(unittest.TestCase):
- global conn
+# start indication support methods
+def createFilter( ch, query='select * from CIM_ProcessIndication',
+ ns='root/interop',
+ querylang='WQL',
+ src_ns='root/cimv2',
+ in_name=None):
+ name = in_name or 'cimfilter%s'%time.time()
+ filterinst=pywbem.CIMInstance('CIM_IndicationFilter')
+ filterinst['CreationClassName']='CIM_IndicationFilter'
+ filterinst['SystemCreationClassName']='CIM_ComputerSystem'
+ filterinst['SystemName']=getfqdn()
+ filterinst['Name']=name
+ filterinst['Query']=query
+ filterinst['QueryLanguage']=querylang
+ filterinst['SourceNamespace']=src_ns
+ cop = pywbem.CIMInstanceName('CIM_IndicationFilter')
+ cop.keybindings = { 'CreationClassName':'CIM_IndicationFilter',
+ 'SystemClassName':'CIM_ComputerSystem',
+ 'SystemName':getfqdn(),
+ 'Name':name }
+ cop.namespace=ns
+ filterinst.path = cop
+ filtercop = ch.CreateInstance(filterinst)
+ return filtercop
- # start indication support methods
- def _createFilter(self,
- ch, query='select * from CIM_ProcessIndication',
- ns='root/interop',
- querylang='WQL',
- src_ns='root/cimv2',
- in_name=None):
- name = in_name or 'cimfilter%s'%time.time()
- filterinst=pywbem.CIMInstance('CIM_IndicationFilter')
- filterinst['CreationClassName']='CIM_IndicationFilter'
- filterinst['SystemCreationClassName']='CIM_ComputerSystem'
- filterinst['SystemName']=getfqdn()
- filterinst['Name']=name
- filterinst['Query']=query
- filterinst['QueryLanguage']=querylang
- filterinst['SourceNamespace']=src_ns
- cop = pywbem.CIMInstanceName('CIM_IndicationFilter')
- cop.keybindings = { 'CreationClassName':'CIM_IndicationFilter',
- 'SystemClassName':'CIM_ComputerSystem',
- 'SystemName':getfqdn(),
- 'Name':name }
- cop.namespace=ns
- filterinst.path = cop
- filtercop = ch.CreateInstance(filterinst)
- return filtercop
+def createDest( ch, destination='http://localhost:5988',
+ ns='root/interop',
+ in_name=None):
+ name = in_name or 'cimlistener%s'%time.time()
+ destinst=pywbem.CIMInstance('CIM_ListenerDestinationCIMXML')
+ destinst['CreationClassName']='CIM_ListenerDestinationCIMXML'
+ destinst['SystemCreationClassName']='CIM_ComputerSystem'
+ destinst['SystemName']=getfqdn()
+ destinst['Name']=name
+ destinst['Destination']=destination
+ cop = pywbem.CIMInstanceName('CIM_ListenerDestinationCIMXML')
+ cop.keybindings = { 'CreationClassName':'CIM_ListenerDestinationCIMXML',
+ 'SystemClassName':'CIM_ComputerSystem',
+ 'SystemName':getfqdn(),
+ 'Name':name }
+ cop.namespace=ns
+ destinst.path = cop
+ destcop = ch.CreateInstance(destinst)
+ return destcop
- def _createDest(self,
- ch, destination='http://localhost:5988',
- ns='root/interop',
- in_name=None):
- name = in_name or 'cimlistener%s'%time.time()
- destinst=pywbem.CIMInstance('CIM_ListenerDestinationCIMXML')
- destinst['CreationClassName']='CIM_ListenerDestinationCIMXML'
- destinst['SystemCreationClassName']='CIM_ComputerSystem'
- destinst['SystemName']=getfqdn()
- destinst['Name']=name
- destinst['Destination']=destination
- cop = pywbem.CIMInstanceName('CIM_ListenerDestinationCIMXML')
- cop.keybindings = { 'CreationClassName':'CIM_ListenerDestinationCIMXML',
- 'SystemClassName':'CIM_ComputerSystem',
- 'SystemName':getfqdn(),
- 'Name':name }
- cop.namespace=ns
- destinst.path = cop
- destcop = ch.CreateInstance(destinst)
- return destcop
+def createSubscription(ch, ns='root/interop'):
+ replace_ns = ch.default_namespace
+ ch.default_namespace=ns
+ indfilter=createFilter(ch)
+ indhandler=createDest(ch)
+ subinst=pywbem.CIMInstance('CIM_IndicationSubscription')
+ subinst['Filter']=indfilter
+ subinst['Handler']=indhandler
+ cop = pywbem.CIMInstanceName('CIM_IndicationSubscription')
+ cop.keybindings = { 'Filter':indfilter,
+ 'Handler':indhandler }
+ cop.namespace=ns
+ subinst.path = cop
+ subcop = ch.CreateInstance(subinst)
+ ch.default_namespace=replace_ns
+ return subcop
- def _createSubscription(self, ch, ns='root/interop'):
- replace_ns = ch.default_namespace
- ch.default_namespace=ns
- indfilter=self._createFilter(ch)
- indhandler=self._createDest(ch)
- subinst=pywbem.CIMInstance('CIM_IndicationSubscription')
- subinst['Filter']=indfilter
- subinst['Handler']=indhandler
- cop = pywbem.CIMInstanceName('CIM_IndicationSubscription')
- cop.keybindings = { 'Filter':indfilter,
- 'Handler':indhandler }
- cop.namespace=ns
- subinst.path = cop
- subcop = ch.CreateInstance(subinst)
- ch.default_namespace=replace_ns
- return subcop
+def deleteSubscription(ch, subcop):
+ indfilter = subcop['Filter']
+ indhandler= subcop['Handler']
+ ch.DeleteInstance(subcop)
+ ch.DeleteInstance(indfilter)
+ ch.DeleteInstance(indhandler)
- def _deleteSubscription(self, ch, subcop):
- indfilter = subcop['Filter']
- indhandler= subcop['Handler']
- ch.DeleteInstance(subcop)
- ch.DeleteInstance(indfilter)
- ch.DeleteInstance(indhandler)
+# end indication support methods
- # end indication support methods
+
+################################################################################
+class UpcallAtomTest(unittest.TestCase):
+
+ global conn
+
+
def setUp(self):
unittest.TestCase.setUp(self)
self.conn = conn
@@ -116,20 +118,22 @@
print('\t -- %s --' % msg)
else:
print('')
-
+
+ '''
def test_a_upcalls_all(self):
rv,outs = self.conn.InvokeMethod('test_all_upcalls', 'Test_UpcallAtom')
self.assertEquals(rv, 'Success!')
self.assertFalse(outs)
+ '''
def test_b_indications(self):
numrcv = 0
- self.subcop = self._createSubscription(self.conn)
+ self.subcop = createSubscription(self.conn)
num_to_send = pywbem.Uint16(7)
self.conn.InvokeMethod('reset_indication_count', 'Test_UpcallAtom')
countsent,outs = self.conn.InvokeMethod('send_indications', 'Test_UpcallAtom', num_to_send=num_to_send)
numsent,outs = self.conn.InvokeMethod('get_indication_send_count', 'Test_UpcallAtom')
- self._deleteSubscription(self.conn, self.subcop)
+ deleteSubscription(self.conn, self.subcop)
if (countsent != numsent):
self.fail("send_indications NumSent(%d) doesn't match get_indication_send_count NumSent(%d)"%(countsent, numsent));
if (numrcv != numsent):
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
