At first, we define test vector generators, who generate a collection of test vectors. Each test vector is descriped as [(factor, value), (factor, value), ...].
In [1]:
import itertools
import collections
import os
import json
import re
In [2]:
def parse_json(fn):
'''Parse json object
This function parses a JSON file. Unlike the builtin `json` module, it
supports "//" like comments, uses 'str' for string representation and
preserves the key orders.
Args:
fn (str): Name of the file to parse
Returns:
OrderedDict: A dict representing the file content
'''
def ununicodify(obj):
result = None
if isinstance(obj, collections.OrderedDict):
result = collections.OrderedDict()
for k, v in obj.iteritems():
k1 = str(k) if isinstance(k, unicode) else k
result[k1] = ununicodify(v)
elif isinstance(obj, list):
result = []
for v in obj:
result.append(ununicodify(v))
elif isinstance(obj, unicode):
result = str(obj)
else:
result = obj
return result
content = file(fn).read()
content = re.sub(r"\s+//.*$", "", content)
return ununicodify(json.loads(content, object_pairs_hook=collections.OrderedDict))
In [3]:
class SimpleVectorGenerator:
def __init__(self, test_factors, raw_vectors=None):
self.test_factors = test_factors
self.raw_vectors = raw_vectors if raw_vectors else []
def iteritems(self):
# expand each vector to support `[0, [1, 2], [3, 4]]`
for item in self.raw_vectors:
iters = [x if isinstance(x, list) else [x] for x in item]
for v in itertools.product(*iters):
yield collections.OrderedDict(zip(self.test_factors, v))
In [4]:
test_factors = ["nnodes", "nmics", "test_id"]
raw_vectors = [[0, 1, 2], [1, [1, 2], 3], [2, 1, [3, 4]]]
generator = SimpleVectorGenerator(test_factors, raw_vectors)
for item in generator.iteritems():
print item
In [5]:
class CartProductVectorGenerator:
def __init__(self, test_factors, factor_values):
self.test_factors = test_factors
self.factor_values = factor_values
def iteritems(self):
ordered_factor_values = [self.factor_values[k] for k in self.test_factors]
for v in itertools.product(*ordered_factor_values):
yield collections.OrderedDict(zip(self.test_factors, v))
In [6]:
test_factors = ["nnodes", "nmics", "test_id"]
factor_values = {
"nnodes": [1, 2, 3],
"nmics": [0, 1, 2, 3],
"test_id": [1]
}
generator = CartProductVectorGenerator(test_factors, factor_values)
for item in generator.iteritems():
print item
We then define some test case generators. Test case generator takes an test vector as the input, together with some auxiliary information, such as the project root, the output root and the working directory. It's responsible to generate necessary aux files as well as a formal description of the test case.
In [7]:
class CustomCaseGenerator:
def __init__(self, module, func, args):
if not os.path.exists(module):
raise RuntimeError("Module '%s' does not exists" % module)
import_result = {}
execfile(module, import_result)
if not func in import_result:
raise RuntimeError("Can not find function '%s' in '%s'" % (func, module))
self.func = import_result[func]
self.args = args
def make_case(self, conf_root, output_root, case_path, test_vector):
'''Generate a test case according to the specified test vector
Args:
conf_root (str): Absolute path containing the project config.
output_root (str): Absolute path for the output root.
case_path (str): Absolute path for the test case.
test_vector (OrderedDict): Test case identification.
Returns:
dict: Test case specification
Test case specification containing the following information to run a test case:
{
"cmd": ["ls", "-l"] # The command and its arguments
"envs": {"K": "V", ...} # The environment variables to set
"results": ["STDOUT"] # The result files to preserve
"run": {"nnodes": 1, ...} # The runner specific information
}
'''
args = dict(self.args)
args["conf_root"] = conf_root
args["output_root"] = output_root
args["case_path"] = case_path
args["test_vector"] = test_vector
case_spec = self.func(**args)
return case_spec
class OutputOrganizer:
def __init__(self, version=1):
if version != 1:
raise RangeError("Unsupported output version '%s': only allow 1" % version)
self.version = version
def get_case_path(self, test_vector):
segs = ["{0}-{1}".format(k, v) for k, v in test_vector.iteritems()]
return os.path.join(*segs)
def get_project_info_path(self):
return "TestProject.json"
def get_case_spec_path(self, test_vector):
return os.path.join(self.get_case_path(test_vector), "TestCase.json")
The design is that we seperate the test vector generation, test case generation and test case organization. So we can add more vector generation method, case generation method and organization methods as needed.
Generate necessary files for a specified test case in a specified directory. It can also generate files in anywhere inside the output directory. But it should not rely on the directory layout of the cases. The idea is that case generator shall put all necessary files in case-specific directory, so the case is self-contained. But since case usually rely on some shared public files, they can be put in the output directory.
In [8]:
class TestProject:
def __init__(self, conf_root):
if not os.path.isabs(conf_root):
conf_root = os.path.abspath(conf_root)
self.conf_root = conf_root
spec_file = os.path.join(self.conf_root, "TestProjectConfig.json")
spec = parse_json(spec_file)
# TODO: Refactor to support multiple versions in the future.
project_format = spec["format"]
if int(project_format) != 1:
raise RuntimeError("Unsupported project format '%s': only allow '1'" % project_format)
# basic project information
project_info = spec["project"]
self.name = project_info["name"]
self.test_factors = project_info["test_factors"]
data_files = project_info.get("data_files", [])
self.data_files = []
for item in data_files:
if os.path.isabs(item):
self.data_files.append(item)
else:
path = os.path.normpath(os.path.join(self.conf_root, item))
self.data_files.append(path)
# build test vector generator
test_vector_generator_name = project_info["test_vector_generator"]
if test_vector_generator_name == "cart_product":
args = spec["cart_product_vector_generator"]
test_factor_values = args["test_factor_values"]
self.test_vector_generator = CartProductVectorGenerator(self.test_factors,
test_factor_values)
elif test_vector_generator_name == "simple":
args = spec["simple_vector_generator"]
test_vectors = args["test_vectors"]
self.test_vector_generator = SimpleVectorGenerator(self.test_factors,
test_vectors)
else:
raise RangeError("Unknown test vector generator '%s'" % test_vector_generator_name)
# build test case generator
test_case_generator_name = project_info["test_case_generator"]
if test_case_generator_name == "custom":
info = spec["custom_case_generator"]
module = info["import"]
if not os.path.isabs(module):
module = os.path.normpath(os.path.join(self.conf_root, module))
func = info["func"]
args = info["args"]
self.test_case_generator = CustomCaseGenerator(module, func, args)
else:
raise RangeError("Unknown test case generator '%s'" % test_case_generator_name)
# build output organizer
self.output_organizer = OutputOrganizer(version=1)
def write(self, output_root):
if not os.path.isabs(output_root):
output_root = os.path.abspath(output_root)
if not os.path.exists(output_root):
os.makedirs(output_root)
for case in self.test_vector_generator.iteritems():
case_path = self.output_organizer.get_case_path(case)
case_fullpath = os.path.join(output_root, case_path)
if not os.path.exists(case_fullpath):
os.makedirs(case_fullpath)
cwd = os.path.abspath(os.getcwd())
os.chdir(case_fullpath)
try:
case_spec = self.test_case_generator.make_case(self.conf_root, output_root, case_fullpath, case)
finally:
os.chdir(cwd)
case_spec_path = self.output_organizer.get_case_spec_path(case)
case_spec_fullpath = os.path.join(output_root, case_spec_path)
json.dump(case_spec, file(case_spec_fullpath, "w"), indent=4)
# TODO: handle data_files
info = [("name", self.name), ("test_factors", self.test_factors), ("data_files", self.data_files)]
info = collections.OrderedDict(info)
x = [case.values() for case in self.test_vector_generator.iteritems()]
y = [self.output_organizer.get_case_path(case) for case in self.test_vector_generator.iteritems()]
test_defs = collections.OrderedDict()
test_defs["test_vectors"] = x
test_defs["case_paths"] = y
info["test_cases"] = test_defs
project_info_path = self.output_organizer.get_project_info_path()
project_info_fullpath = os.path.join(output_root, project_info_path)
json.dump(info, file(project_info_fullpath, "w"), indent=4)
In [9]:
project = TestProject("tests/generator/new")
project.write("result")
In [ ]: