Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _test_run_with_empty_data_frame(cmdline_args: List[str], test_run_params: test_run):
from unittest.mock import patch
try:
gokart.run(cmdline_args=cmdline_args)
except SystemExit as e:
assert e.code == 0, f'original workflow does not run properly. It exited with error code {e}.'
with CmdlineParser.global_instance(cmdline_args) as cp:
all_tasks = _get_all_tasks(cp.get_task_obj())
if test_run_params.namespace is not None:
all_tasks = [t for t in all_tasks if t.task_namespace == test_run_params.namespace]
with patch('gokart.TaskOnKart.load_data_frame', new=lambda *args, required_columns=None, **kwargs: pd.DataFrame(columns=required_columns)):
with patch('gokart.TaskOnKart.dump', new=lambda *args, **kwargs: None):
test_status_list = [_run_with_test_status(t) for t in all_tasks]
test_logger.info('gokart test results:\n' + '\n'.join(s.format() for s in test_status_list))
if any(s.fail() for s in test_status_list):
exit(1)
def test_run_with_undefined_environ(self):
config_file_path = os.path.join(os.path.dirname(__name__), 'test_config.ini')
luigi.configuration.LuigiConfigParser.add_config_path(config_file_path)
with self.assertRaises(luigi.parameter.MissingParameterException) as missing_parameter:
gokart.run()
def test_fail_with_None(self):
with self.assertRaises(SystemExit) as exit_code:
gokart.run()
self.assertNotEqual(exit_code.exception.code, 0) # raise Error
def test_run(self):
config_file_path = os.path.join(os.path.dirname(__name__), 'test_config.ini')
luigi.configuration.LuigiConfigParser.add_config_path(config_file_path)
os.environ.setdefault('test_param', 'test')
with self.assertRaises(SystemExit) as exit_code:
gokart.run()
self.assertEqual(exit_code.exception.code, 0)
def test_run_with_namespace(self):
argv = [f'{__name__}.DummyWorkFlowWithoutError', '--local-scheduler', '--test-run-pandas', f'--test-run-namespace={__name__}', '--log-level=CRITICAL',
'--no-lock']
logger = logging.getLogger('gokart.testing.check_if_run_with_empty_data_frame')
with patch.object(logger, 'info') as mock_debug:
with self.assertRaises(SystemExit) as exit_code:
gokart.run(argv)
log_str = mock_debug.call_args[0][0]
self.assertEqual(exit_code.exception.code, 0)
self.assertTrue('DummyModelTask' not in log_str)
def test_run_tree_info(self):
config_file_path = os.path.join(os.path.dirname(__name__), 'test_config.ini')
luigi.configuration.LuigiConfigParser.add_config_path(config_file_path)
os.environ.setdefault('test_param', 'test')
tree_info = gokart.tree_info(mode='simple', output_path='tree.txt')
with self.assertRaises(SystemExit):
gokart.run()
self.assertTrue(gokart.make_tree_info(_DummyTask(param='test')), tree_info.output().load())
# //---------------------------------------------------------------------
# Please set fix_random_seed_methods parameter.
# Change seed if you change sample_param.
#
# //--- The output is as follows every time (with pytorch installed). ---
# {'random': [65, 41, 61, 37, 55, 81, 48, 2, 94, 21],
# 'numpy': [79, 86, 5, 22, 79, 98, 56, 40, 81, 37], 'torch': []}
# 'torch': [0.14460121095180511, -0.11649507284164429,
# 0.6928958296775818, -0.916053831577301, 0.7317505478858948]}
#
# //------------------------- without pytorch ---------------------------
# {'random': [65, 41, 61, 37, 55, 81, 48, 2, 94, 21],
# 'numpy': [79, 86, 5, 22, 79, 98, 56, 40, 81, 37], 'torch': []}
#
# //---------------------------------------------------------------------
gokart.run([
'sample_fix_random_seed.SampleTask', '--local-scheduler', '--rerun', '--sample-param=a',
'--fix-random-seed-methods=["random.seed","numpy.random.seed","torch.random.manual_seed"]', '--fix-random-seed-value=57'
])
number = luigi.IntParameter()
def run(self):
self.dump(f'this is sample output. model number: {self.number}')
@requires(SampleTask)
class SecondTask(gokart.TaskOnKart):
task_namespace = 'sample'
param = luigi.Parameter()
def run(self):
sample = self.load()
self.dump(sample + f'add task: {self.param}')
gokart.run()
import gokart
from gokart.info import tree_info
class SampleTaskLog(gokart.TaskOnKart):
def run(self):
self.task_log['sample key'] = 'sample value'
if __name__ == '__main__':
SampleTaskLog().run()
tree_info()
gokart.run(
['--tree-info-mode=all', '--tree-info-output-path=sample_task_log.txt', 'SampleTaskLog', '--local-scheduler'])
# The target file path is
# '{TaskOnKart.workspace_directory}/output_of_task_b_{self.make_unique_id()}.pkl'.
return self.make_target('output_of_task_b.pkl')
def run(self):
# `load` loads input data. In this case, this loads the output of `TaskA`.
output_of_task_a = self.load()
results = f'"{output_of_task_a}" is loaded in TaskB.'
# `dump` writes `results` to the file path of `self.output()`.
self.dump(results)
if __name__ == '__main__':
# luigi.build([TaskB(param='Hello')], local_scheduler=True)
# gokart.run(['--tree-info-mode=simple', '--tree-info-output-path=tree_simple.txt', 'TaskB', '--param=Hello', '--local-scheduler'])
gokart.run(['--tree-info-mode=all', '--tree-info-output-path=tree_all.txt', 'TaskB', '--param=Hello', '--local-scheduler'])