import shutil
def normalize_ext(e, ext):
- alias = e.cfgs["ALIAS_EXT_" + ext]
+ alias = e["ALIAS_EXT_" + ext]
return alias if alias != "" else ext
def try_ext(e, ext):
- for e in e.cfgs["EXTENSIONS"].split():
+ for e in e["EXTENSIONS"].split():
if e == ext:
return
raise moe.MoeError, "Unknown extension: " + ext
def locate(e, filename=None):
e.log.progress("Locating source... ")
- task = e.cfgs["TASK"]
+ task = e["TASK"]
if filename is None:
dir = ""
file = task
else:
dir, file = os.path.split(filename)
if dir == "":
- dir = e.cfgs["SDIR"]
+ dir = e["SDIR"]
base, ext = os.path.splitext(file)
if ext != "":
try_ext(e, ext)
else:
found = []
- for ext in e.cfgs["EXTENSIONS"].split():
+ for ext in e["EXTENSIONS"].split():
if os.path.exists(os.path.join(dir, base + "." + ext)):
found.append(ext)
if len(found) == 0:
norm_ext = normalize_ext(e, ext)
e.log.verbose("Found solution %s\n" % orig_path)
- copy = e.cfgs["TASK"] + "." + norm_ext
- copy_path = os.path.join(e.cfgs["TDIR"], copy)
+ copy = e["TASK"] + "." + norm_ext
+ copy_path = os.path.join(e["TDIR"], copy)
if file != copy:
e.log.verbose("Renaming to %s\n" % copy)
moe.util.link_or_copy(orig_path, copy_path)
def compile_init(e):
e.log.progress("Compiling... ")
boxdir = moe.box.setup(e)
- pdir = e.cfgs["PDIR"]
- tdir = e.cfgs["TDIR"]
- shutil.copyfile(os.path.join(tdir, e.cfgs["SRC"]), os.path.join(boxdir, e.cfgs["SRC"]))
- for x in e.cfgs["EXTRAS"].split() + e.cfgs["COMP_EXTRAS"].split():
+ pdir = e["PDIR"]
+ tdir = e["TDIR"]
+ shutil.copyfile(os.path.join(tdir, e["SRC"]), os.path.join(boxdir, e["SRC"]))
+ for x in e["EXTRAS"].split() + e["COMP_EXTRAS"].split():
xx = os.path.join(tdir, x)
if not os.path.isfile(xx):
xx = os.path.join(pdir, x)
def compile_run(e):
moe.box.show(e, "compiler input")
- cc = e.cfgs["COMP"]
+ cc = e["COMP"]
e.log.verbose("Compilation command: %s\n" % cc)
- rc = moe.box.run(e, e.cfgs["COMP_SANDBOX_OPTS"], cc)
+ rc = moe.box.run(e, e["COMP_SANDBOX_OPTS"], cc)
if rc > 0:
e.log.progress("FAILED\n")
## FIXME: status file ... or generate an exception?
def compile_done(e):
try:
- shutil.copyfile(os.path.join(e.cfgs["BOXDIR"], e.cfgs["EXE"]), os.path.join(e.cfgs["TDIR"], e.cfgs["EXE"]))
+ shutil.copyfile(os.path.join(e["BOXDIR"], e["EXE"]), os.path.join(e["TDIR"], e["EXE"]))
except IOError:
raise moe.MoeError, "Compiler succeeded, but produced no output"
e.log.progress("OK\n")
def test_in(e):
- tdir = e.cfgs["TDIR"]
+ tdir = e["TDIR"]
boxdir = moe.box.setup(e)
- inn = e.cfgs["TESTCASE_IN"]
- in_type = e.cfgs["IN_TYPE"] or e.cfgs["IO_TYPE"]
- out_type = e.cfgs["OUT_TYPE"] or e.cfgs["IO_TYPE"]
- is_interactive = e.cfgs["TASK_TYPE"] == "interactive"
- sandbox_opts = "-M" + os.path.join(tdir, e.cfgs["TESTCASE_STATUS"])
+ inn = e["TESTCASE_IN"]
+ in_type = e["IN_TYPE"] or e["IO_TYPE"]
+ out_type = e["OUT_TYPE"] or e["IO_TYPE"]
+ is_interactive = e["TASK_TYPE"] == "interactive"
+ sandbox_opts = "-M" + os.path.join(tdir, e["TESTCASE_STATUS"])
- if not os.path.exists(os.path.join(tdir, e.cfgs["EXE"])):
+ if not os.path.exists(os.path.join(tdir, e["EXE"])):
raise TestError("Compilation failed", "CE")
- shutil.copyfile(os.path.join(tdir, e.cfgs["EXE"]), os.path.join(boxdir, e.cfgs["EXE"]))
- os.chmod(os.path.join(boxdir, e.cfgs["EXE"]), 0555)
+ shutil.copyfile(os.path.join(tdir, e["EXE"]), os.path.join(boxdir, e["EXE"]))
+ os.chmod(os.path.join(boxdir, e["EXE"]), 0555)
if in_type == "file":
- in_name = e.cfgs["IN_NAME"]
- e.log.verbose("Input file: %s (copied from %s)\n" % (in_name, os.path.join(e.cfgs["PDIR"], inn)))
+ in_name = e["IN_NAME"]
+ e.log.verbose("Input file: %s (copied from %s)\n" % (in_name, os.path.join(e["PDIR"], inn)))
shutil.copyfile(os.path.join(tdir, inn), os.path.join(boxdir, in_name))
if not is_interactive:
sandbox_opts += " -i/dev/null"
elif in_type == "stdio":
- e.log.verbose("Input file: <stdin> (copied from %s)\n" % os.path.join(e.cfgs["PDIR"], inn))
+ e.log.verbose("Input file: <stdin> (copied from %s)\n" % os.path.join(e["PDIR"], inn))
shutil.copyfile(os.path.join(tdir, inn), os.path.join(boxdir, ".stdin"))
sandbox_opts += " -i.stdin"
elif in_type == "none":
raise MoeError, "Unknown input type %s" % in_type
if out_type == "file":
- out_name = e.cfgs["OUT_NAME"]
+ out_name = e["OUT_NAME"]
e.log.verbose("Output file: %s\n" % out_name)
if not is_interactive:
sandbox_opts += " -o/dev/null"
e.test_builtins.set("BOX_IO_OPTS", sandbox_opts)
def test_run(e):
- e.log.verbose("Time limit: %s s\n" % e.cfgs["TIME_LIMIT"])
- e.log.verbose("Memory limit: %s KB\n" % e.cfgs["MEM_LIMIT"])
+ e.log.verbose("Time limit: %s s\n" % e["TIME_LIMIT"])
+ e.log.verbose("Memory limit: %s KB\n" % e["MEM_LIMIT"])
moe.box.show(e, "test input")
e.log.progress("<run> ")
- rc = moe.box.run(e, e.cfgs["TEST_SANDBOX_OPTS"], e.cfgs["TEST_EXEC_CMD"])
+ rc = moe.box.run(e, e["TEST_SANDBOX_OPTS"], e["TEST_EXEC_CMD"])
moe.testcase.collect_status(e)
moe.box.show(e, "test output")
if rc > 0:
raise moe.TestError("Wrong answer", "WA")
def test_collect(e):
- tdir = e.cfgs["TDIR"]
- boxdir = e.cfgs["BOXDIR"]
- out_type = e.cfgs["OUT_TYPE"] or e.cfgs["IO_TYPE"]
- is_interactive = e.cfgs["TASK_TYPE"] == "interactive"
+ tdir = e["TDIR"]
+ boxdir = e["BOXDIR"]
+ out_type = e["OUT_TYPE"] or e["IO_TYPE"]
+ is_interactive = e["TASK_TYPE"] == "interactive"
if out_type == "file":
- out_path = e.cfgs["OUT_NAME"]
+ out_path = e["OUT_NAME"]
elif out_type == "stdio":
out_path = ".stdout"
if not os.path.exists(os.path.join(boxdir, out_path)):
raise moe.TestError("No output file", "NO")
- shutil.copyfile(os.path.join(boxdir, out_path), os.path.join(tdir, e.cfgs["TESTCASE_OUT"]))
+ shutil.copyfile(os.path.join(boxdir, out_path), os.path.join(tdir, e["TESTCASE_OUT"]))
def tests(e):
e.log.progress("\n")
self.stat = moe.status.MoeStatus()
pass
+ def __getitem__(self, key):
+ return self.cfgs[key]
+
def init(self, overrides=None):
self.log.progress("Initializing... ")
self.init_global(overrides)
self.log.progress("OK\n")
def init_global(self, overrides):
- main_cfg = moe.config.MoeConfig(name = os.path.join(self.cfgs['HOME'], "config"), type="main")
+ main_cfg = moe.config.MoeConfig(name = os.path.join(self['HOME'], "config"), type="main")
self.cfgs.push(main_cfg)
if overrides:
self.cfgs.push(overrides)
def init_test(self):
- test = self.cfgs['TDIR']
+ test = self['TDIR']
if os.path.isdir(test):
shutil.rmtree(test)
try:
def init_logs(self):
self.log = moe.log.MoeLog()
- if self.cfgs["V"]:
- self.log.verbosity = int(self.cfgs["V"])
- self.log.open(os.path.join(self.cfgs["TDIR"], "log"))
+ if self["V"]:
+ self.log.verbosity = int(self["V"])
+ self.log.open(os.path.join(self["TDIR"], "log"))
self.default_log = self.log
moe.log.default = self.log
self.log_config(3, "before loading the task")
def init_task(self):
- task = self.cfgs['TASK']
- task_dir = self.cfgs['PDIR']
+ task = self['TASK']
+ task_dir = self['PDIR']
if not os.path.isdir(task_dir):
raise moe.MoeError, "No such task %s" % task
self.stat["task"] = task
- type = self.cfgs['TASK_TYPE']
+ type = self['TASK_TYPE']
if type == "batch" or type == "interactive":
moe.batch.prepare_pipe(self)
elif type == "opendata":
def run(self):
self.log_config(2, "for the task pipeline")
- self.main_pipe.configure(self.cfgs["HOOKS"])
+ self.main_pipe.configure(self["HOOKS"])
if self.log.verbosity >= 2:
self.main_pipe.dump(self.log.log_file, prefix="\t")
self.main_pipe.run(self)
e.test_builtins.set("TEST", test)
e.cfgs.push(e.test_builtins)
- test_cf = os.path.join(e.cfgs["PDIR"], test + ".config")
+ test_cf = os.path.join(e["PDIR"], test + ".config")
if os.path.exists(test_cf):
cfg = moe.config.MoeConfig(name=test_cf, type="test")
e.cfgs.push(cfg)
e.cfgs.apply_overrides("TEST_" + test + "_")
- ext = e.cfgs["EXT"]
+ ext = e["EXT"]
if ext != "":
e.cfgs.apply_overrides("EXT_" + ext + "_")
log = moe.log.MoeLog()
log.verbosity = e.log.verbosity
- log.open(os.path.join(e.cfgs["TDIR"], test + ".log"))
+ log.open(os.path.join(e["TDIR"], test + ".log"))
log.say("Test case %s\n\n" % test)
e.log = log
moe.log.default = log
e.stat.get_list("tests").append(e.test_stat)
def collect_status(e):
- sf = os.path.join(e.cfgs["TDIR"], e.cfgs["TESTCASE_STATUS"])
+ sf = os.path.join(e["TDIR"], e["TESTCASE_STATUS"])
if os.path.exists(sf):
e.log.verbose("Reading status from %s\n" % sf)
stat = moe.status.MoeStatus()
e.log.verbose("No status file present\n")
def tmpname(e):
- return os.path.join(e.cfgs["TDIR"], e.cfgs["TEST"] + ".tmp")
+ return os.path.join(e["TDIR"], e["TEST"] + ".tmp")
def collect_verdict(e, verdict_file):
try:
pass
def setup(e):
- pdir = e.cfgs["PDIR"]
- tdir = e.cfgs["TDIR"]
- inn = e.cfgs["TESTCASE_IN"]
- out = e.cfgs["TESTCASE_OUT"]
- ok = e.cfgs["TESTCASE_OK"]
+ pdir = e["PDIR"]
+ tdir = e["TDIR"]
+ inn = e["TESTCASE_IN"]
+ out = e["TESTCASE_OUT"]
+ ok = e["TESTCASE_OK"]
if os.path.exists(os.path.join(pdir, inn)):
moe.util.link_or_copy(os.path.join(pdir, inn), os.path.join(tdir, inn))
moe.util.link_or_copy(os.path.join(pdir, out), os.path.join(tdir, ok))
def judge(e):
- cmd = e.cfgs["OUTPUT_CHECK"]
+ cmd = e["OUTPUT_CHECK"]
if cmd == "":
return
verdict_file = tmpname(e)
def points(e):
## FIXME: check $TEST.pts
if e.test_stat["points"] is None:
- e.test_stat["points"] = e.cfgs["POINTS_PER_TEST"]
+ e.test_stat["points"] = e["POINTS_PER_TEST"]
def run_test(e, test):
configure_test(e, test)
## FIXME: interactive tasks
- e.test_pipe.configure(e.cfgs["TESTCASE_HOOKS"])
+ e.test_pipe.configure(e["TESTCASE_HOOKS"])
if e.log.verbosity >= 2:
e.test_pipe.dump(e.log.log_file, prefix="\t")
e.test_pipe.run(e)
e.test_pipe.insert(400, "judge", judge)
e.test_pipe.insert(500, "points", points)
- for test in e.cfgs["TESTS"].split():
+ for test in e["TESTS"].split():
e.log.progress("Test %s: " % test)
old_cfgs = e.cfgs
old_log = e.log