Best Python code snippet using responses
runner.py
Source:runner.py
...64 def run_simulation(self):65 return run_simulation(self.dst_root, popen=False)66 def parse_out(self):67 return parse_out(self.dst_root)68 def run_strict(self, data_map:dict):69 # If efdc_node_list or qser_node_list takes None, the value will not be changed.70 self.write(data_map)71 shell_output = self.run_simulation().decode()72 self.shell_output_list.append(shell_output)73 self.check_shell_output(shell_output)74 return self.parse_out()75 76 def run(self, data_map:dict):77 data_map_filled = data_map_fill(data_map)78 return self.run_strict(data_map_filled)79 def cleanup(self):80 # user may want to keep those files81 rmtree(self.dst_root)82 logging.debug(f"cleanup {self.dst_root}")83 def check_shell_output(self, shell_output:str):84 # TODO: do some check to raise error as early as possible85 # The parsing will fail if model itself failed to complete.86 self.shell_output_parsed_list.append(parse_shell_output(shell_output))87 def __repr__(self):88 return f"Ruuner(src_root={self.src_root}, dst_root={self.dst_root})"89 def debug(self):90 if len(self.shell_output_list) == 0:91 print("No shell output is catched")92 else:93 if len(self.shell_output_list) > 1:94 print("The size of collected shell output size > 1, show last shell output")95 print(self.shell_output_list[-1])96class RunnerInplace(Runner):97 """98 For one who don't want environment isolation (not recommended).99 """100 def __init__(self, src_root):101 self.src_root = Path(src_root)102 self.dst_root = self.src_root103 # create_simulation(src_root, dst_root)104def work(process_args: dict):105 root:str = process_args["root"]106 data_map:dict = process_args["data_map"]107 dst_root = process_args["dst_root"]108 debug_list = process_args["debug_list"]109 idx:int = process_args["idx"]110 runner = Runner(root, dst_root)111 if debug_list is not None:112 debug_list[idx] = runner113 out = runner.run_strict(data_map)114 if debug_list is None:115 runner.cleanup() 116 117 return out118def run_batch(root, data_map_list, pool_size=0, debug_list=None, dst_root_list=None, sequential=False):119 """120 sequential argument is used to control parallel related factor to help debugging121 """122 data_map_list = [x if isinstance(x, dict) else x.data_map for x in data_map_list]123 if pool_size == 0:124 pool_size = get_default_pool_size()125 if dst_root_list is None:126 dst_root_list = [None for _ in data_map_list]127 if debug_list is not None:128 assert len(debug_list) == 0, "debug_list is not None or empty list, maybe mistakenly use a previous list?"129 debug_list.extend([None for _ in data_map_list])130 process_args_list = []131 for idx, (data_map, dst_root) in enumerate(zip(data_map_list, dst_root_list)):132 process_args = {"root":root, "data_map": data_map, "debug_list": debug_list, "dst_root": dst_root, "idx": idx} 133 process_args_list.append(process_args)134 135 if not sequential:136 pool = Pool(pool_size)137 return pool.map(work, process_args_list)138 else:139 return [work(process_arg) for process_arg in process_args_list]140def work_restart(process_args: dict):141 # TODO: An valuable altnative implementation is to just rename three files rather than symbolic link142 runner: Runner = process_args["runner"]143 data_map:dict = process_args["data_map"]144 out = runner.run_strict(data_map)145 return out146def copy_restart_files(src, dst=None):147 src = Path(src)148 dst = Path(dst) if dst is not None else src149 out_to_inp = {150 "RESTART.OUT": "RESTART.INP",151 "TEMPBRST.OUT": "TEMPB.RST",152 "WQWCRST.OUT": "wqini.inp"153 }154 for out_s, inp_s in out_to_inp.items():155 src_p = src / out_s156 dst_p = dst / inp_s157 if dst_p.exists():158 dst_p.unlink()...
test.py
Source:test.py
...121 def run_compare(self, entity, extractor):122 Compare.extractor = extractor123 Compare.run_ann(self, entity)124 Compare.run_out(self, entity, extractor)125 Compare.run_strict(self)126 Compare.run_loose(self)127 Compare.write_to_file(self)128 def run_strict(self): #strict is based on matching start and end span129 if Compare.asspan is not None and Compare.psspan is not None:130 if Compare.asspan == Compare.psspan and Compare.aespan == Compare.pespan:131 Compare.scval = "TP"132 else:133 Compare.scval = "FP"134 elif Compare.asspan is not None:135 Compare.scval = "FN"136 elif Compare.psspan is not None:137 Compare.scval = "FP"138 else:139 Compare.scval = "TN"140 def run_loose(self): #loose is based on matching content string141 if Compare.avc is not None and Compare.pv is not None:142 if Compare.avc == Compare.pv:...
block_matrix.py
Source:block_matrix.py
...6from tqdm import tqdm7from numba import jit8from .socio_matrix import SocioMatrix9log = logging.getLogger()10def run_strict(metric):11 """decorator for metric functions.12 returns a np.nan in case of a RuntimeWarning13 """14 def new(i, j, m):15 with warnings.catch_warnings():16 warnings.simplefilter("error")17 try:18 return metric(i, j, m)19 except RuntimeWarning:20 log.error("Runtime Warning: i=%d, j=%d" % (i, j))21 return np.nan22 return new23@jit("double(int64, int64, float32[:, :])", nopython=True)24def euclidean_metric(i, j, m):...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!