m's blog

🧀🧀🧀

0%

Posted on In

通过main()去运行测试

1
2
3
4
...

if __name__ == '__main__':
    unittest.main()

命令行接口方式进行测试

1
2
3
python -m unittest test_module1 test_module2
python -m unittest test_module.TestClass
python -m unittest test_module.TestClass.test_method

测试中主要通过3个方法进行判断是否正常：

  1. assertEqual()去检查结果是否和期望的结果相同；
  2. assertTrue()用于核实最终状态；
  3. assertRaises()去检查是否抛出期望的异常。

还有其他的一些如下表：

Method Checks that
assertEqual(a, b) a == b
assertNotEqual(a, b) a != b
assertTrue(x) bool(x) is True
assertFalse(x) bool(x) is False
assertIs(a, b) a is b
assertIsNot(a, b) a is not b
assertIsNone(x) x is None
assertIsNotNone(x) x is not None
assertIn(a, b) a in b
assertNotIn(a, b) a not in b
assertIsInstance(a, b) isinstance(a, b)
assertNotIsInstance(a, b) not isinstance(a, b)
assertRaises(exc, fun, *args, **kwds) fun(*args, **kwds) raises exc
assertRaisesRegexp(exc, r, fun, *args, **kwds) fun(*args, **kwds) raises exc and the message matches regex r

在TestCase的类里，setUp()会先被加载，然后再去运行需要测试的方法

1
2
3
4
5
6
7
import unittest

class CustomTestCase(unittest.TestCase):
    def setUp(self):
        pass

    ...

如果实在生产环境下运行的测试，在一般情况下，是在setUp()创建或更新一些运行测试必要的数据，整个测试运行完后，tearDown()会被执行，删除或恢复数据的操作可以放在tearDown()里边，如：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class ListApiTests(unittest.TestCase):
    def setUp(self):
        # 先创建一个测试的用户
        user_info = {
            'username': 'test_user_actmerce',
            'password': make_password('yoyoqiekenao'),
            'real_name': '晓测'
        }
        user = UserProfile.objects.create(**user_info)
        self.user = user

    def tearDown(self):
        # 删除新增的数据
        self.user.delete()
        
    def test_one(self):
        pass

参考：

Posted on In

今天遇到问题主要是对比两个不同时间、同个数据库导出的csv文件。因为这次主要用pandas去处理csv文件，那么问题主要就是怎么实现通过对比两个dataframe获取新的数据以及更新了的数据。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import chardet

import pandas as pd


def get_file_encoding(file_path):
    f = open(file_path, 'rb')
    file_encoding = chardet.detect(f.read())['encoding']
    f.close()
    return file_encoding


def get_df(file_path, id_name, field_list):
    # 获取file 的编码格式
    file_encoding = get_file_encoding(file_path)

    # 初始化df，包括把id_name替换成索引，以及替换nan为空，不然会有异常，最后再来一个过滤显示的字段
    df = pd.read_csv(file_path, encoding=file_encoding).set_index([id_name])[field_list].fillna('').sort_index()
    return df


def contrast_df(new_path, old_path, id_name, field_list, save_file_path='./auto_create/'):
    # 把id的列移除list
    field_list.pop(field_list.index(id_name))

    # 获取file_name
    file_name = new_path.split('/')[-1]

    # 获取新旧的df
    new_df = get_df(new_path, id_name, field_list)
    old_df = get_df(old_path, id_name, field_list)

    # 比较数据是否有更新前，须要先看是否有新增的数据，不然size不同会报错
    old_index = old_df.index.values
    add_df = new_df[~new_df.index.isin(old_index)]
    if not add_df.empty:
        print('有{}条数据新插入了'.format(len(add_df)))
        # 把新增的数据从new_df 中删除
        add_index = add_df.index.values
        new_df = new_df[~new_df.index.isin(add_index)]
        
        # 新建csv文件
        add_df.to_csv(save_file_path + 'add_' + file_name)
    else:
        print('没有数据新插入')
        
    ne_stacked = (new_df != old_df).stack()
    changed = ne_stacked[ne_stacked]
    changed.index.names = ['ID_P', 'col']
    changed_index = set([seri[0] for seri in changed.index])
    update_df = new_df[new_df.index.isin(changed_index)]
    
    if not update_df.empty:
        print('有{}条数据更新'.format(len(update_df)))
        # 新建csv文件
        update_df.to_csv(save_file_path + 'update_' + file_name)
    else:
        print('没有数据更新')

    return add_df, update_df

基本思路是：

  1. 先把新增的数据移除，并保存成新的csv文件；
  2. 两个新旧的dataframe进行比较；
  3. 通过行列转化的形式获取changed,并从中获得发生改变的数据的index；
  4. 将更新的数据保存成新的csv文件。

最后补充一下：

  1. 上边的changed能做得更多，它还能识别数据哪个字段发生了改变；
  2. 中间还试了别的方法去完成这个功能，但是没有跑通，不过思路还是挺有趣的，代码丢上来mark一下
1
2
3
4
df_3 = pd.concat([new_df, old_df])
df_gpby = df_3.groupby(list(df_3.columns))
idx = [x[0] for x in df_gpby.groups.values() if len(x) == 1]
df_3.iloc[idx]