From 4bffbe37409e1027cf00a5a3184e8da53c823f90 Mon Sep 17 00:00:00 2001 From: xiaozai Date: Wed, 11 Dec 2024 11:39:14 +0800 Subject: [PATCH] optimize some code in some test modules; add some method for querylayerinrepo; add line breaks for README --- README.en.md | 56 ++++++++++++++++---------------- README.md | 54 +++++++++++++++--------------- src/rpmquery/querylayerinrepo.py | 24 +++++++++++--- test/test_algclassify.py | 17 ++++------ test/test_dataparse.py | 50 ++++++++++++---------------- test/test_depparse.py | 49 +++++++++++++--------------- test/test_isocheck.py | 9 +++-- test/test_repocheck.py | 7 ++-- test/test_rpmhdrinfo.py | 32 ++++++++---------- test/test_util.py | 25 +++++++------- 10 files changed, 155 insertions(+), 168 deletions(-) diff --git a/README.en.md b/README.en.md index aaa24e2..691fdb7 100644 --- a/README.en.md +++ b/README.en.md @@ -8,37 +8,37 @@ Supports x86_64 and aarch64 architecture systems #### Installation -Tool Installation: -git clone https://gitee.com/openeuler/kyclassifier -mv kyclassifier kyclassifier-1.1 -tar -czvf kyclassifier-1.1.tar.gz kyclassifier-1.1 -mkdir -p ~/rpmbuild/{BUILD,RPMS,SOURCES,SPECS,SRPMS} -mv kyclassifier-1.1.tar.gz ~/rpmbuild/SOURCES -cp kyclassifier-1.1/kyclassifier.spec ~/rpmbuild/SPECS -rpmbuild -ba ~/rpmbuild/SPECS/kyclassifier.spec -rpm -ivh ~/rpmbuild/RPMS/xxxx/kyclassifier-1.1-0.xxxx.rpm - -Install dependent packages: -pip3 install pycdlib isoparser -yum install python3-hawkey +Tool Installation:
+git clone https://gitee.com/openeuler/kyclassifier
+mv kyclassifier kyclassifier-1.1
+tar -czvf kyclassifier-1.1.tar.gz kyclassifier-1.1
+mkdir -p ~/rpmbuild/{BUILD,RPMS,SOURCES,SPECS,SRPMS}
+mv kyclassifier-1.1.tar.gz ~/rpmbuild/SOURCES
+cp kyclassifier-1.1/kyclassifier.spec ~/rpmbuild/SPECS
+rpmbuild -ba ~/rpmbuild/SPECS/kyclassifier.spec
+rpm -ivh ~/rpmbuild/RPMS/xxxx/kyclassifier-1.1-0.xxxx.rpm
+ +Install dependent packages:
+pip3 install pycdlib isoparser
+yum install python3-hawkey
#### Instructions -kyclassifier -h -usage: kyclassifier -h - -iso ISO_FILE_PATH - -repo - -local - -console_log - -q_rpminiso - -optional arguments: - -h, --help show this help message and exit - -iso ISO Input ISO file path - -repo Whether to analyze repo packages. - -local Whether to analyze local installed packages. - -console_log Output log to console. - -q_rpminiso Query input rpm layer in iso +kyclassifier -h
+usage: kyclassifier -h
+ -iso ISO_FILE_PATH
+ -repo
+ -local
+ -console_log
+ -q_rpminiso
+ +optional arguments:
+ -h, --help show this help message and exit
+ -iso ISO Input ISO file path
+ -repo Whether to analyze repo packages.
+ -local Whether to analyze local installed packages.
+ -console_log Output log to console.
+ -q_rpminiso Query input rpm layer in iso
#### Contribution diff --git a/README.md b/README.md index 7a9b470..071dc47 100644 --- a/README.md +++ b/README.md @@ -9,38 +9,38 @@ #### 安装教程 -工具安装： -git clone https://gitee.com/openeuler/kyclassifier -mv kyclassifier kyclassifier-1.1 -tar -czvf kyclassifier-1.1.tar.gz kyclassifier-1.1 -mkdir -p ~/rpmbuild/{BUILD,RPMS,SOURCES,SPECS,SRPMS} -mv kyclassifier-1.1.tar.gz ~/rpmbuild/SOURCES -cp kyclassifier-1.1/kyclassifier.spec ~/rpmbuild/SPECS -rpmbuild -ba ~/rpmbuild/SPECS/kyclassifier.spec -rpm -ivh ~/rpmbuild/RPMS/xxxx/kyclassifier-1.1-0.xxxx.rpm +工具安装：
+git clone https://gitee.com/openeuler/kyclassifier
+mv kyclassifier kyclassifier-1.1
+tar -czvf kyclassifier-1.1.tar.gz kyclassifier-1.1
+mkdir -p ~/rpmbuild/{BUILD,RPMS,SOURCES,SPECS,SRPMS}
+mv kyclassifier-1.1.tar.gz ~/rpmbuild/SOURCES
+cp kyclassifier-1.1/kyclassifier.spec ~/rpmbuild/SPECS
+rpmbuild -ba ~/rpmbuild/SPECS/kyclassifier.spec
+rpm -ivh ~/rpmbuild/RPMS/xxxx/kyclassifier-1.1-0.xxxx.rpm
-依赖安装： -yum install python3-hawkey -pip3 install pycdlib Jinja2 MarkupSafe psutil +依赖安装：
+yum install python3-hawkey
+pip3 install pycdlib Jinja2 MarkupSafe psutil
#### 使用说明 -kyclassifier -h -usage: kyclassifier -h - -iso ISO_FILE_PATH - -repo - -local - -console_log - -q_rpminiso - -optional arguments: - -h, --help show this help message and exit - -iso ISO Input ISO file path - -repo Whether to analyze repo packages. - -local Whether to analyze local installed packages. - -console_log Output log to console. - -q_rpminiso Query input rpm layer in iso +kyclassifier -h
+usage: kyclassifier -h
+ -iso ISO_FILE_PATH
+ -repo
+ -local
+ -console_log
+ -q_rpminiso
+ +optional arguments:
+ -h, --help show this help message and exit
+ -iso ISO Input ISO file path
+ -repo Whether to analyze repo packages.
+ -local Whether to analyze local installed packages.
+ -console_log Output log to console.
+ -q_rpminiso Query input rpm layer in iso
#### 参与贡献 diff --git a/src/rpmquery/querylayerinrepo.py b/src/rpmquery/querylayerinrepo.py index 9c907c3..f32dc88 100644 --- a/src/rpmquery/querylayerinrepo.py +++ b/src/rpmquery/querylayerinrepo.py @@ -14,12 +14,14 @@ """ from .rpmquery import RpmQuery - +from src.main.alglayer import AlgLayer +from src.utils.depparse import RepoDepParse +from src.utils.config import BaseConfig class QueryLayerInRepo(RpmQuery): def __init__(self,rpm): - pass + self._rpm = rpm def run(self): """入口函数 @@ -39,13 +41,25 @@ class QueryLayerInRepo(RpmQuery): def _get_repopkgs_layer(self): """获取repopkgs的分层结果 """ - pass + dep_obj = RepoDepParse() + init_f = BaseConfig.LAYERDATA + return AlgLayer.run(dep_obj, init_f) + + + @classmethod def check(self): """检查输入rpm以及配置的repo + Returns: + bool """ - pass + rpm = args[0] + iso = args[1] + if RpmCheck.check(rpm) and RepoCheck.check(): + return True + return False + @property def rpm(self): - pass \ No newline at end of file + return self._rpm diff --git a/test/test_algclassify.py b/test/test_algclassify.py index 3515289..f862da6 100644 --- a/test/test_algclassify.py +++ b/test/test_algclassify.py @@ -69,9 +69,8 @@ class TestAlgClassify(unittest.TestCase): """ if not os.path.exists(self.data_f): self.skipTest("Test File not exists, test skiped!") - else: - result = AlgClassify._load_data(self.data_f) - self.assertIsInstance(result,dict,"_load_data test failed!") + result = AlgClassify._load_data(self.data_f) + self.assertIsInstance(result,dict,"_load_data test failed!") def test_get_pkgs(self): """ @@ -135,9 +134,8 @@ class TestAlgClassify(unittest.TestCase): """ if not os.path.exists(self.data_f): self.skipTest("Test File not exists, test skiped!") - else: - result = AlgClassify._get_pkg2category_by_jsonf(self.data_f) - self.assertIsInstance(result,dict,"_get_pkg2category_by_jsonf test failed!") + result = AlgClassify._get_pkg2category_by_jsonf(self.data_f) + self.assertIsInstance(result,dict,"_get_pkg2category_by_jsonf test failed!") def test_merge_pkg2category_dict(self): """ @@ -164,9 +162,8 @@ class TestAlgClassify(unittest.TestCase): """ if not os.path.exists(self.data_f): self.skipTest("Test File not exists, test skiped!") - else: - result = AlgClassify.run(self.data_obj,self.data_f) - self.assertIsInstance(result,dict,"run test failed!") + result = AlgClassify.run(self.data_obj,self.data_f) + self.assertIsInstance(result,dict,"run test failed!") if __name__ == "__main__": - unittest.main(verbosity=2) \ No newline at end of file + unittest.main(verbosity=2) diff --git a/test/test_dataparse.py b/test/test_dataparse.py index ecf9c64..0be5921 100644 --- a/test/test_dataparse.py +++ b/test/test_dataparse.py @@ -65,9 +65,8 @@ class TestDataParse(unittest.TestCase): """ if not self._init_iso_dataparse(): self.skipTest("Init iso_dataparse failed, test skiped!") - else: - result = self.iso_dataparse.get_pkgname_set(self.files_path) - self.assertIsInstance(result, set, "get_pkgname_set test failed!") + result = self.iso_dataparse.get_pkgname_set(self.files_path) + self.assertIsInstance(result, set, "get_pkgname_set test failed!") def test_iso_get_pkgsinfo_list(self): """ @@ -77,9 +76,8 @@ class TestDataParse(unittest.TestCase): """ if not self._init_iso_dataparse(): self.skipTest("Init iso_dataparse failed, test skiped!") - else: - result = self.iso_dataparse.get_pkgsinfo_list(self.files_path) - self.assertIsInstance(result, list, "get_pkgsinfo_list test failed!") + result = self.iso_dataparse.get_pkgsinfo_list(self.files_path) + self.assertIsInstance(result, list, "get_pkgsinfo_list test failed!") def test_iso_get_pkginfo_byname(self): """ @@ -91,10 +89,9 @@ class TestDataParse(unittest.TestCase): """ if not self._init_iso_dataparse(): self.skipTest("Init iso_dataparse failed, test skiped!") - else: - pkgname = 'kernel' - result = self.iso_dataparse.get_pkginfo_byname(pkgname) - self.assertIsInstance(result, list, "get_pkginfo_byname test failed!") + pkgname = 'kernel' + result = self.iso_dataparse.get_pkginfo_byname(pkgname) + self.assertIsInstance(result, list, "get_pkginfo_byname test failed!") def test__list2dict(self): """ @@ -119,9 +116,8 @@ class TestDataParse(unittest.TestCase): """ if not RepoCheck.check(): self.skipTest("Repo check failed, test skiped!") - else: - result = RepoDataParse._load_data() - self.assertIsInstance(result,list,"_load_data test failed!") + result = RepoDataParse._load_data() + self.assertIsInstance(result,list,"_load_data test failed!") def test_repo_get_pkgname_set(self): """ @@ -131,9 +127,8 @@ class TestDataParse(unittest.TestCase): """ if not RepoCheck.check(): self.skipTest("Repo check failed, test skiped!") - else: - result = RepoDataParse.get_pkgname_set() - self.assertIsInstance(result,set,"get_pkgname_set test failed!") + result = RepoDataParse.get_pkgname_set() + self.assertIsInstance(result,set,"get_pkgname_set test failed!") def test_repo_get_pkgsinfo_list(self): """ @@ -143,9 +138,8 @@ class TestDataParse(unittest.TestCase): """ if not RepoCheck.check(): self.skipTest("Repo check failed, test skiped!") - else: - result = RepoDataParse.get_pkgsinfo_list() - self.assertIsInstance(result,list,"get_pkgsinfo_list test failed!") + result = RepoDataParse.get_pkgsinfo_list() + self.assertIsInstance(result,list,"get_pkgsinfo_list test failed!") def test_repo_list2dict(self): """ @@ -155,12 +149,11 @@ class TestDataParse(unittest.TestCase): """ if not RepoCheck.check(): self.skipTest("Repo check failed, test skiped!") - else: - dict_list = [{'pkgname':'aaa'}, - {'pkgname':'bbb'}] - key = 'pkgname' - result = RepoDataParse._list2dict(dict_list,key) - self.assertIsInstance(result,dict,"_list2dict test failed!") + dict_list = [{'pkgname':'aaa'}, + {'pkgname':'bbb'}] + key = 'pkgname' + result = RepoDataParse._list2dict(dict_list,key) + self.assertIsInstance(result,dict,"_list2dict test failed!") def test_local_get_pkgname_set(self): """ @@ -188,9 +181,8 @@ class TestDataParse(unittest.TestCase): """ if not self._init_local_dataparse(): self.skipTest("Init local_dataparse failed, test skiped!") - else: - result = LocalInstalledDataParse.get_os_vendor(self.local_dataparse.pkgs_info) - self.assertIsInstance(result,str,"get_os_vendor test failed!") + result = LocalInstalledDataParse.get_os_vendor(self.local_dataparse.pkgs_info) + self.assertIsInstance(result,str,"get_os_vendor test failed!") def test_local_list2dict(self): """ @@ -234,4 +226,4 @@ class TestDataParse(unittest.TestCase): self.assertDictEqual(result,expect_dict,"_list2dict test failed!") if __name__ == "__main__": - unittest.main(verbosity=2) \ No newline at end of file + unittest.main(verbosity=2) diff --git a/test/test_depparse.py b/test/test_depparse.py index f63570f..44f040b 100644 --- a/test/test_depparse.py +++ b/test/test_depparse.py @@ -53,15 +53,14 @@ class TestDepParse(unittest.TestCase): Returns: dict """ - if self.is_exists: - if not self._init_iso_depparse(): - self.skipTest("Init iso_depparse failed, test skiped!") - else: - result = self.iso_depparse._get_repo_pkg_deps() - self.assertIsInstance(result,dict,"iso_depparse._get_repo_pkg_deps test failed!") - else: + if not self.is_exists: self.skipTest("_get_repo_pkg_deps test skiped!") - + if not self._init_iso_depparse(): + self.skipTest("Init iso_depparse failed, test skiped!") + result = self.iso_depparse._get_repo_pkg_deps() + self.assertIsInstance(result,dict,"iso_depparse._get_repo_pkg_deps test failed!") + + def test_iso_get_repo_pkg_deps_by(self): """ Test class ISODepParse method _get_repo_pkg_deps_by() @@ -70,9 +69,8 @@ class TestDepParse(unittest.TestCase): """ if not self._init_iso_depparse(): self.skipTest("Init iso_depparse failed, test skiped!") - else: - result = self.iso_depparse._get_repo_pkg_deps_by() - self.assertIsInstance(result,dict,"iso_depparse._get_repo_pkg_deps_by test failed!") + result = self.iso_depparse._get_repo_pkg_deps_by() + self.assertIsInstance(result,dict,"iso_depparse._get_repo_pkg_deps_by test failed!") def test_iso_get_all_pkgs(self): """ @@ -80,14 +78,12 @@ class TestDepParse(unittest.TestCase): Returns: set """ - if self.is_exists: - if not self._init_iso_depparse(): - self.skipTest("Init iso_depparse failed, test skiped!") - else: - result = self.iso_depparse._get_all_pkgs() - self.assertIsInstance(result,set,"iso_depparse._get_all_pkgs test failed!") - else: + if not self.is_exists: self.skipTest("_get_all_pkgs test skiped!") + if not self._init_iso_depparse(): + self.skipTest("Init iso_depparse failed, test skiped!") + result = self.iso_depparse._get_all_pkgs() + self.assertIsInstance(result,set,"iso_depparse._get_all_pkgs test failed!") def _init_repo_depparse(self): """ @@ -107,15 +103,14 @@ class TestDepParse(unittest.TestCase): Returns: list """ - if self.is_exists: - if not RepoCheck: - self.skipTest("Repo check failed, test skiped!") - if not self._init_repo_depparse(): - self.skipTest("RepoDepParse obj init failed, test skiped!") - result = self.repo_depparse._load_data() - self.assertIsInstance(result,list,"repo_depparse._load_data test failed!") - else: + if not self.is_exists: self.skipTest("_load_data test skiped!") + if not RepoCheck: + self.skipTest("Repo check failed, test skiped!") + if not self._init_repo_depparse(): + self.skipTest("RepoDepParse obj init failed, test skiped!") + result = self.repo_depparse._load_data() + self.assertIsInstance(result,list,"repo_depparse._load_data test failed!") def test_repo_get_all_pkgs(self): """ @@ -145,4 +140,4 @@ class TestDepParse(unittest.TestCase): if __name__ == "__main__": - unittest.main(verbosity=2) \ No newline at end of file + unittest.main(verbosity=2) diff --git a/test/test_isocheck.py b/test/test_isocheck.py index eff85c0..89a991c 100644 --- a/test/test_isocheck.py +++ b/test/test_isocheck.py @@ -48,12 +48,11 @@ class TestIsoCheck(unittest.TestCase): Returns: bool """ - if self.iso_check.check_exist(): - result = self.iso_check.check_format() - self.assertIn(result, [True, False], "check_format test failed!") - else: + if not self.iso_check.check_exist(): self.skipTest("iso file not exist,skip test check_format") + result = self.iso_check.check_format() + self.assertIn(result, [True, False], "check_format test failed!") if __name__ == "__main__": - unittest.main(verbosity=2) \ No newline at end of file + unittest.main(verbosity=2) diff --git a/test/test_repocheck.py b/test/test_repocheck.py index 44fd30f..1d34405 100644 --- a/test/test_repocheck.py +++ b/test/test_repocheck.py @@ -72,11 +72,10 @@ class TestRepoCheck(unittest.TestCase): "baseurl": "http://url/path/updates/arch" } ] - if os.path.exists(self.repo_path): - result = self.repo_check._create_repofile(repos_l) - self.assertIn(result, [True, False], "_create_repofile test failed!") - else: + if not os.path.exists(self.repo_path): self.skipTest("_create_repofile test skiped!") + result = self.repo_check._create_repofile(repos_l) + self.assertIn(result, [True, False], "_create_repofile test failed!") def test_move_repofiles(self): diff --git a/test/test_rpmhdrinfo.py b/test/test_rpmhdrinfo.py index e9e4c52..63687e7 100644 --- a/test/test_rpmhdrinfo.py +++ b/test/test_rpmhdrinfo.py @@ -58,9 +58,8 @@ class TestRpmHdrInfo(unittest.TestCase): """ if not self._init_alglayer(): self.skipTest("Init RpmHdrInfo failed, test skiped!") - else: - result = self.rpmhdrinfo.as_dict() - self.assertIsInstance(result, dict, "as_dict test failed!") + result = self.rpmhdrinfo.as_dict() + self.assertIsInstance(result, dict, "as_dict test failed!") def test_epoch(self): """ @@ -70,9 +69,8 @@ class TestRpmHdrInfo(unittest.TestCase): """ if not self._init_alglayer(): self.skipTest("Init RpmHdrInfo failed, test skiped!") - else: - result = self.rpmhdrinfo.as_dict() - self.assertIsInstance(result, dict, "epoch test failed!") + result = self.rpmhdrinfo.as_dict() + self.assertIsInstance(result, dict, "epoch test failed!") def test_vendor(self): """ @@ -82,9 +80,8 @@ class TestRpmHdrInfo(unittest.TestCase): """ if not self._init_alglayer(): self.skipTest("Init RpmHdrInfo failed, test skiped!") - else: - result = self.rpmhdrinfo.as_dict() - self.assertIsInstance(result, dict, "vendor test failed!") + result = self.rpmhdrinfo.as_dict() + self.assertIsInstance(result, dict, "vendor test failed!") def test_group(self): """ @@ -94,9 +91,8 @@ class TestRpmHdrInfo(unittest.TestCase): """ if not self._init_alglayer(): self.skipTest("Init RpmHdrInfo failed, test skiped!") - else: - result = self.rpmhdrinfo.as_dict() - self.assertIsInstance(result, dict, "group test failed!") + result = self.rpmhdrinfo.as_dict() + self.assertIsInstance(result, dict, "group test failed!") def test_rpm_sourcerpm(self): """ @@ -106,9 +102,8 @@ class TestRpmHdrInfo(unittest.TestCase): """ if not self._init_alglayer(): self.skipTest("Init RpmHdrInfo failed, test skiped!") - else: - result = self.rpmhdrinfo.as_dict() - self.assertIsInstance(result, dict, "rpm_sourcerpm test failed!") + result = self.rpmhdrinfo.as_dict() + self.assertIsInstance(result, dict, "rpm_sourcerpm test failed!") def test_requires(self): """ @@ -118,8 +113,7 @@ class TestRpmHdrInfo(unittest.TestCase): """ if not self._init_alglayer(): self.skipTest("Init RpmHdrInfo failed, test skiped!") - else: - result = self.rpmhdrinfo.as_dict() - self.assertIsInstance(result, dict, "requires test failed!") + result = self.rpmhdrinfo.as_dict() + self.assertIsInstance(result, dict, "requires test failed!") if __name__ == "__main__": - unittest.main(verbosity=2) \ No newline at end of file + unittest.main(verbosity=2) diff --git a/test/test_util.py b/test/test_util.py index 9eb11b8..1e303d6 100644 --- a/test/test_util.py +++ b/test/test_util.py @@ -57,30 +57,27 @@ class TestUtil(unittest.TestCase): """Test static method parse_iso_repodata(iso_path,repodata_dir='/opt/kyclassifier/iso_parse/repodata') """ repodata_dir='/opt/kyclassifier/iso_parse/repodata' - if os.path.exists(self.iso_path): - self.iso_utils.parse_iso_repodata(self.iso_path) - self.assertTrue(os.path.exists(repodata_dir), "parse_iso_repodata test failed!") - else: + if not os.path.exists(self.iso_path): self.skipTest("trans_set2list test skiped!") + self.iso_utils.parse_iso_repodata(self.iso_path) + self.assertTrue(os.path.exists(repodata_dir), "parse_iso_repodata test failed!") def test_get_repo_from_dir(self): """Test static method get_repo_from_dir(repodata_dir='/opt/kyclassifier/iso_parse/repodata') """ repodata_dir='/opt/kyclassifier/iso_parse/repodata' - if os.path.exists(repodata_dir): - repomd_fn, primary_fn, filelists_fn = self.iso_utils.get_repo_from_dir(repodata_dir) - result = all(['repomd.xml' in repomd_fn, 'primary.xml' in primary_fn, 'filelists.xml' in filelists_fn]) - self.assertTrue(result, "get_repo_from_dir test failed!") - else: + if not os.path.exists(repodata_dir): self.skipTest("get_repo_from_dir test skiped!") + repomd_fn, primary_fn, filelists_fn = self.iso_utils.get_repo_from_dir(repodata_dir) + result = all(['repomd.xml' in repomd_fn, 'primary.xml' in primary_fn, 'filelists.xml' in filelists_fn]) + self.assertTrue(result, "get_repo_from_dir test failed!") def test_parase_iso_repofile(self): """Test class method parase_iso_repofile(iso_path,target_dir='/opt/kyclassifier/iso_parse/repodata') """ target_dir='/opt/kyclassifier/iso_parse/repodata' - if os.path.exists(target_dir) and os.path.exists(self.iso_path): - repomd_fn, primary_fn, filelists_fn = self.iso_utils.parase_iso_repofile(self.iso_path) - result = all(['repomd.xml' in repomd_fn, 'primary.xml' in primary_fn, 'filelists.xml' in filelists_fn]) - self.assertTrue(result, "parase_iso_repofile test failed!") - else: + if not os.path.exists(target_dir) or not os.path.exists(self.iso_path): self.skipTest("parase_iso_repofile test skiped!") + repomd_fn, primary_fn, filelists_fn = self.iso_utils.parase_iso_repofile(self.iso_path) + result = all(['repomd.xml' in repomd_fn, 'primary.xml' in primary_fn, 'filelists.xml' in filelists_fn]) + self.assertTrue(result, "parase_iso_repofile test failed!") -- Gitee