Berlinium tests

Author: blokhin

tests/__init__.py

@@ -3,23 +3,27 @@
 import os, sys
 import time
 import subprocess
+import unittest
 
-if __name__ == "__main__":
- try:
- prcnum = int(sys.argv[1])
- except (IndexError, ValueError):
- sys.exit("Usage: script #processes")
 
- basedir = os.path.dirname(os.path.realpath(os.path.abspath(__file__)))
- daemon = subprocess.Popen([sys.executable, os.path.join(basedir, 'asleep_server.py')])
- time.sleep(2) # wait for initialization
+basedir = os.path.dirname(os.path.realpath(os.path.abspath(__file__)))
 
- children = []
- for i in range(prcnum):
- children.append( subprocess.Popen([sys.executable, os.path.join(basedir, 'asleep_client.py')]) )
+class Test_Async_Sleep_Server(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.daemon = subprocess.Popen([sys.executable, os.path.join(basedir, 'asleep_server.py')])
+ time.sleep(2) # wait for initialization
 
- time.sleep(prcnum + 2)
- daemon.terminate()
+ def test_response(self):
+ prcnum = 5
+ children = []
+ for i in range(prcnum):
+ children.append( subprocess.Popen([sys.executable, os.path.join(basedir, 'asleep_client.py')]) )
+ time.sleep(7)
 
- for i in children:
- assert i.poll() == 0
+ for i in children:
+ self.assertEqual(i.poll(), 0, "Server response is too slow")
+
+ @classmethod
+ def tearDownClass(cls):
+ cls.daemon.terminate()

tests/berlinium/asleep_test.py

@@ -3,23 +3,27 @@
 import os, sys
 import time
 import subprocess
+import unittest
 
-if __name__ == "__main__":
- try:
- prcnum = int(sys.argv[1])
- except (IndexError, ValueError):
- sys.exit("Usage: script #processes")
 
- basedir = os.path.dirname(os.path.realpath(os.path.abspath(__file__)))
- daemon = subprocess.Popen([sys.executable, os.path.join(basedir, 'basic_server.py')])
- time.sleep(2) # wait for initialization
+basedir = os.path.dirname(os.path.realpath(os.path.abspath(__file__)))
 
- children = []
- for i in range(prcnum):
- children.append( subprocess.Popen([sys.executable, os.path.join(basedir, 'basic_client.py')]) )
+class Test_Async_Basic_Server(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.daemon = subprocess.Popen([sys.executable, os.path.join(basedir, 'basic_server.py')])
+ time.sleep(2) # wait for initialization
 
- time.sleep(prcnum + 2)
- daemon.terminate()
+ def test_response(self):
+ prcnum = 5
+ children = []
+ for i in range(prcnum):
+ children.append( subprocess.Popen([sys.executable, os.path.join(basedir, 'basic_client.py')]) )
+ time.sleep(7)
 
- for i in children:
- assert i.poll() == 0
+ for i in children:
+ self.assertEqual(i.poll(), 0, "Server response is too slow")
+
+ @classmethod
+ def tearDownClass(cls):
+ cls.daemon.terminate()

tests/berlinium/basic_test.py

@@ -3,23 +3,27 @@
 import os, sys
 import time
 import subprocess
+import unittest
 
-if __name__ == "__main__":
- try:
- prcnum = int(sys.argv[1])
- except (IndexError, ValueError):
- sys.exit("Usage: script #processes")
 
- basedir = os.path.dirname(os.path.realpath(os.path.abspath(__file__)))
- daemon = subprocess.Popen([sys.executable, os.path.join(basedir, '../../utils/gui_server.py')])
- time.sleep(2) # wait for initialization
+basedir = os.path.dirname(os.path.realpath(os.path.abspath(__file__)))
 
- children = []
- for i in range(prcnum):
- children.append( subprocess.Popen([sys.executable, os.path.join(basedir, 'gui_client.py')]) )
+class Test_GUI_Server(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.daemon = subprocess.Popen([sys.executable, os.path.join(basedir, '../../utils/gui_server.py')])
+ time.sleep(2) # wait for initialization
 
- time.sleep(prcnum + 2)
- daemon.terminate()
+ def test_response(self):
+ prcnum = 10
+ children = []
+ for i in range(prcnum):
+ children.append( subprocess.Popen([sys.executable, os.path.join(basedir, 'gui_client.py')]) )
+ time.sleep(6)
 
- for i in children:
- assert i.poll() == 0
+ for i in children:
+ self.assertEqual(i.poll(), 0, "Server response is too slow")
+
+ @classmethod
+ def tearDownClass(cls):
+ cls.daemon.terminate()

tests/berlinium/gui_test.py

@@ -16,44 +16,92 @@
 from . import TestLayerDB
 
 
-class Test_Query_Approaches(TestLayerDB):
- __test_calcs_dir__ = EXAMPLE_DIR
-
- _work = API()
- searched_category = (80, 'plane waves') # defined in /init-data.sql
+class Hierarchy_Params(object):
+ work = API()
+ searched_category = (80, 'gaussians') # defined in /init-data.sql
  searched_category_num = None # defined in hierarchy
  space_grp_topic_source = 'ng' # defined in parsers and set in API.classify()
- _categ_found = False
- for cid, series in _work.hierarchy_values.iteritems():
- if _categ_found: break
+ categ_found = False
+ for cid, series in work.hierarchy_values.iteritems():
+ if categ_found: break
  elif cid == searched_category[0]:
  for num, name in series.iteritems():
  if name == searched_category[1]:
- _categ_found, searched_category_num = True, num
+ categ_found, searched_category_num = True, num
  break
  else: raise RuntimeError("Cannot determine the category number!")
 
+def get_serialized_results(sess_handle, topic_cid, topic_num, topic_src):
+ serialized_results = {}
+ for checksum, grid_json in sess_handle.query(model.Grid.checksum, model.Grid.info).join(model.tags, model.Grid.checksum == model.tags.c.checksum).join(model.Topic, model.tags.c.tid == model.Topic.tid).filter(model.Topic.cid == topic_cid, model.Topic.topic == topic_num).all():
+ grid_json = json.loads(grid_json)
+ try: serialized_results[checksum] = grid_json[topic_src]
+ except KeyError: pass # no such property in JSON
+ return serialized_results
+
+def get_relational_results(sess_handle, topic_num):
+ relational_results = {}
+ for checksum, space_grp in sess_handle.query(model.Calculation.checksum, model.Spacegroup.n).join(model.Spacegroup, model.Calculation.checksum == model.Spacegroup.checksum).join(model.Basis, model.Calculation.checksum == model.Basis.checksum).filter(model.Basis.kind == topic_num).all():
+ relational_results[checksum] = space_grp
+ return relational_results
+
+class Test_Query_Approaches(TestLayerDB):
+ __test_calcs_dir__ = EXAMPLE_DIR
+
  @classmethod
  def setUpClass(cls):
  super(Test_Query_Approaches, cls).setUpClass(dbname=__name__.split('.')[-1])
 
  def test_equality(self):
- serialized_results = {}
  tick1 = time.time()
- for checksum, grid_json in self.db.session.query(model.Grid.checksum, model.Grid.info).join(model.tags, model.Grid.checksum == model.tags.c.checksum).join(model.Topic, model.tags.c.tid == model.Topic.tid).filter(model.Topic.cid == self.searched_category[0], model.Topic.topic == self.searched_category_num).all():
- grid_json = json.loads(grid_json)
- try: serialized_results[checksum] = grid_json[self.space_grp_topic_source]
- except KeyError: pass # no such property in JSON
- self.report.warning("Query in serialized approach (%s results) took %1.2f sc" % (len(serialized_results), time.time() - tick1))
+ serialized_results = get_serialized_results(
+ self.db.session,
+ Hierarchy_Params.searched_category[0],
+ Hierarchy_Params.searched_category_num,
+ Hierarchy_Params.space_grp_topic_source
+ )
+ tick2 = time.time()
+ self.report.warning("Query in serialized approach (%s results) took %1.2f sc" % (len(serialized_results), tick2 - tick1))
 
- relational_results = {}
+ tick1 = time.time()
+ relational_results = get_relational_results(
+ self.db.session,
+ Hierarchy_Params.searched_category_num
+ )
  tick2 = time.time()
- for checksum, space_grp in self.db.session.query(model.Calculation.checksum, model.Spacegroup.n).join(model.Spacegroup, model.Calculation.checksum == model.Spacegroup.checksum).join(model.Basis, model.Calculation.checksum == model.Basis.checksum).filter(model.Basis.kind == self.searched_category_num).all():
- relational_results[checksum] = space_grp
- self.report.warning("Query in relational approach (%s results) took %1.2f sc" % (len(relational_results), time.time() - tick2))
+ self.report.warning("Query in relational approach (%s results) took %1.2f sc" % (len(relational_results), tick2 - tick1))
 
  try: self.assertEqual(serialized_results, relational_results,
  "Approaches give different results!\nSerialized:\n%s\nRelational:\n%s\n" % (serialized_results, relational_results))
  except:
  TestLayerDB.failed = True
  raise
+
+
+if __name__ == "__main__":
+ # NB -m tests.functional.query_approaches_test
+
+ from tilde.core.settings import settings, connect_database
+
+ session = connect_database(settings, default_actions=False)
+ print "DB: %s" % (settings['db']['default_sqlite_db'] if settings['db']['engine'] == 'sqlite' else settings['db']['dbname'])
+
+ tick1 = time.time()
+ serialized_results = get_serialized_results(
+ session,
+ Hierarchy_Params.searched_category[0],
+ Hierarchy_Params.searched_category_num,
+ Hierarchy_Params.space_grp_topic_source
+ )
+ tick2 = time.time()
+ print "Query in serialized approach (%s results) took %1.2f sc" % (len(serialized_results), tick2 - tick1)
+
+ tick1 = time.time()
+ relational_results = get_relational_results(
+ session,
+ Hierarchy_Params.searched_category_num
+ )
+ tick2 = time.time()
+ print "Query in relational approach (%s results) took %1.2f sc" % (len(relational_results), tick2 - tick1)
+
+ print "Results equal?", serialized_results==relational_results

tests/functional/query_approaches_test.py

@@ -3,3 +3,4 @@
 nosetests $(dirname $0)/unit/*.py
 #nosetests --nocapture $(dirname $0)/functional/*.py
 nosetests $(dirname $0)/functional/*.py
+nosetests $(dirname $0)/berlinium/*_test.py