5:10 p.m.
Deepti B. Kalakeri wrote:
# HG changeset patch
# User Deepti B.Kalakeri <deeptik(a)linux.vnet.ibm.com>
# Date 1247143660 25200
# Node ID 73aedcd4026381023be4d25d741ce74113b8f8e6
# Parent 7b18b827fca6916fd87ff417f79a96c8a19b4913
[TEST] Added the indications.py to support Indications.
Included the following important functions in indications.py
- sub_ind() --> Subscribe for the indications to be watched.
- handle_request() --> Filters the required indications.
- poll_for_ind() --> Wait for the required indications to be triggered.
Oops, I think I was unclear in my previous mail. I think the indication
tester should stay intact... The reason we decided to copy the
indication_tester from libcmpiutil to cimtest is so that in bugs that
get fixed in one get fixed in the other. That's why they need to stay
in sync.
So, if you find a bug in filter_xml(), you want to be sure its fixed in
both places. The problem was that cimtest wasn't publicly available at
the time, and indications aren't trivial to test using standard CIM
related tools. So we wanted to offer users a way to test their
indications, which is why it exists in libcmpiutil.
We could remove the indication_tester.py from libcmpiutil and leave the
only copy in cimtest, however, I think we should have two copies of the
same code in cimtest.
So, I suggest leave indication_tester.py as is and place the functions
below in their own file. Does that seem reasonable? That way, we don't
duplicate code - if we fix a bug in one spot, we won't have to fix the
same bug somewhere else.
+def sub_ind(ip, virt, ind_names):
+ dict = set_default(ip)
+ sub_list = {}
+ port = 5
+
+ for ind, iname in ind_names.iteritems():
+ ind_name = get_typed_class(virt, iname)
+
+ sub_name = "Test%s" % ind_name
+ port += 1
+
+ sub = CIMIndicationSubscription(sub_name, ind_name,
+ dict['default_ns'],
+ dict['default_print_ind'],
+ dict['default_sysname'],
+ port)
+ sub.subscribe(dict['default_url'], dict['default_auth'])
+ logger.info("Watching for %s", iname)
+ ind_names[ind] = ind_name
+ sub_list[ind] = sub
+
+ return sub_list, ind_names, dict
+
+def handle_request(sub, ind_name, dict, exp_ind_ct):
+ #sfcb delivers indications to all registrations, even if the indication
+ #isn't what the registration was subscribed to. So, for modified and
+ #deleted indications, we must loop through until the indication we are
+ #looking for is triggered.
+ for i in range(0, exp_ind_ct):
+ sub.server.handle_request()
+ if len(sub.server.indications) < 1:
+ logger.error("No valid indications received")
+ return FAIL
+
+ if str(sub.server.indications[0]) == ind_name:
+ sub.unsubscribe(dict['default_auth'])
+ logger.info("Cancelling subscription for %s", ind_name)
+ return PASS
+ else:
+ sub.server.indications.remove(sub.server.indications[0])
+
+ logger.error("Did not recieve indication %s", ind_name)
+ return FAIL
+
+def poll_for_ind(pid, ind_name, timeout=20):
+ status = FAIL
+ for i in range(0, timeout):
+ pw = waitpid(pid, WNOHANG)
+
+ # If pid exits, waitpid returns [pid, return_code]
+ # If pid is still running, waitpid returns [0, 0]
+ # Only return a success if waitpid returns the expected pid
+ # and the return code is 0.
+ if pw[0] == pid and pw[1] == 0:
+ logger.info("Great, got '%s' indication successfully",
ind_name)
+ status = PASS
+ break
+ elif pw[1] == 0 and i < timeout:
+ if i % 10 == 0:
+ logger.info("In child, waiting for '%s' indication",
ind_name)
+ sleep(1)
+ else:
+ # Time is up and waitpid never returned the expected pid
+ if pw[0] != pid:
+ logger.error("Waited too long for '%s' indication",
ind_name)
+ kill(pid, SIGKILL)
+ else:
+ logger.error("Received Indication error: '%d'",
pw[1])
+
+ status = FAIL
+ break
+
+ return status
--
Kaitlin Rupert
IBM Linux Technology Center
kaitlin(a)linux.vnet.ibm.com