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_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",
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 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"))
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 _get_dir_structure(data_mode):
if data_mode == DataMode.mdst:
return b2luigi.get_setting("mdst_dir_structure",
"/hsm/belle2/bdata/Data/release-{p.release}/DB{p.database:08d}/prod{p.prod:08d}/" + \
"e{p.experiment_number:04d}/4S/r{p.run_number:05d}/all/mdst/sub00/" + \
"mdst.{p.prefix}.{p.experiment_number:04d}.{p.run_number:05d}.{p.file_name}.root")
elif data_mode == DataMode.cdst:
return b2luigi.get_setting("cdst_dir_structure",
"/hsm/belle2/bdata/Data/release-{p.release}/DB{p.database:08d}/prod{p.prod:08d}/" + \
"e{p.experiment_number:04d}/4S/r{p.run_number:05d}/all/cdst/sub00/" + \
"cdst.{p.prefix}.{p.experiment_number:04d}.{p.run_number:05d}.{p.file_name}.root")
elif data_mode == DataMode.skimmed_raw:
return b2luigi.get_setting("skimmed_raw_dir_structure",
"/hsm/belle2/bdata/Data/release-{p.release}/DB{p.database:08d}/prod{p.prod:08d}/" + \
"e{p.experiment_number:04d}/4S/r{p.run_number:05d}/all/raw/sub00/" + \
"raw.{p.prefix}.{p.file_name}.{p.experiment_number:04d}.{p.run_number:05d}.root")
elif data_mode == DataMode.raw:
return b2luigi.get_setting("raw_dir_structure",
"/ghi/fs01/belle2/bdata/Data/Raw/e{p.experiment_number:04d}/r{p.run_number:05d}/sub00/" + \
def _get_dir_structure(data_mode):
if data_mode == DataMode.mdst:
return b2luigi.get_setting("mdst_dir_structure",
"/hsm/belle2/bdata/Data/release-{p.release}/DB{p.database:08d}/prod{p.prod:08d}/" + \
"e{p.experiment_number:04d}/4S/r{p.run_number:05d}/all/mdst/sub00/" + \
"mdst.{p.prefix}.{p.experiment_number:04d}.{p.run_number:05d}.{p.file_name}.root")
elif data_mode == DataMode.cdst:
return b2luigi.get_setting("cdst_dir_structure",
"/hsm/belle2/bdata/Data/release-{p.release}/DB{p.database:08d}/prod{p.prod:08d}/" + \
"e{p.experiment_number:04d}/4S/r{p.run_number:05d}/all/cdst/sub00/" + \
"cdst.{p.prefix}.{p.experiment_number:04d}.{p.run_number:05d}.{p.file_name}.root")
elif data_mode == DataMode.skimmed_raw:
return b2luigi.get_setting("skimmed_raw_dir_structure",
"/hsm/belle2/bdata/Data/release-{p.release}/DB{p.database:08d}/prod{p.prod:08d}/" + \
"e{p.experiment_number:04d}/4S/r{p.run_number:05d}/all/raw/sub00/" + \
"raw.{p.prefix}.{p.file_name}.{p.experiment_number:04d}.{p.run_number:05d}.root")
elif data_mode == DataMode.raw:
return b2luigi.get_setting("raw_dir_structure",
"/ghi/fs01/belle2/bdata/Data/Raw/e{p.experiment_number:04d}/r{p.run_number:05d}/sub00/" + \
"{p.prefix}.{p.experiment_number:04d}.{p.run_number:05d}.{p.file_name}.root")
def _get_dir_structure(data_mode):
if data_mode == DataMode.mdst:
return b2luigi.get_setting("mdst_dir_structure",
"/hsm/belle2/bdata/Data/release-{p.release}/DB{p.database:08d}/prod{p.prod:08d}/" + \
"e{p.experiment_number:04d}/4S/r{p.run_number:05d}/all/mdst/sub00/" + \
"mdst.{p.prefix}.{p.experiment_number:04d}.{p.run_number:05d}.{p.file_name}.root")
elif data_mode == DataMode.cdst:
return b2luigi.get_setting("cdst_dir_structure",
"/hsm/belle2/bdata/Data/release-{p.release}/DB{p.database:08d}/prod{p.prod:08d}/" + \
"e{p.experiment_number:04d}/4S/r{p.run_number:05d}/all/cdst/sub00/" + \
"cdst.{p.prefix}.{p.experiment_number:04d}.{p.run_number:05d}.{p.file_name}.root")
elif data_mode == DataMode.skimmed_raw:
return b2luigi.get_setting("skimmed_raw_dir_structure",
"/hsm/belle2/bdata/Data/release-{p.release}/DB{p.database:08d}/prod{p.prod:08d}/" + \
"e{p.experiment_number:04d}/4S/r{p.run_number:05d}/all/raw/sub00/" + \
"raw.{p.prefix}.{p.file_name}.{p.experiment_number:04d}.{p.run_number:05d}.root")
elif data_mode == DataMode.raw:
return b2luigi.get_setting("raw_dir_structure",
"/ghi/fs01/belle2/bdata/Data/Raw/e{p.experiment_number:04d}/r{p.run_number:05d}/sub00/" + \
"{p.prefix}.{p.experiment_number:04d}.{p.run_number:05d}.{p.file_name}.root")