这是单元测试套件:
util.py
:
class Util:
@staticmethod
def create_n_insert_into_sql(table_name, data):
# logic to create table in mysql
pass
@staticmethod
def create_n_insert_into_hive(table_name, data):
# logic to create hive table
pass
@staticmethod
def create_folder_hdfs(folder_name):
# logic to create hdfs folder
pass
@staticmethod
def get_data_from_external_source(source_name):
# logic to fetch data
pass
base.py
:
from util import Util as u
class BaseImporter:
__source = None
__table_name = None
__folder_name = None
def __init__(self, folder_name: str, source: str, table: str) -> None:
self.__source = source
self.__table_name = table
self.__folder_name = folder_name
def run_importer(self):
data = u.get_data_from_external_source(self.__source)
u.create_n_insert_into_sql(self.__table_name, data)
u.create_n_insert_into_hive(self.__table_name, data)
u.create_folder_hdfs(self.__folder_name)
child.py
:
from base import BaseImporter
class ChildImporter(BaseImporter):
def __init__(self, folder_name: str, source: str, table: str):
super().__init__(
folder_name='my_folder',
source='mysql',
table='accounts',
)
test_base.py
:
import unittest
from base import BaseImporter
from unittest.mock import patch
class TestBaseImporter(unittest.TestCase):
@patch('base.u', autospec=True)
def test_run_importer(self, mock_u):
base_importer = BaseImporter(folder_name='folder_name', source='source', table='table')
mock_u.get_data_from_external_source.return_value = 'mocked data'
base_importer.run_importer()
mock_u.get_data_from_external_source.assert_called_once_with('source')
mock_u.create_n_insert_into_sql.assert_called_once_with('table', 'mocked data')
mock_u.create_n_insert_into_hive.assert_called_once_with('table', 'mocked data')
mock_u.create_folder_hdfs.assert_called_once_with('folder_name')
test_child.py
:
import unittest
from child import ChildImporter
from base import BaseImporter
from unittest.mock import patch
class TestChildImporter(unittest.TestCase):
@patch('base.u', autospec=True)
def test_run_importer(self, mock_u):
child_importer = ChildImporter(folder_name='folder_name', source='source', table='table')
mock_u.get_data_from_external_source.return_value = 'mocked data'
child_importer.run_importer()
mock_u.get_data_from_external_source.assert_called_once_with('mysql')
mock_u.create_n_insert_into_sql.assert_called_once_with('accounts', 'mocked data')
mock_u.create_n_insert_into_hive.assert_called_once_with('accounts', 'mocked data')
mock_u.create_folder_hdfs.assert_called_once_with('my_folder')
self.assertIsInstance(child_importer, BaseImporter)
tests.py
:
import unittest
from test_base import TestBaseImporter
from test_child import TestChildImporter
if __name__ == '__main__':
test_loader = unittest.TestLoader()
test_classes_to_run = [TestBaseImporter, TestChildImporter]
suites_list = []
for test_class in test_classes_to_run:
suite = test_loader.loadTestsFromTestCase(test_class)
suites_list.append(suite)
big_suite = unittest.TestSuite(suites_list)
unittest.TextTestRunner().run(big_suite)
带有覆盖率报告的单元测试结果:
(venv) ☁ python-codelab [master] ⚡ coverage run /Users/ldu020/workspace/github.com/mrdulin/python-codelab/src/stackoverflow/56339991/tests.py && coverage report -m
..
----------------------------------------------------------------------
Ran 2 tests in 0.027s
OK
Name Stmts Miss Cover Missing
------------------------------------------------------------------------
src/stackoverflow/56339991/base.py 14 0 100%
src/stackoverflow/56339991/child.py 4 0 100%
src/stackoverflow/56339991/test_base.py 12 0 100%
src/stackoverflow/56339991/test_child.py 14 0 100%
src/stackoverflow/56339991/tests.py 12 0 100%
src/stackoverflow/56339991/util.py 9 4 56% 5, 10, 15, 20
------------------------------------------------------------------------
TOTAL 65 4 94%
源代码:https ://github.com/mrdulin/python-codelab/tree/master/src/stackoverflow/56339991