提交 97c95cfa 编写于 作者: G gongzt

Improvement of code defects

......@@ -3,7 +3,7 @@
Description: Interception before request
"""
from flask import request
from packageship.application import OPERATION
from packageship import application
from packageship.application.apps.package.url import urls
......@@ -20,8 +20,8 @@ def identity_verification():
if request.url_rule:
url_rule = request.url_rule.rule
for view, url, authentication in urls:
if url == url_rule and OPERATION in authentication.keys():
if request.method not in authentication.get(OPERATION):
if url == url_rule and application.OPERATION in authentication.keys():
if request.method not in authentication.get(application.OPERATION):
return False
break
return True
......
from flask.blueprints import Blueprint
from flask_restful import Api
from packageship.application.apps.package.url import urls
from packageship.application import OPERATION
from packageship import application
package = Blueprint('package', __name__)
......@@ -9,7 +9,7 @@ package = Blueprint('package', __name__)
api = Api()
for view, url, operation in urls:
if OPERATION and OPERATION in operation.keys():
if application.OPERATION and application.OPERATION in operation.keys():
api.add_resource(view, url)
......
......@@ -119,7 +119,7 @@ class Packages(Resource):
class SinglePack(Resource):
"""
description: single package management
Restful API: getput
Restful API: get, put
ChangeLog:
"""
......@@ -273,14 +273,24 @@ class InstallDepend(Resource):
"""
Description: Query a package's install depend(support
querying in one or more databases)
Args:
binaryName:name of the bin package
dbPreority:the array for database preority
Returns:
{
"code": "",
"data": "",
"msg": ""
input:
binaryName
dbPreority: the array for database preority
return:
resultDict{
binary_name: //binary package name
[
src, //the source package name for
that binary packge
dbname,
version,
[
parent_node, //the binary package name which is
the install depend for binaryName
type //install install or build, which
depend on the function
]
]
}
Raises:
"""
......@@ -557,8 +567,8 @@ class Repodatas(Resource):
file_context.read(), Loader=yaml.FullLoader)
if init_database_date is None:
raise ContentNoneException(
"The content of the database initialization configuration "
"file cannot be empty")
"The content of the database initialization configuration "
"file cannot be empty ")
init_database_date.sort(
key=lambda x: x['priority'], reverse=False)
return jsonify(
......
......@@ -22,8 +22,7 @@ from packageship.application.models.temporarydb import src_requires
from packageship.application.models.temporarydb import bin_package
from packageship.application.models.temporarydb import bin_requiresment
from packageship.application.models.temporarydb import bin_provides
from packageship.system_config import DATABASE_FILE_INFO
from packageship.system_config import DATABASE_FOLDER_PATH
from packageship import system_config
LOGGER = Log(__name__)
......@@ -615,9 +614,9 @@ class InitDataBase():
IOError: File or network operation io abnormal
"""
try:
if not os.path.exists(DATABASE_FILE_INFO):
pathlib.Path(DATABASE_FILE_INFO).touch()
with open(DATABASE_FILE_INFO, 'a+', encoding='utf8') as file_context:
if not os.path.exists(system_config.DATABASE_FILE_INFO):
pathlib.Path(system_config.DATABASE_FILE_INFO).touch()
with open(system_config.DATABASE_FILE_INFO, 'a+', encoding='utf8') as file_context:
setting_content = []
if 'database_content' in Kwargs.keys():
content = Kwargs.get('database_content')
......@@ -643,8 +642,8 @@ class InitDataBase():
"""
try:
if os.path.exists(DATABASE_FILE_INFO):
os.remove(DATABASE_FILE_INFO)
if os.path.exists(system_config.DATABASE_FILE_INFO):
os.remove(system_config.DATABASE_FILE_INFO)
except (IOError, Error) as exception_msg:
LOGGER.logger.error(exception_msg)
return False
......@@ -670,14 +669,14 @@ class InitDataBase():
if del_result:
try:
file_read = open(DATABASE_FILE_INFO, 'r', encoding='utf-8')
file_read = open(system_config.DATABASE_FILE_INFO, 'r', encoding='utf-8')
_databases = yaml.load(
file_read.read(), Loader=yaml.FullLoader)
for database in _databases:
if database.get('database_name') == db_name:
_databases.remove(database)
# Delete the successfully imported database configuration node
with open(DATABASE_FILE_INFO, 'w+', encoding='utf-8') as file_context:
with open(system_config.DATABASE_FILE_INFO, 'w+', encoding='utf-8') as file_context:
yaml.safe_dump(_databases, file_context)
except (IOError, Error) as del_config_error:
LOGGER.logger.error(del_config_error)
......@@ -858,7 +857,7 @@ class SqliteDatabaseOperations():
self.database_file_folder = self._read_config.get_system(
'data_base_path')
if not self.database_file_folder:
self.database_file_folder = DATABASE_FOLDER_PATH
self.database_file_folder = system_config.DATABASE_FOLDER_PATH
if not os.path.exists(self.database_file_folder):
try:
......
Name: pkgship
Version: 1.0
Release: 1
Release: 2
Summary: Pkgship implements rpm package dependence ,maintainer, patch query and so no.
License: Mulan 2.0
URL: https://gitee.com/openeuler/openEuler-Advisor
Source0: https://gitee.com/openeuler/openEuler-Advisor/pkgship-%{version}.tar
Source0: https://gitee.com/openeuler/openEuler-Advisor/pkgship-%{version}.tar.gz
BuildArch: noarch
Requires: python3-pip python3-flask-restful python3-flask python3.7 python3-pyyaml
Requires: python3-pip python3-flask-restful python3-flask python3 python3-pyyaml
Requires: python3-sqlalchemy python3-prettytable python3-requests
#Requires: pyinstaller python3-flask-session python3-flask-script marshmallow uwsig
......@@ -30,13 +30,21 @@ Pkgship implements rpm package dependence ,maintainer, patch query and so no.
%post
#build cli bin
if [ -f "/usr/bin/pkgship" ];then
if [ -f "/usr/bin/pkgship" ]; then
rm -rf /usr/bin/pkgship
fi
cd %{python3_sitelib}/packageship/
/usr/local/bin/pyinstaller -F pkgship.py
if [ -f "/usr/bin/pyinstaller" ]; then
/usr/bin/pyinstaller -F pkgship.py
elif [ -f "/usr/local/bin/pyinstaller" ]; then
/usr/local/bin/pyinstaller -F pkgship.py
else
echo "pkship install fail,there is no pyinstaller!"
exit
fi
sed -i "s/hiddenimports\=\[\]/hiddenimports\=\['pkg_resources.py2_warn'\]/g" pkgship.spec
/usr/local/bin/pyinstaller pkgship.spec
cp dist/pkgship /usr/bin/
......@@ -53,6 +61,9 @@ rm -rf %{python3_sitelib}/packageship/build %{python3_sitelib}/packageship/dist
%changelog
* Sat JUL 4 2020 Yiru Wang <wangyiru1@huawei.com> - 1.0-2
- cheange requires python3.7 to python3,add check pyinstaller file.
* Tue JUN 30 2020 Yiru Wang <wangyiru1@huawei.com> - 1.0-1
- add pkgshipd file
......
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
Compare the values in two Python data types for equality, ignoring the order of values
......@@ -38,7 +39,7 @@ def get_correct_json_by_filename(filename):
"test",
"common_files",
"correct_test_result_json",
f"{filename}.json")
"{}.json".format(filename))
try:
with open(json_path, "r") as json_fp:
correct_list = json.loads(json_fp.read())
......
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
InitConf
"""
import os
from configparser import ConfigParser
from packageship import system_config
......@@ -7,6 +10,10 @@ import yaml
class InitConf:
"""
InitConf
"""
def __init__(self):
base_path = os.path.join(os.path.dirname(system_config.BASE_PATH),
"test",
......@@ -23,8 +30,10 @@ class InitConf:
origin_yaml = yaml.load(f.read(), Loader=yaml.FullLoader)
for index, obj in enumerate(origin_yaml, 1):
src_path = os.path.join(base_path, "db_origin", f"data_{str(index)}_src.sqlite")
bin_path = os.path.join(base_path, "db_origin", f"data_{str(index)}_bin.sqlite")
src_path = os.path.join(base_path, "db_origin",
"data_{}_src.sqlite".format(str(index)))
bin_path = os.path.join(base_path, "db_origin",
"data_{}_bin.sqlite".format(str(index)))
obj["src_db_file"] = [src_path]
obj["bin_db_file"] = [bin_path]
with open(conf_path, 'w', encoding='utf-8') as w_f:
......
#!/usr/bin/python3
"""
Inherited from unittest.TestResult,
The simple statistical function is realized.
"""
import sys
import unittest
class MyTestResult(unittest.TestResult):
"""
Inherited from unittest.TestResult,
The simple statistical function is realized.
"""
def __init__(self, verbosity=0):
super(MyTestResult, self).__init__()
self.success_case_count = 0
self.err_case_count = 0
self.failure_case_count = 0
self.verbosity = verbosity
def addSuccess(self, test):
"""When the use case is executed successfully"""
self.success_case_count += 1
super(MyTestResult, self).addSuccess(test)
if self.verbosity > 0:
sys.stderr.write('Success ')
sys.stderr.write(str(test))
sys.stderr.write('\n')
def addError(self, test, err):
"""When a code error causes a use case to fail"""
self.err_case_count += 1
super(MyTestResult, self).addError(test, err)
if self.verbosity > 0:
sys.stderr.write('Error ')
sys.stderr.write(str(test)+'\n')
_,err_info = self.errors[-1]
sys.stderr.write(err_info)
sys.stderr.write('\n')
def addFailure(self, test, err):
"""When the assertion is false"""
self.failure_case_count += 1
super(MyTestResult, self).addFailure(test, err)
if self.verbosity > 0:
sys.stderr.write('Failure ')
sys.stderr.write(str(test)+'\n')
_, err_info = self.failures[-1]
sys.stderr.write(err_info)
sys.stderr.write('\n')
class MyTestRunner():
"""
Run All TestCases
"""
def __init__(self, verbosity=0):
self.verbosity = verbosity
def run(self, test):
"""run MyTestResult and return result"""
result = MyTestResult(self.verbosity)
test(result)
return result
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
OperateTestBase
"""
import os
import unittest
from packageship.libs.exception import Error
try:
from packageship import system_config
......@@ -21,14 +26,14 @@ try:
from test.base_code.init_config_path import init_config
from packageship.manage import app
except Exception as e:
except Error:
raise
class OperateTestBase(unittest.TestCase):
"""
OperateTestBase
"""
def setUp(self) -> None:
def setUp(self):
self.client = app.test_client()
#!/usr/bin/python3
# -*- coding:utf-8 -*-
import os
import unittest
from packageship.libs.exception import Error
try:
from packageship import system_config
......@@ -22,10 +24,10 @@ try:
from test.base_code.init_config_path import init_config
from packageship.selfpkg import app
except Exception as e:
except Error:
raise
class ReadTestBase(unittest.TestCase):
def setUp(self) -> None:
def setUp(self):
self.client = app.test_client()
# -*- coding:utf-8 -*-
"""Test case of init data"""
from test.test_module.init_system_tests.test_importdata import test_import_data_suit
def execute_init():
"""Test case of init data"""
test_import_data_suit()
if __name__ == '__main__':
execute_init()
# -*- coding:utf-8 -*-
"""
Test case of reading data
"""
from test.test_module.dependent_query_tests.test_build_depend import test_build_depend_suit
from test.test_module.packages_tests.test_packages import test_packages_suit
from test.test_module.single_package_tests.test_get_singlepack import test_get_single_package_suit
from test.test_module.repodatas_test.test_get_repodatas import test_get_repodatas_suit
from test.test_module.dependent_query_tests.test_be_depend import test_be_depend_suit
from test.test_module.dependent_query_tests.test_install_depend import test_install_depend_suit
from test.test_module.dependent_query_tests.test_self_depend import test_self_depend_suit
def execute_read():
"""Test case of reading data"""
test_build_depend_suit()
test_packages_suit()
test_get_single_package_suit()
test_get_repodatas_suit()
test_be_depend_suit()
test_install_depend_suit()
test_self_depend_suit()
if __name__ == '__main__':
execute_read()
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
Execute all test cases
"""
import unittest
import datetime
from test.base_code.my_test_runner import MyTestRunner
RUNNER = MyTestRunner(verbosity=1)
def import_data_tests():
"""Initialize related test cases"""
from test.test_module.init_system_tests.test_importdata import ImportData
suite = unittest.TestSuite()
suite.addTests(unittest.TestLoader().loadTestsFromTestCase(ImportData))
return RUNNER.run(suite)
def read_data_tests():
"""Test cases with read operations on data"""
from test.test_module.dependent_query_tests.test_install_depend import TestInstallDepend
from test.test_module.dependent_query_tests.test_self_depend import TestSelfDepend
from test.test_module.dependent_query_tests.test_be_depend import TestBeDepend
from test.test_module.repodatas_test.test_get_repodatas import TestGetRepodatas
from test.test_module.dependent_query_tests.test_build_depend import TestBuildDepend
from test.test_module.packages_tests.test_packages import TestPackages
from test.test_module.single_package_tests.test_get_singlepack import TestGetSinglePack
suite = unittest.TestSuite()
classes = [TestInstallDepend, TestSelfDepend, TestBeDepend,
TestGetRepodatas, TestBuildDepend, TestPackages, TestGetSinglePack]
for cls in classes:
suite.addTests(unittest.TestLoader().loadTestsFromTestCase(cls))
return RUNNER.run(suite)
def write_data_tests():
"""Test cases with write operations on data"""
from test.test_module.repodatas_test.test_delete_repodatas import TestDeleteRepodatas
from test.test_module.single_package_tests.test_update_singlepack import TestUpdatePackage
suite = unittest.TestSuite()
suite.addTests(unittest.TestLoader().loadTestsFromTestCase(TestDeleteRepodatas))
suite.addTests(unittest.TestLoader().loadTestsFromTestCase(TestUpdatePackage))
return RUNNER.run(suite)
def main():
"""Test case execution entry function"""
start_time = datetime.datetime.now()
result_4_import = import_data_tests()
result_4_read = read_data_tests()
result_4_write = write_data_tests()
stop_time = datetime.datetime.now()
print('\nA total of %s test cases were run: \nsuccessful:%s\tfailed:%s\terror:%s\n' % (
int(result_4_import.testsRun) + int(result_4_read.testsRun) + int(result_4_write.testsRun),
int(
result_4_import.success_case_count
) + int(result_4_read.success_case_count) + int(result_4_write.success_case_count),
int(
result_4_import.failure_case_count
) + int(result_4_read.failure_case_count) + int(result_4_write.failure_case_count),
int(
result_4_import.err_case_count
) + int(result_4_read.err_case_count) + int(result_4_write.err_case_count)
))
print('Total Time: %s' % (stop_time - start_time))
main()
# -*- coding:utf-8 -*-
"""Test case of writing data"""
from test.test_module.repodatas_test.test_delete_repodatas import test_delete_repodatas_suit
from test.test_module.single_package_tests.test_update_singlepack import test_updata_single_package_suit
def execute_operate():
"""Test case of writing data"""
test_updata_single_package_suit()
test_delete_repodatas_suit()
if __name__ == '__main__':
execute_operate()
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
Less transmission is always parameter transmission
......@@ -8,10 +9,11 @@ from test.base_code.read_data_base import ReadTestBase
from test.base_code.common_test_code import compare_two_values, get_correct_json_by_filename
from packageship.application.apps.package.function.constants import ResponseCode
class TestBeDepend(ReadTestBase):
'''
"""
The dependencies of the package are tested
'''
"""
def test_lack_parameter(self):
"""
......@@ -355,18 +357,3 @@ class TestBeDepend(ReadTestBase):
resp_dict = json.loads(resp.data)
self.assertTrue(compare_two_values(output_for_input, resp_dict),
msg="The answer is not correct")
def test_be_depend_suit():
"""
Start the test case function
"""
print("---TestBeDepend START---")
suite = unittest.TestSuite()
suite.addTest(TestBeDepend("test_lack_parameter"))
suite.addTest(TestBeDepend("test_wrong_parameter"))
suite.addTest(TestBeDepend("test_true_params_result"))
unittest.TextTestRunner().run(suite)
# if __name__ == '__main__':
# unittest.main()
#!/usr/bin/python3
# -*- coding:utf-8 -*-
'''
"""
build_depend unittest
'''
"""
import json
import unittest
......@@ -11,15 +12,15 @@ from packageship.application.apps.package.function.constants import ResponseCode
class TestBuildDepend(ReadTestBase):
'''
"""
class for test build_depend
'''
"""
def test_empty_source_name_dblist(self):
'''
"""
test empty parameters:sourceName,dbList
:return:
'''
"""
resp = self.client.post("/packages/findBuildDepend",
data="{}",
content_type="application/json")
......@@ -57,10 +58,10 @@ class TestBuildDepend(ReadTestBase):
self.assertIsNotNone(resp_dict.get("data"), msg="Error in data information return")
def test_wrong_source_name_dblist(self):
'''
"""
test wrong parameters:sourceName,dbList
:return: None
'''
"""
resp = self.client.post("/packages/findBuildDepend",
data=json.dumps({"sourceName": 0}),
content_type="application/json")
......@@ -136,6 +137,11 @@ class TestBuildDepend(ReadTestBase):
self.assertIsNone(resp_dict.get("data"), msg="Error in data information return")
def test_true_params_result(self):
"""
test_true_params_result
Returns:
"""
correct_list = get_correct_json_by_filename("build_depend")
self.assertNotEqual([], correct_list, msg="Error reading JSON file")
......@@ -151,14 +157,5 @@ class TestBuildDepend(ReadTestBase):
msg="The answer is not correct")
def test_build_depend_suit():
print("---TestBuildDepend START---")
suite = unittest.TestSuite()
suite.addTest(TestBuildDepend("test_empty_source_name_dblist"))
suite.addTest(TestBuildDepend("test_wrong_source_name_dblist"))
suite.addTest(TestBuildDepend("test_true_params_result"))
unittest.TextTestRunner().run(suite)
if __name__ == '__main__':
unittest.main()
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
TestInstallDepend
"""
import unittest
import json
......@@ -8,10 +12,16 @@ from packageship.application.apps.package.function.constants import ResponseCode
class TestInstallDepend(ReadTestBase):
"""
TestInstallDepend
"""
def test_empty_binaryName_dbList(self):
"""
test_empty_binaryName_dbList
Returns:
"""
resp = self.client.post("/packages/findInstallDepend",
data="{}",
content_type="application/json")
......@@ -49,6 +59,11 @@ class TestInstallDepend(ReadTestBase):
self.assertIsNotNone(resp_dict.get("data"), msg="Error in data information return")
def test_wrong_binaryName_dbList(self):
"""
test_empty_binaryName_dbList
Returns:
"""
resp = self.client.post("/packages/findInstallDepend",
data=json.dumps({"binaryName": 0}),
content_type="application/json")
......@@ -125,6 +140,11 @@ class TestInstallDepend(ReadTestBase):
self.assertIsNone(resp_dict.get("data"), msg="Error in data information return")
def test_true_params_result(self):
"""
test_empty_binaryName_dbList
Returns:
"""
correct_list = get_correct_json_by_filename("install_depend")
self.assertNotEqual([], correct_list, msg="Error reading JSON file")
......@@ -138,12 +158,7 @@ class TestInstallDepend(ReadTestBase):
resp_dict = json.loads(resp.data)
self.assertTrue(compare_two_values(output_for_input, resp_dict),
msg="The answer is not correct")
def test_install_depend_suit():
suite = unittest.TestSuite()
suite.addTest(TestInstallDepend("test_empty_binaryName_dbList"))
suite.addTest(TestInstallDepend("test_wrong_binaryName_dbList"))
suite.addTest(TestInstallDepend("test_true_params_result"))
unittest.TextTestRunner().run(suite)
if __name__ == '__main__':
unittest.main()
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
TestSelfDepend
"""
import unittest
import json
......@@ -9,8 +13,16 @@ from packageship.application.apps.package.function.searchdb import db_priority
class TestSelfDepend(ReadTestBase):
"""
TestSelfDepend
"""
def test_empty_parameter(self):
"""
test_empty_parameter
Returns:
"""
resp = self.client.post("/packages/findSelfDepend",
data='{}',
content_type="application/json")
......@@ -116,6 +128,11 @@ class TestSelfDepend(ReadTestBase):
self.assertIsNotNone(resp_dict.get("data"), msg="Data return error!")
def test_wrong_parameter(self):
"""
test_wrong_parameter
Returns:
"""
resp = self.client.post("/packages/findSelfDepend",
data=json.dumps({
"packagename": "wukong"
......@@ -248,6 +265,11 @@ class TestSelfDepend(ReadTestBase):
self.assertIsNone(resp_dict.get("data"), msg="Data return error!")
def test_true_params_result(self):
"""
test_true_params_result
Returns:
"""
correct_list = get_correct_json_by_filename("self_depend")
self.assertNotEqual([], correct_list, msg="Error reading JSON file")
......@@ -259,22 +281,8 @@ class TestSelfDepend(ReadTestBase):
content_type="application/json")
output_for_input = correct_data["output"]
resp_dict = json.loads(resp.data)
if not compare_two_values(output_for_input, resp_dict):
from pprint import pprint
print(input_value)
# pprint(output_for_input)
pprint(resp_dict)
# self.assertTrue(compare_two_values(output_for_input, resp_dict),
# msg="The answer is not correct")
def test_self_depend_suit():
suite = unittest.TestSuite()
suite.addTest(TestSelfDepend("test_empty_parameter"))
suite.addTest(TestSelfDepend("test_wrong_parameter"))
suite.addTest(TestSelfDepend("test_true_params_result"))
unittest.TextTestRunner().run(suite)
self.assertTrue(compare_two_values(output_for_input, resp_dict),
msg="The answer is not correct")
if __name__ == '__main__':
......
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
test import_databases
......@@ -5,8 +6,11 @@ test import_databases
import os
import shutil
import unittest
import warnings
from configparser import ConfigParser
import yaml
from packageship import system_config
from packageship.libs.exception import Error
try:
......@@ -29,13 +33,12 @@ try:
from test.base_code.init_config_path import init_config
except Exception:
raise
import warnings
import yaml
except Error:
raise Error
from packageship.application.initsystem.data_import import InitDataBase
from packageship.libs.exception import ContentNoneException, DatabaseRepeatException
from packageship.libs.exception import ContentNoneException
from packageship.libs.exception import DatabaseRepeatException
from packageship.libs.configutils.readconfig import ReadConfig
......@@ -49,13 +52,12 @@ class ImportData(unittest.TestCase):
warnings.filterwarnings("ignore")
def test_empty_param(self):
# If init is not obtained_ conf_ Path parameter
"""If init is not obtained_ conf_ Path parameter"""
try:
InitDataBase(config_file_path=None).init_data()
except Exception as e:
except ContentNoneException as error:
self.assertEqual(
e.__class__,
error.__class__,
ContentNoneException,
msg="No init in package_ conf_ Path parameter, wrong exception type thrown")
......@@ -69,9 +71,9 @@ class ImportData(unittest.TestCase):
w_f.write("")
InitDataBase(config_file_path=_config_path).init_data()
except Exception as e:
except ContentNoneException as error:
self.assertEqual(
e.__class__,
error.__class__,
ContentNoneException,
msg="Yaml file exists, but the content is empty. The exception type is wrong")
finally:
......@@ -83,18 +85,18 @@ class ImportData(unittest.TestCase):
try:
_config_path = ReadConfig().get_system('init_conf_path')
shutil.copyfile(_config_path, _config_path + '.bak')
with open(_config_path, 'r', encoding='utf-8') as f:
origin_yaml = yaml.load(f.read(), Loader=yaml.FullLoader)
with open(_config_path, 'r', encoding='utf-8') as file:
origin_yaml = yaml.load(file.read(), Loader=yaml.FullLoader)
for obj in origin_yaml:
obj["dbname"] = "openEuler"
with open(_config_path, 'w', encoding='utf-8') as w_f:
yaml.dump(origin_yaml, w_f)
InitDataBase(config_file_path=_config_path).init_data()
except Exception as e:
except DatabaseRepeatException as error:
self.assertEqual(
e.__class__,
error.__class__,
DatabaseRepeatException,
msg="Yaml file exists but DB_ Name duplicate exception type is wrong")
finally:
......@@ -103,8 +105,8 @@ class ImportData(unittest.TestCase):
os.rename(_config_path + '.bak', _config_path)
def test_wrong_param(self):
# If the corresponding current init_ conf_ The directory specified by
# path is incorrect
"""If the corresponding current init_ conf_ The directory
specified by path is incorrect"""
try:
# Back up source files
shutil.copyfile(
......@@ -118,9 +120,9 @@ class ImportData(unittest.TestCase):
_config_path = ReadConfig().get_system('init_conf_path')
InitDataBase(config_file_path=_config_path).init_data()
except Exception as e:
except FileNotFoundError as error:
self.assertEqual(
e.__class__,
error.__class__,
FileNotFoundError,
msg="init_ conf_ Path specified directory is empty exception type is wrong")
finally:
......@@ -144,10 +146,10 @@ class ImportData(unittest.TestCase):
_config_path = ReadConfig().get_system('init_conf_path')
InitDataBase(config_file_path=None).init_data()
except Exception as e:
except Error as error:
self.assertEqual(
e.__class__,
Exception,
error.__class__,
Error,
msg="Wrong exception type thrown when dbtype is wrong")
finally:
# To restore a file, delete the file first and then rename it back
......@@ -157,61 +159,60 @@ class ImportData(unittest.TestCase):
system_config.SYS_CONFIG_PATH)
def test_dbname(self):
"""test dbname"""
try:
_config_path = ReadConfig().get_system('init_conf_path')
shutil.copyfile(_config_path, _config_path + '.bak')
with open(_config_path, 'r', encoding='utf-8') as f:
origin_yaml = yaml.load(f.read(), Loader=yaml.FullLoader)
with open(_config_path, 'r', encoding='utf-8') as file:
origin_yaml = yaml.load(file.read(), Loader=yaml.FullLoader)
for obj in origin_yaml:
obj["dbname"] = ""
with open(_config_path, 'w', encoding='utf-8') as w_f:
yaml.dump(origin_yaml, w_f)
InitDataBase(config_file_path=_config_path).init_data()
with open(system_config.DATABASE_FILE_INFO, 'r', encoding='utf-8') as file_context:
init_database_date = yaml.load(
file_context.read(), Loader=yaml.FullLoader)
except DatabaseRepeatException as error:
self.assertEqual(
init_database_date,
None,
msg=" Priority must be a positive integer between 0 and 100 ")
except Exception as e:
return
error.__class__,
DatabaseRepeatException,
msg="Yaml file exists but DB_ Name duplicate exception type is wrong")
finally:
# Restore files
os.remove(_config_path)
os.rename(_config_path + '.bak', _config_path)
def test_src_db_file(self):
"""test src db file"""
try:
_config_path = ReadConfig().get_system('init_conf_path')
shutil.copyfile(_config_path, _config_path + '.bak')
with open(_config_path, 'r', encoding='utf-8') as f:
origin_yaml = yaml.load(f.read(), Loader=yaml.FullLoader)
with open(_config_path, 'r', encoding='utf-8') as file:
origin_yaml = yaml.load(file.read(), Loader=yaml.FullLoader)
for obj in origin_yaml:
obj["src_db_file"] = ""
with open(_config_path, 'w', encoding='utf-8') as w_f:
yaml.dump(origin_yaml, w_f)
InitDataBase(config_file_path=_config_path).init_data()
with open(system_config.DATABASE_FILE_INFO, 'r', encoding='utf-8') as file_context:
init_database_date = yaml.load(
file_context.read(), Loader=yaml.FullLoader)
except TypeError as error:
self.assertEqual(
init_database_date,
None,
msg=" Priority must be a positive integer between 0 and 100 ")
except Exception as e:
return
error.__class__,
TypeError,
msg="Yaml file exists but DB_ Name duplicate exception type is wrong")
finally:
# Restore files
os.remove(_config_path)
os.rename(_config_path + '.bak', _config_path)
def test_priority(self):
"""test priority"""
try:
_config_path = ReadConfig().get_system('init_conf_path')
shutil.copyfile(_config_path, _config_path + '.bak')
with open(_config_path, 'r', encoding='utf-8') as f:
origin_yaml = yaml.load(f.read(), Loader=yaml.FullLoader)
with open(_config_path, 'r', encoding='utf-8') as file:
origin_yaml = yaml.load(file.read(), Loader=yaml.FullLoader)
for obj in origin_yaml:
obj["priority"] = "-1"
with open(_config_path, 'w', encoding='utf-8') as w_f:
......@@ -224,7 +225,7 @@ class ImportData(unittest.TestCase):
init_database_date,
None,
msg=" Priority must be a positive integer between 0 and 100 ")
except Exception as e:
except FileNotFoundError:
return
finally:
# Restore files
......@@ -232,9 +233,9 @@ class ImportData(unittest.TestCase):
os.rename(_config_path + '.bak', _config_path)
def test_true_init_data(self):
'''
"""
Initialization of system data
'''
"""
# Normal configuration
_config_path = ReadConfig().get_system('init_conf_path')
InitDataBase(config_file_path=_config_path).init_data()
......@@ -250,11 +251,11 @@ class ImportData(unittest.TestCase):
# And there is data in this file, and it comes from the yaml file of
# conf
with open(_config_path, 'r', encoding='utf-8') as f:
yaml_config = yaml.load(f.read(), Loader=yaml.FullLoader)
with open(_config_path, 'r', encoding='utf-8') as file:
yaml_config = yaml.load(file.read(), Loader=yaml.FullLoader)
with open(path, 'r', encoding='utf-8') as f:
yaml_success = yaml.load(f.read(), Loader=yaml.FullLoader)
with open(path, 'r', encoding='utf-8') as files:
yaml_success = yaml.load(files.read(), Loader=yaml.FullLoader)
self.assertEqual(
len(yaml_config),
......@@ -264,7 +265,8 @@ class ImportData(unittest.TestCase):
# Compare name and priority
success_name_priority = dict()
config_name_priority = dict()
for i in range(len(yaml_config)):
len_con = len(yaml_config)
for i in range(len_con):
success_name_priority[yaml_success[i]["database_name"]] = \
yaml_success[i]["priority"]
config_name_priority[yaml_config[i]["dbname"]] = \
......@@ -273,17 +275,5 @@ class ImportData(unittest.TestCase):
self.assertEqual(
success_name_priority,
config_name_priority,
msg="The database and priority after initialization are inconsistent with the original file")
def test_import_data_suit():
"""test_import_data_suit"""
print("-----ImportData START----")
suite = unittest.TestSuite()
suite.addTest(ImportData("test_empty_param"))
suite.addTest(ImportData("test_wrong_param"))
suite.addTest(ImportData("test_dbname"))
suite.addTest(ImportData("test_src_db_file"))
suite.addTest(ImportData("test_priority"))
suite.addTest(ImportData("test_true_init_data"))
unittest.TextTestRunner().run(suite)
msg="The database and priority after initialization are"
"inconsistent with the original file")
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
packges test
......@@ -14,7 +15,7 @@ class TestPackages(ReadTestBase):
All package test cases
"""
def test_empty_dbName(self):
def test_empty_dbname(self):
"""dbName is none"""
resp = self.client.get("/packages")
......@@ -53,7 +54,7 @@ class TestPackages(ReadTestBase):
resp_dict.get("data"),
msg="Error in data information return")
def test_wrong_dbName(self):
def test_wrong_dbname(self):
"""dbName is err"""
resp = self.client.get("/packages?dbName=test")
......@@ -77,13 +78,5 @@ class TestPackages(ReadTestBase):
msg="Error in data information return")
def test_packages_suit():
print("---TestPackages START---")
suite = unittest.TestSuite()
suite.addTest(TestPackages("test_empty_dbName"))
suite.addTest(TestPackages("test_wrong_dbName"))
unittest.TextTestRunner().run(suite)
if __name__ == '__main__':
unittest.main()
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
test delete repodatas
......@@ -9,7 +10,7 @@ from test.base_code.operate_data_base import OperateTestBase
import unittest
import json
from packageship import system_config
# from test.base_code.operate_data_base import system_config
from packageship.libs.exception import Error
from packageship.application.apps.package.function.constants import ResponseCode
......@@ -89,20 +90,10 @@ class TestDeleteRepodatas(OperateTestBase):
self.assertIsNone(
resp_dict.get("data"),
msg="Error in data information return")
except Exception as e:
except Error:
return None
finally:
os.remove(system_config.DATABASE_FILE_INFO)
os.rename(system_config.DATABASE_FILE_INFO + '.bak', system_config.DATABASE_FILE_INFO)
shutil.rmtree(system_config.DATABASE_FOLDER_PATH)
os.rename(system_config.DATABASE_FOLDER_PATH + '.bak', system_config.DATABASE_FOLDER_PATH)
def test_delete_repodatas_suit():
"""unit testing"""
print("---TestDeleteRepodatas START---")
suite = unittest.TestSuite()
suite.addTest(TestDeleteRepodatas("test_wrong_dbname"))
suite.addTest(TestDeleteRepodatas("test_true_dbname"))
unittest.TextTestRunner().run(suite)
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
test get repodatas
......@@ -61,13 +62,5 @@ class TestGetRepodatas(ReadTestBase):
msg="Error in data information return")
def test_get_repodatas_suit():
"""unit testing"""
print("---TestGetRepodatas START---")
suite = unittest.TestSuite()
suite.addTest(TestGetRepodatas("test_dbname"))
unittest.TextTestRunner().run(suite)
if __name__ == '__main__':
unittest.main()
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
test_get_single_packages
......@@ -8,7 +9,6 @@ from test.base_code.read_data_base import ReadTestBase
import unittest
import json
from packageship.application.apps.package.function.constants import ResponseCode
......@@ -16,6 +16,7 @@ class TestGetSinglePack(ReadTestBase):
"""
Single package test case
"""
def test_error_sourcename(self):
"""sourceName is none or err"""
......@@ -148,14 +149,6 @@ class TestGetSinglePack(ReadTestBase):
resp_dict.get("data"),
msg="Error in data information return")
def test_get_single_package_suit():
"""unit testing"""
print("---TestSinglePack START---")
suite = unittest.TestSuite()
suite.addTest(TestGetSinglePack("test_error_sourcename"))
suite.addTest(TestGetSinglePack("test_true_dbname"))
suite.addTest(TestGetSinglePack("test_wrong_dbname"))
unittest.TextTestRunner().run(suite)
if __name__ == '__main__':
unittest.main()
#!/usr/bin/python3
"""TestUpdatePackage"""
# -*- coding:utf-8 -*-
from test.base_code.operate_data_base import OperateTestBase
import unittest
import json
from packageship.application.apps.package.function.constants import ResponseCode
class TestUpdatePackage(OperateTestBase):
"""TestUpdatePackage"""
def test_empty_dbname(self):
"""Parameter error"""
......@@ -148,15 +149,6 @@ class TestUpdatePackage(OperateTestBase):
resp_dict.get("data"),
msg="Error in data information return")
def test_updata_single_package_suit():
"""unit testing"""
print("---TestUpdatePackage START---")
suite = unittest.TestSuite()
suite.addTest(TestUpdatePackage("test_empty_dbname"))
suite.addTest(TestUpdatePackage("test_empty_sourcename"))
suite.addTest(TestUpdatePackage("test_true_parram"))
unittest.TextTestRunner().run(suite)
if __name__ == '__main__':
unittest.main()
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册