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_set_by_file(self):
with open("settings.json", "w") as f:
json.dump({"my_setting": "my file value"}, f)
self.assertEqual("my file value", b2luigi.get_setting("my_setting"))
b2luigi.set_setting("my_setting", "my value")
self.assertEqual("my value", b2luigi.get_setting("my_setting"))
def run(self):
print("Hello!")
with open(self.get_output_file_name("some_file.txt"), "w") as f:
f.write("Done")
print("Bye!")
import sys
sys.stdout.flush()
os.kill(os.getpid(), 11)
with open(self.get_output_file_name("some_other_file.txt"), "w") as f:
f.write("Done")
if __name__ == "__main__":
b2luigi.set_setting("result_dir", "results")
b2luigi.process(MyTask())
def test_deprecated_settings(self):
self.assertRaises(ValueError, b2luigi.get_setting, key="my_setting",
deprecated_keys=["my_old_setting"])
b2luigi.set_setting("my_old_setting", "my value")
with warnings.catch_warnings(record=True) as w:
self.assertEqual("my value", b2luigi.get_setting("my_setting",
deprecated_keys=["my_old_setting"]))
self.assertEqual(len(w), 1)
self.assertIsInstance(w[-1].message, DeprecatedSettingsWarning)
self.assertIn("deprecated", str(w[-1].message))
b2luigi.set_setting("my_setting", "my new_value")
with warnings.catch_warnings(record=True) as w:
self.assertEqual("my new_value", b2luigi.get_setting("my_setting", default="default",
deprecated_keys=["my_old_setting"]))
self.assertEqual(len(w), 0)
def process(self):
print("Hello!")
with open(self.get_output_file_name("some_file.txt"), "w") as f:
f.write("Done")
print("Bye!")
import sys
sys.stdout.flush()
os.kill(os.getpid(), 11)
with open(self.get_output_file_name("some_other_file.txt"), "w") as f:
f.write("Done")
if __name__ == "__main__":
b2luigi.set_setting("result_dir", "results")
b2luigi.process(MyTask())
def test_file_path_usage(self):
class TaskA(b2luigi.Task):
some_parameter = b2luigi.IntParameter()
def output(self):
yield self.add_to_output("file_a")
yield self.add_to_output("file_b")
task = TaskA(some_parameter=3)
b2luigi.set_setting("result_dir", "results/some_crazy_path")
self.assertEqual(get_filled_params(task), {"some_parameter": 3})
self.assertFalse(task.get_input_file_names())
self.assertRaises(KeyError, lambda: task._get_input_targets("some_file"))
self.assertEqual(task._get_output_target("file_a").path, task.get_output_file_name("file_a"))
self.assertIn("file_a", task.get_output_file_name("file_a"))
self.assertIn("file_b", task.get_output_file_name("file_b"))
self.assertIn("some_parameter=3", task.get_output_file_name("file_a"))
self.assertIn("some_crazy_path", task.get_output_file_name("file_a"))
import b2luigi
class MyTask(b2luigi.Task):
def output(self):
yield self.add_to_output("test.txt")
@b2luigi.on_temporary_files
def run(self):
with open(self.get_output_file_name("test.txt"), "w") as f:
f.write("Test")
if __name__ == "__main__":
b2luigi.set_setting("result_dir", "results")
b2luigi.process(MyTask())
class MyNumberTask(b2luigi.Task):
some_parameter = b2luigi.IntParameter()
def output(self):
yield self.add_to_output("output_file.txt")
def run(self):
random_number = random.random()
with open(self.get_output_file_name("output_file.txt"), "w") as f:
f.write(f"{random_number}\n")
if __name__ == "__main__":
b2luigi.set_setting("result_dir", "results")
b2luigi.process([MyNumberTask(some_parameter=i) for i in range(100)],
workers=200)
def test_set_by_function(self):
self.assertRaises(ValueError, b2luigi.get_setting, "my_setting")
self.assertEqual("default", b2luigi.get_setting("my_setting", "default"))
b2luigi.set_setting("my_setting", "my value")
self.assertEqual("my value", b2luigi.get_setting("my_setting"))
self.assertEqual("my value", b2luigi.get_setting("my_setting", "default"))