CYCAMORE
helper.py
Go to the documentation of this file.
1 """A set of tools for use in integration tests."""
2 import os
3 import tempfile
4 import subprocess
5 import sys
6 from hashlib import sha1
7 import numpy as np
8 import tables
9 from nose.tools import assert_equal
10 
11 if sys.version_info[0] >= 3:
12  str_types = (bytes, str)
13 else:
14  str_types = (str, unicode)
15 
16 def hasher(x):
17  return int(sha1(x.encode()).hexdigest(), 16)
18 
19 def idx(h):
20  ind = [None] * 5
21  for i in range(4, -1, -1):
22  h, ind[i] = divmod(h, 2**32)
23  return tuple(ind)
24 
25 sha1array = lambda x: np.array(idx(hasher(x)), np.uint32)
26 
27 def table_exist(db, tables):
28  """Checks if hdf5 database contains the specified tables.
29  """
30  return all([t in db.root for t in tables])
31 
32 def find_ids(data, data_table, id_table):
33  """Finds ids of the specified data located in the specified data_table,
34  and extracts the corresponding id from the specified id_table.
35  """
36  ids = []
37  for i, d in enumerate(data_table):
38  if isinstance(d, np.ndarray) and isinstance(data, np.ndarray):
39  if (d == data).all():
40  ids.append(id_table[i])
41  elif isinstance(d, np.ndarray) and not isinstance(data, np.ndarray):
42  if (d == sha1array(data)).all():
43  ids.append(id_table[i])
44  elif d == data:
45  ids.append(id_table[i])
46  return ids
47 
48 def exit_times(agent_id, exit_table):
49  """Finds exit times of the specified agent from the exit table.
50  """
51  i = 0
52  exit_times = []
53  for index in exit_table["AgentId"]:
54  if index == agent_id:
55  exit_times.append(exit_table["ExitTime"][i])
56  i += 1
57 
58  return exit_times
59 
60 def run_cyclus(cyclus, cwd, in_path, out_path):
61  """Runs cyclus with various inputs and creates output databases
62  """
63  holdsrtn = [1] # needed because nose does not send() to test generator
64  # make sure the output target directory exists
65  cmd = [cyclus, "-o", out_path, "--input-file", in_path]
66  check_cmd(cmd, cwd, holdsrtn)
67 
68 def check_cmd(args, cwd, holdsrtn):
69  """Runs a command in a subprocess and verifies that it executed properly.
70  """
71  if not isinstance(args, str_types):
72  args = " ".join(args)
73  print("TESTING: running command in {0}:\n\n{1}\n".format(cwd, args))
74  env = dict(os.environ)
75  env['_'] = subprocess.check_output(['which', 'cyclus'], cwd=cwd).strip()
76  with tempfile.NamedTemporaryFile() as f:
77  rtn = subprocess.call(args, shell=True, cwd=cwd, stdout=f, stderr=f, env=env)
78  if rtn != 0:
79  f.seek(0)
80  print("STDOUT + STDERR:\n\n" + f.read().decode())
81  holdsrtn[0] = rtn
82  assert_equal(rtn, 0)
cycamore::GrowthRegion int
def run_cyclus(cyclus, cwd, in_path, out_path)
Definition: helper.py:60
def hasher(x)
Definition: helper.py:16
def exit_times(agent_id, exit_table)
Definition: helper.py:48
def idx(h)
Definition: helper.py:19
def check_cmd(args, cwd, holdsrtn)
Definition: helper.py:68
def table_exist(db, tables)
Definition: helper.py:27
sha1array
Definition: helper.py:25
def find_ids(data, data_table, id_table)
Definition: helper.py:32