diff --git a/pleskdistup/common/src/leapp_configs.py b/pleskdistup/common/src/leapp_configs.py index 74ed4f0..511416c 100644 --- a/pleskdistup/common/src/leapp_configs.py +++ b/pleskdistup/common/src/leapp_configs.py @@ -306,6 +306,55 @@ def set_package_repository(package: str, repository: str, leapp_pkgs_conf_path: files.rewrite_json_file(leapp_pkgs_conf_path, pkg_mapping) +def take_free_packageset_id(pkg_mapping: dict) -> int: + max_id: int = 0 + if "packageinfo" not in pkg_mapping: + return 1 + + for info in pkg_mapping["packageinfo"]: + input_set_id = 0 + if "in_packageset" in info and info["in_packageset"] and info["in_packageset"]["set_id"]: + input_set_id = int(info["in_packageset"]["set_id"]) + + output_set_id = 0 + if "out_packageset" in info and info["out_packageset"] and info["out_packageset"]["set_id"]: + output_set_id = int(info["out_packageset"]["set_id"]) + + max_id = max(max_id, input_set_id, output_set_id) + + return max_id + 1 + + +def set_package_mapping( + in_package: str, + source_repository: str, + out_package: str, + target_repository: str, + leapp_pkgs_conf_path: str = LEAPP_PKGS_CONF_PATH +) -> None: + pkg_mapping = None + log.debug(f"Reconfigure mapping for package {in_package!r} from repository {source_repository!r} to package {out_package!r} from repository {target_repository!r}") + with open(leapp_pkgs_conf_path, "r") as pkg_mapping_file: + pkg_mapping = json.load(pkg_mapping_file) + for info in pkg_mapping["packageinfo"]: + if not info["in_packageset"] or not info["in_packageset"]["package"]: + continue + + if all(inpkg["name"] != in_package or inpkg["repository"] != source_repository for inpkg in info["in_packageset"]["package"]): + continue + + log.debug(f"Change '{in_package}' package repository in info '{info['id']}' -> in packageset '{info['in_packageset']['set_id']}'") + if "out_packageset" not in info or not info["out_packageset"]: + info["out_packageset"] = {"set_id": take_free_packageset_id(pkg_mapping), "package": []} + info["out_packageset"]["package"].append({"name": out_package, "repository": target_repository}) + else: + info["out_packageset"]["package"] = [] + info["out_packageset"]["package"].append({"name": out_package, "repository": target_repository}) + + log.debug(f"Write json into '{leapp_pkgs_conf_path}'") + files.rewrite_json_file(leapp_pkgs_conf_path, pkg_mapping) + + # The following types are defined in the leapp-repository repository and can be used # to define the action type of the package in the pes-events.json file. class LeappActionType(IntEnum): @@ -335,3 +384,23 @@ def set_package_action(package: str, actionType: LeappActionType, leapp_pkgs_con log.debug(f"Write json into '{leapp_pkgs_conf_path}'") files.rewrite_json_file(leapp_pkgs_conf_path, pkg_mapping) + + +def remove_package_action(package: str, repository: str, leapp_pkgs_conf_path: str = LEAPP_PKGS_CONF_PATH): + log.debug(f"Remove action for package '{package}' from repository '{repository}'") + + def is_action_for_target_package(action: dict) -> bool: + if "in_packageset" not in action or action["in_packageset"] is None or "package" not in action["in_packageset"]: + return False + return any(inpackage["name"] == package and inpackage["repository"] == repository for inpackage in action["in_packageset"]["package"]) + + pkg_mapping = None + with open(leapp_pkgs_conf_path, "r") as pkg_mapping_file: + pkg_mapping = json.load(pkg_mapping_file) + if "packageinfo" not in pkg_mapping: + return + + pkg_mapping["packageinfo"] = [action for action in pkg_mapping["packageinfo"] if not is_action_for_target_package(action)] + + log.debug(f"Write json into '{leapp_pkgs_conf_path}'") + files.rewrite_json_file(leapp_pkgs_conf_path, pkg_mapping) diff --git a/pleskdistup/common/tests/leapp_configs_tests.py b/pleskdistup/common/tests/leapp_configs_tests.py index 6d5ad1d..1f602ec 100644 --- a/pleskdistup/common/tests/leapp_configs_tests.py +++ b/pleskdistup/common/tests/leapp_configs_tests.py @@ -635,3 +635,441 @@ def test_set_unexcited_package_action(self): with open(self.JSON_FILE_PATH, "r") as f: json_data = json.load(f) self.assertEqual(json_data, self.INITIAL_JSON) + + +class TakeFreePackagesetIdTests(unittest.TestCase): + def test_simple_take(self): + json_data = { + "packageinfo": [ + { + "action": 1, + "id": "1", + "in_packageset": { + "package": [ + { + "name": "some", + "repository": "some-repo", + }, + { + "name": "some2", + "repository": "some-repo", + }, + ], + "set_id": "1", + }, + "out_packageset": { + "package": [ + { + "name": "some", + "repository": "other-repo", + }, + { + "name": "some2", + "repository": "other-repo", + }, + ], + "set_id": "2", + }, + } + ] + } + self.assertEqual(leapp_configs.take_free_packageset_id(json_data), 3) + + def test_no_data(self): + json_data = {} + self.assertEqual(leapp_configs.take_free_packageset_id(json_data), 1) + + def test_empty_packageinfo(self): + json_data = {"packageinfo": []} + self.assertEqual(leapp_configs.take_free_packageset_id(json_data), 1) + + def test_reversed_ids(self): + json_data = { + "packageinfo": [ + { + "action": 1, + "id": "1", + "in_packageset": { + "package": [ + { + "name": "some", + "repository": "some-repo", + }, + { + "name": "some2", + "repository": "some-repo", + }, + ], + "set_id": "2", + }, + "out_packageset": { + "package": [ + { + "name": "some", + "repository": "other-repo", + }, + { + "name": "some2", + "repository": "other-repo", + }, + ], + "set_id": "1", + }, + } + ] + } + self.assertEqual(leapp_configs.take_free_packageset_id(json_data), 3) + + def test_no_in_packageset(self): + json_data = { + "packageinfo": [ + { + "action": 1, + "id": "1", + "in_packageset": None, + "out_packageset": { + "package": [ + { + "name": "some", + "repository": "other-repo", + }, + { + "name": "some2", + "repository": "other-repo", + }, + ], + "set_id": "2", + }, + } + ] + } + self.assertEqual(leapp_configs.take_free_packageset_id(json_data), 3) + + def test_no_out_packageset(self): + json_data = { + "packageinfo": [ + { + "action": 1, + "id": "1", + "in_packageset": { + "package": [ + { + "name": "some", + "repository": "other-repo", + }, + { + "name": "some2", + "repository": "other-repo", + }, + ], + "set_id": "1", + }, + "out_packageset": None, + } + ] + } + self.assertEqual(leapp_configs.take_free_packageset_id(json_data), 2) + + def test_several_packageinfos(self): + json_data = { + "packageinfo": [ + { + "action": 1, + "id": "1", + "in_packageset": { + "package": [ + { + "name": "some", + "repository": "some-repo", + }, + { + "name": "some2", + "repository": "some-repo", + }, + ], + "set_id": "1", + }, + "out_packageset": { + "package": [ + { + "name": "some", + "repository": "other-repo", + }, + { + "name": "some2", + "repository": "other-repo", + }, + ], + "set_id": "2", + }, + }, + { + "action": 1, + "id": "2", + "in_packageset": { + "package": [ + { + "name": "some", + "repository": "some-repo", + }, + { + "name": "some2", + "repository": "some-repo", + }, + ], + "set_id": "3", + }, + "out_packageset": { + "package": [ + { + "name": "some", + "repository": "other-repo", + }, + { + "name": "some2", + "repository": "other-repo", + }, + ], + "set_id": "4", + }, + } + ] + } + self.assertEqual(leapp_configs.take_free_packageset_id(json_data), 5) + + +class SetPackageMapptingTests(unittest.TestCase): + INITIAL_JSON = { + "packageinfo": [ + { + "id": "1", + "action": 1, + "in_packageset": { + "package": [ + { + "name": "some", + "repository": "some-repo", + }, + ], + "set_id": "1", + }, + }, + { + "id": "2", + "action": 4, + "in_packageset": { + "package": [ + { + "name": "other", + "repository": "some-repo", + }, + ], + "set_id": "2", + }, + "out_packageset": { + "package": [ + { + "name": "other", + "repository": "other-repo", + }, + ], + "set_id": "3", + }, + }, + { + "id": "3", + "action": 4, + "in_packageset": { + "package": [ + { + "name": "known", + "repository": "some-repo", + }, + ], + "set_id": "4", + }, + "out_packageset": { + "package": [ + { + "name": "unknown", + "repository": "other-repo", + }, + ], + "set_id": "5", + }, + }, + { + "id": "4", + "action": 4, + "in_packageset": None, + "out_packageset": None, + }, + ] + } + + JSON_FILE_PATH = "leapp_upgrade_repositories.json" + + def setUp(self): + with open(self.JSON_FILE_PATH, "w") as f: + f.write(json.dumps(self.INITIAL_JSON, indent=4)) + + def tearDown(self): + if os.path.exists(self.JSON_FILE_PATH): + os.remove(self.JSON_FILE_PATH) + + def test_add_missing_out_packageset(self): + leapp_configs.set_package_mapping("some", "some-repo", "some", "right-repo", leapp_pkgs_conf_path=self.JSON_FILE_PATH) + + with open(self.JSON_FILE_PATH) as f: + json_data = json.load(f) + self.assertEqual(json_data["packageinfo"][0]["out_packageset"]["package"][0]["name"], "some") + self.assertEqual(json_data["packageinfo"][0]["out_packageset"]["package"][0]["repository"], "right-repo") + + def test_change_existed_out_packageset(self): + leapp_configs.set_package_mapping("other", "some-repo", "other", "right-repo", leapp_pkgs_conf_path=self.JSON_FILE_PATH) + + with open(self.JSON_FILE_PATH) as f: + json_data = json.load(f) + self.assertEqual(json_data["packageinfo"][1]["out_packageset"]["package"][0]["name"], "other") + self.assertEqual(json_data["packageinfo"][1]["out_packageset"]["package"][0]["repository"], "right-repo") + + def test_replace_into_existed_out_packageset(self): + leapp_configs.set_package_mapping("known", "some-repo", "known", "right-repo", leapp_pkgs_conf_path=self.JSON_FILE_PATH) + + with open(self.JSON_FILE_PATH) as f: + json_data = json.load(f) + self.assertEqual(json_data["packageinfo"][2]["out_packageset"]["package"][0]["name"], "known") + self.assertEqual(json_data["packageinfo"][2]["out_packageset"]["package"][0]["repository"], "right-repo") + self.assertEqual(len(json_data["packageinfo"][2]["out_packageset"]["package"]), 1) + + def test_no_target_package(self): + leapp_configs.set_package_mapping("no", "some-repo", "other", "right-repo", leapp_pkgs_conf_path=self.JSON_FILE_PATH) + + with open(self.JSON_FILE_PATH) as f: + json_data = json.load(f) + self.assertEqual(json_data, self.INITIAL_JSON) + + def test_no_target_repo(self): + leapp_configs.set_package_mapping("some", "unknown-repo", "other", "right-repo", leapp_pkgs_conf_path=self.JSON_FILE_PATH) + + with open(self.JSON_FILE_PATH) as f: + json_data = json.load(f) + self.assertEqual(json_data, self.INITIAL_JSON) + + +class RemovePackageActionTests(unittest.TestCase): + INITIAL_JSON = { + "packageinfo": [ + { + "id": "1", + "action": 1, + "in_packageset": { + "package": [ + { + "name": "some", + "repository": "some-repo", + }, + ], + "set_id": "1", + }, + }, + { + "id": "2", + "action": 4, + "in_packageset": { + "package": [ + { + "name": "other", + "repository": "other-repo", + }, + ], + "set_id": "3", + }, + }, + { + "id": "3", + "action": 4, + "in_packageset": { + "package": [ + { + "name": "known", + "repository": "some-repo", + }, + ], + "set_id": "4", + }, + "out_packageset": { + "package": [ + { + "name": "unknown", + "repository": "other-repo", + }, + ], + "set_id": "5", + }, + }, + { + "id": "4", + "action": 4, + "in_packageset": None, + "out_packageset": None, + }, + { + "id": "6", + "action": 1, + "in_packageset": { + "package": [ + { + "name": "some", + "repository": "some-repo", + }, + ], + "set_id": "1", + }, + }, + ] + } + + JSON_FILE_PATH = "leapp_upgrade_repositories.json" + + def setUp(self): + with open(self.JSON_FILE_PATH, "w") as f: + f.write(json.dumps(self.INITIAL_JSON, indent=4)) + + def tearDown(self): + if os.path.exists(self.JSON_FILE_PATH): + os.remove(self.JSON_FILE_PATH) + + def test_remove_single_action(self): + leapp_configs.remove_package_action("other", "other-repo", leapp_pkgs_conf_path=self.JSON_FILE_PATH) + + with open(self.JSON_FILE_PATH) as f: + json_data = json.load(f) + self.assertTrue(all("2" != package["id"] for package in json_data["packageinfo"])) + + def test_remove_multiple_actions(self): + leapp_configs.remove_package_action("some", "some-repo", leapp_pkgs_conf_path=self.JSON_FILE_PATH) + + with open(self.JSON_FILE_PATH) as f: + json_data = json.load(f) + self.assertTrue(all("1" != package["id"] for package in json_data["packageinfo"])) + self.assertTrue(all("6" != package["id"] for package in json_data["packageinfo"])) + + def test_no_target_package(self): + leapp_configs.remove_package_action("no", "no-repo", leapp_pkgs_conf_path=self.JSON_FILE_PATH) + + with open(self.JSON_FILE_PATH) as f: + json_data = json.load(f) + self.assertEqual(json_data, self.INITIAL_JSON) + + def test_empty_json_target_package(self): + if os.path.exists(self.JSON_FILE_PATH): + os.remove(self.JSON_FILE_PATH) + with open(self.JSON_FILE_PATH, "w") as f: + f.write(json.dumps({}, indent=4)) + + leapp_configs.remove_package_action("no", "no-repo", leapp_pkgs_conf_path=self.JSON_FILE_PATH) + + with open(self.JSON_FILE_PATH) as f: + json_data = json.load(f) + self.assertEqual({}, json_data)