linse/sics
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Push test_suite, sicsconfig and hipadaba down to base class setup

Browse Source
This commit is contained in:
Douglas Clowes
2014-05-21 12:27:22 +10:00
parent 9c08424524
commit 70e1cdcd01
1 changed files with 79 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

146
site_ansto/instrument/TEST_SICS/unit_tests/sics_test.py
View File

@ -15,7 +15,6 @@ from twisted.internet import reactor, protocol, defer
import ConfigParser, StringIO import ConfigParser, StringIO
from datetime import datetime as dt from datetime import datetime as dt
class Able(unittest.TestCase): class Able(unittest.TestCase):
def setUp(self): def setUp(self):
#print "Connecting ...", #print "Connecting ...",
@ -24,7 +23,17 @@ class Able(unittest.TestCase):
self.deferred = defer.Deferred() self.deferred = defer.Deferred()
def cb_login(client, *args, **kw): def cb_login(client, *args, **kw):
#print "cb_login:", client, args, kw #print "cb_login:", client, args, kw
self.deferred.callback(True) if 'test_suite' not in globals():
d = self.load_test_suite()
d.addCallback(cb_login)
elif 'sicsconfig' not in globals():
d = self.load_sicsconfig()
d.addCallback(cb_login)
elif 'hipadaba' not in globals():
d = self.load_hipadaba()
d.addCallback(cb_login)
else:
self.deferred.callback(True)
def cb_connect(client, *args, **kw): def cb_connect(client, *args, **kw):
#print "cb_connect:", client, client.login, args, kw #print "cb_connect:", client, client.login, args, kw
self.sics = client self.sics = client
@ -41,8 +50,44 @@ class Able(unittest.TestCase):
#print "Command:", command, d #print "Command:", command, d
return d return d
def load_test_suite(self):
debug = False
def cb1(result, *args, **kw):
global test_suite
if debug:
print "fileeval:", result, args, kw
if result[0] != "OK":
raise Exception("fileeval returned " + repr(result))
test_suite = True
d = self.send_command("fileeval TEST_SICS/unit_tests/test_suite.tcl")
d.addCallback(cb1)
return d
def load_sicsconfig(self):
debug = False
def cb2(result, *args, **kw):
global sicsconfig
with open("sicsconfig.ini", "w") as outfile:
for line in result:
outfile.write(line + '\n')
if debug:
print "config:", result, args, kw
ini = StringIO.StringIO("\n".join(result))
if debug:
print "ini:", ini.readlines()
ini.seek(0)
print "ini:", ini.read()
ini.seek(0)
print "ini:", dir(ini)
sicsconfig = ConfigParser.SafeConfigParser()
sicsconfig.readfp(ini)
d = self.send_command("tcl:test_suite::show_config ::config_dict")
d.addCallback(cb2)
return d
def load_hipadaba(self): def load_hipadaba(self):
def cb0(result, *args, **kw): def cb0(result, *args, **kw):
global hipadaba
with open("hipadaba.xml", "w") as outfile: with open("hipadaba.xml", "w") as outfile:
for line in result: for line in result:
outfile.write(line + '\n') outfile.write(line + '\n')
@ -51,11 +96,10 @@ class Able(unittest.TestCase):
if result[-1] != '</hipadaba:SICS>': if result[-1] != '</hipadaba:SICS>':
raise Exception("Tree does not end with </hipadaba:SICS>:" + result[-1]) raise Exception("Tree does not end with </hipadaba:SICS>:" + result[-1])
self.hipadaba = HipadabaTree('\n'.join(result)) self.hipadaba = HipadabaTree('\n'.join(result))
self.tree_deferred.callback(self.hipadaba) hipadaba = self.hipadaba
self.tree_deferred = defer.Deferred()
d = self.send_command("tcl:test_suite::getsicsxml /") d = self.send_command("tcl:test_suite::getsicsxml /")
d.addCallback(cb0) d.addCallback(cb0)
return self.tree_deferred return d
def tearDown(self): def tearDown(self):
#print "tearDownModule" #print "tearDownModule"
@ -70,60 +114,34 @@ class Baker(Able):
timeout = 5 timeout = 5
def test_000_000_login(self): def test_000_000_login(self):
debug = False debug = False
self.deferred = defer.Deferred()
def cb0(result, *args, **kw): def cb0(result, *args, **kw):
if debug: if debug:
print "Login:", result, args, kw print "Login:", result, args, kw
if self.sics.login != 2: if self.sics.login != 2:
raise Exception("Bad login:" + repr(self.sics.login)) raise Exception("Bad login:" + repr(self.sics.login))
self.deferred.callback(None)
d = self.sics.login_deferred d = self.sics.login_deferred
d.addCallback(cb0) d.addCallback(cb0)
return self.deferred return d
def test_000_001_fileeval(self): def test_000_001_fileeval(self):
debug = False self.assertTrue('test_suite' in globals())
self.deferred = defer.Deferred() self.assertTrue(test_suite)
def cb1(result, *args, **kw):
if debug:
print "fileeval:", result, args, kw
if result[0] != "OK":
raise Exception("fileeval returned " + repr(result))
self.deferred.callback(None)
d = self.send_command("fileeval TEST_SICS/unit_tests/test_suite.tcl")
d.addCallback(cb1)
return self.deferred
def test_001_000_config(self): def test_000_002_config(self):
debug = False debug = False
self.deferred = defer.Deferred() self.assertTrue('sicsconfig' in globals())
def cb2(result, *args, **kw): if debug:
global config print "config:", repr(sicsconfig)
if debug: print "config:", dir(sicsconfig)
print "config:", result, args, kw for section in sorted(sicsconfig.sections()):
ini = StringIO.StringIO("\n".join(result)) print ("[%s]\n" % section),
if debug: for option in sorted(sicsconfig.options(section)):
print "ini:", ini.readlines() print ("%s = %s\n" % (option, sicsconfig.get(section, option))),
ini.seek(0) print
print "ini:", ini.read()
ini.seek(0) def test_000_003_hipadaba(self):
print "ini:", dir(ini) debug = False
config = ConfigParser.SafeConfigParser() self.assertTrue('hipadaba' in globals())
if debug:
print "config:", repr(config)
config.readfp(ini)
if debug:
print "config:", repr(config)
print "config:", dir(config)
for section in sorted(config.sections()):
print ("[%s]\n" % section),
for option in sorted(config.options(section)):
print ("%s = %s\n" % (option, config.get(section, option))),
print
self.deferred.callback(None)
d = self.send_command("tcl:test_suite::show_config ::config_dict")
d.addCallback(cb2)
return self.deferred
def test_001_001_clock(self): def test_001_001_clock(self):
debug = False debug = False
@ -148,14 +166,8 @@ class Baker(Able):
return self.deferred return self.deferred
def test_001_002_hipadaba(self): def test_001_002_hipadaba(self):
self.deferred = defer.Deferred() debug = False
def cb3(result, *args, **kw): self.assertTrue('hipadaba' in globals())
global hipadaba
hipadaba = result
self.deferred.callback(None)
d = self.load_hipadaba()
d.addCallback(cb3)
return self.deferred
def test_001_003_status(self): def test_001_003_status(self):
d = self.send_command("status") d = self.send_command("status")
@ -252,8 +264,8 @@ class Baker(Able):
self.assertTrue('control' in id_list) self.assertTrue('control' in id_list)
def test_004_002_gumtreexml(self): def test_004_002_gumtreexml(self):
global config, hipadaba global sicsconfig
if not ('pfeiffer_hg' in config.sections() and config.get("pfeiffer_hg", "enabled") == "True"): if not ('pfeiffer_hg' in sicsconfig.sections() and sicsconfig.get("pfeiffer_hg", "enabled") == "True"):
raise unittest.SkipTest("Pfeiffer PS9 is not configured") raise unittest.SkipTest("Pfeiffer PS9 is not configured")
child = gumtreexml child = gumtreexml
for node in "/sample/ps9/".strip('/').split('/'): for node in "/sample/ps9/".strip('/').split('/'):
@ -369,11 +381,11 @@ class Baker(Able):
if not result[0].startswith("m2 = "): if not result[0].startswith("m2 = "):
raise Exception("Unexpected m2 response: " + repr(result)) raise Exception("Unexpected m2 response: " + repr(result))
self.m2 = float(result[0].split("=")[1]) self.m2 = float(result[0].split("=")[1])
if self.m1 > 20: if self.m1 > 20.1:
self.m1 = round(self.m1 - 1, 0) self.m1 = round(self.m1 - 1, 0)
else: else:
self.m1 = round(self.m1 + 1, 0) self.m1 = round(self.m1 + 1, 0)
if self.m2 > 20: if self.m2 > 20.1:
self.m2 = round(self.m2 - 1, 0) self.m2 = round(self.m2 - 1, 0)
else: else:
self.m2 = round(self.m2 + 1, 0) self.m2 = round(self.m2 + 1, 0)
@ -389,20 +401,20 @@ class Baker(Able):
d = self.send_command("m1") d = self.send_command("m1")
d.addCallback(cb1) d.addCallback(cb1)
return self.deferred return self.deferred
test_006_001_drive_one.timeout = 20 test_006_001_drive_one.timeout = 30
def Test_007_000_mercury(self): def Test_007_000_mercury(self):
raise unittest.SkipTest("Not doing this test now (borken)") raise unittest.SkipTest("Not doing this test now (borken)")
global config, hipadaba global sicsconfig
debug = False debug = False
def cb0(result, *args, **kw): def cb0(result, *args, **kw):
if debug: if debug:
print "\nMercury:", result.short_tree() print "\nMercury:", result.short_tree()
self.deferred.callback(result) self.deferred.callback(result)
if not ("mercury_scpi" in config.sections() and config.get("mercury_scpi", "enabled") == "True"): if not ("mercury_scpi" in sicsconfig.sections() and sicsconfig.get("mercury_scpi", "enabled") == "True"):
raise unittest.SkipTest("mercury_scpi is not configured") raise unittest.SkipTest("mercury_scpi is not configured")
self.deferred = defer.Deferred() self.deferred = defer.Deferred()
d = self.load_hipadaba("/sample/%s" % config.get("mercury_scpi", "name")) d = self.load_hipadaba("/sample/%s" % sicsconfig.get("mercury_scpi", "name"))
d.addCallback(cb0) d.addCallback(cb0)
return self.deferred return self.deferred
Test_007_000_mercury.timeout = 2 Test_007_000_mercury.timeout = 2
@ -412,9 +424,9 @@ class Baker(Able):
""" """
global hipadaba global hipadaba
debug = False debug = False
if not 'hipadaba' in globals() or hipadaba is None: self.assertTrue('hipadaba' in globals())
raise unittest.SkipTest("hipadaba is missing") self.assertFalse(hipadaba is None, "hipadaba is missing")
if not ('mercury_scpi' in config.sections() and config.get("mercury_scpi", "enabled") == "True"): if not ('mercury_scpi' in sicsconfig.sections() and sicsconfig.get("mercury_scpi", "enabled") == "True"):
raise unittest.SkipTest("Mercury TC9 is not configured") raise unittest.SkipTest("Mercury TC9 is not configured")
if False: if False: