Skip to content

Commit cdef2cc

Browse files
author
panglei
committed
test: 更新测试脚本
1 parent 4b80f6d commit cdef2cc

6 files changed

Lines changed: 520 additions & 121 deletions

File tree

‎tests/conftest.py‎

Lines changed: 163 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,163 @@
1+
#!/usr/bin/env python
2+
# -*- coding: utf-8 -*-
3+
"""共享 fixtures：数据库连接、测试数据生成、常用对象构造"""
4+
import os
5+
import pathlib
6+
import random
7+
import datetime
8+
9+
import pytest
10+
from faker import Faker
11+
from peewee import (
12+
Model, CharField, IntegerField, DateField,
13+
DatabaseProxy, SqliteDatabase,
14+
)
15+
from playhouse.db_url import connect
16+
17+
from data_watchtower import DbServices, Watchtower, DatabaseLoader
18+
from data_watchtower.model.models import (
19+
database_proxy as backend_db_proxy,
20+
WatchtowerModel, ValidationDetailModel, ValidatorRelationModel,
21+
)
22+
23+
# ---------------------------------------------------------------------------
24+
# 常量
25+
# ---------------------------------------------------------------------------
26+
# connectorx 将 sqlite:///relative.db 解析为绝对路径 /relative.db，
27+
# 因此需要使用绝对路径以确保 connectorx 和 peewee 都能正确定位文件。
28+
_project_root = pathlib.Path(__file__).parent.parent
29+
30+
NUM_OF_STUDENTS = 20
31+
NUM_OF_DAYS = 10
32+
33+
# ---------------------------------------------------------------------------
34+
# 测试数据源 — Score / Student 模型（文件级 SQLite）
35+
# ---------------------------------------------------------------------------
36+
_test_data_db_proxy = DatabaseProxy()
37+
38+
39+
class _BaseModel(Model):
40+
class Meta:
41+
database = _test_data_db_proxy
42+
43+
44+
class Student(_BaseModel):
45+
name = CharField()
46+
age = IntegerField()
47+
48+
49+
class Score(_BaseModel):
50+
name = CharField()
51+
date = DateField()
52+
chinese = IntegerField(null=True)
53+
english = IntegerField(null=True)
54+
math = IntegerField(null=True)
55+
56+
57+
def _generate_test_data(db_url):
58+
"""创建 Student / Score 表并填充随机测试数据"""
59+
database = connect(db_url)
60+
_test_data_db_proxy.initialize(database)
61+
62+
with database:
63+
database.drop_tables([Student, Score])
64+
database.create_tables([Student, Score])
65+
66+
fake = Faker(locale='zh_CN')
67+
students = []
68+
for _ in range(NUM_OF_STUDENTS):
69+
student = Student.create(
70+
name=fake.unique.name(),
71+
age=fake.random_int(min=10, max=18),
72+
)
73+
students.append(student)
74+
75+
for i in range(NUM_OF_DAYS):
76+
date = (datetime.datetime.now() - datetime.timedelta(i)).date()
77+
random.shuffle(students)
78+
for j, student in enumerate(students):
79+
if j == 0:
80+
# 让一个学生缺考
81+
continue
82+
chinese = fake.random_int(min=0, max=100)
83+
english = fake.random_int(min=0, max=100)
84+
math = fake.random_int(min=0, max=100)
85+
if fake.random_int(min=1, max=10) == 1:
86+
chinese = None
87+
if fake.random_int(min=1, max=10) == 1:
88+
english = None
89+
if fake.random_int(min=1, max=10) == 1:
90+
math = None
91+
Score.create(
92+
name=student.name,
93+
date=date,
94+
chinese=chinese,
95+
english=english,
96+
math=math,
97+
)
98+
99+
return database
100+
101+
102+
# ---------------------------------------------------------------------------
103+
# Fixtures
104+
# ---------------------------------------------------------------------------
105+
106+
@pytest.fixture(scope='session')
107+
def test_data_db_url():
108+
"""被监控数据源的 SQLite URL（绝对路径）"""
109+
return os.getenv(
110+
'DW_TEST_DATA_DB_URL',
111+
f'sqlite:///{_project_root / "test.db"}',
112+
)
113+
114+
115+
@pytest.fixture(scope='session')
116+
def backend_db_url():
117+
"""后端存储的 SQLite URL（绝对路径）"""
118+
return os.getenv(
119+
'DW_BACKEND_DB_URL',
120+
f'sqlite:///{_project_root / "data.db"}',
121+
)
122+
123+
124+
@pytest.fixture(scope='session')
125+
def test_data_db(test_data_db_url):
126+
"""Session 级别：生成 Student / Score 测试数据，返回 peewee Database"""
127+
database = _generate_test_data(test_data_db_url)
128+
yield database
129+
database.close()
130+
131+
132+
@pytest.fixture
133+
def db_svr():
134+
"""函数级别：使用内存数据库的 DbServices，每个测试隔离"""
135+
database = SqliteDatabase(':memory:')
136+
backend_db_proxy.initialize(database)
137+
database.create_tables([WatchtowerModel, ValidationDetailModel, ValidatorRelationModel])
138+
svr = DbServices(database)
139+
yield svr
140+
database.close()
141+
142+
143+
@pytest.fixture
144+
def file_db_svr(backend_db_url):
145+
"""函数级别：使用文件数据库的 DbServices（集成测试用）"""
146+
return DbServices(backend_db_url)
147+
148+
149+
@pytest.fixture
150+
def sample_watchtower():
151+
"""创建一个简单的 Watchtower 用于测试"""
152+
data_loader = DatabaseLoader(query="SELECT 1", connection="sqlite:///test.db")
153+
return Watchtower(name='test_wt', data_loader=data_loader)
154+
155+
156+
@pytest.fixture
157+
def custom_macro_map():
158+
"""通用的自定义宏配置"""
159+
return {
160+
'today': {'impl': lambda: datetime.datetime.today().strftime("%Y-%m-%d")},
161+
'start_date': '2024-04-01',
162+
'column': 'name',
163+
}

‎tests/test_data_loader.py‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
#!/usr/bin/env python
2+
# -*- coding: utf-8 -*-
3+
"""DatabaseLoader 测试
4+
5+
实际 DatabaseLoader._load() 使用 pl.read_database_uri()，
6+
不涉及手动连接管理。
7+
"""
8+
import pytest
9+
from unittest.mock import patch, MagicMock
10+
from data_watchtower import DatabaseLoader
11+
12+
13+
class TestDatabaseLoaderLoad:
14+
"""测试 DatabaseLoader._load() 方法"""
15+
16+
def test_load_returns_data_on_success(self):
17+
"""_load() 成功时返回 read_database_uri 的结果"""
18+
mock_data = MagicMock()
19+
20+
with patch('data_watchtower.core.data_loaders.pl.read_database_uri', return_value=mock_data) as mock_read:
21+
loader = DatabaseLoader(query="SELECT 1", connection="sqlite:///test.db")
22+
result = loader._load()
23+
24+
mock_read.assert_called_once_with("SELECT 1", uri="sqlite:///test.db")
25+
assert result is mock_data
26+
27+
def test_load_raises_on_query_error(self):
28+
"""查询失败时 _load() 应抛出异常"""
29+
with patch('data_watchtower.core.data_loaders.pl.read_database_uri', side_effect=RuntimeError("query failed")):
30+
loader = DatabaseLoader(query="SELECT bad", connection="sqlite:///test.db")
31+
with pytest.raises(RuntimeError, match="query failed"):
32+
loader._load()

‎tests/test_flow.py‎

Lines changed: 30 additions & 121 deletions
Original file line numberDiff line numberDiff line change
@@ -1,124 +1,40 @@
1-
import os
2-
import pytest
3-
import random
4-
import datetime
5-
from faker import Faker
6-
from peewee import *
7-
from playhouse.db_url import connect
8-
from data_watchtower import (DbServices, Watchtower, DatabaseLoader,
9-
ExpectRowCountToBeBetween, ExpectColumnValuesToNotBeNull)
10-
11-
dw_test_data_db_url = os.getenv('DW_TEST_DATA_DB_URL', 'sqlite:///test.db')
12-
dw_backend_db_url = os.getenv('DW_BACKEND_DB_URL', "sqlite:///data.db")
13-
NUM_OF_STUDENTS = 20
14-
NUM_OF_DAY = 10
15-
16-
database_proxy = DatabaseProxy()
17-
18-
19-
class BaseModel(Model):
20-
class Meta:
21-
database = database_proxy
22-
23-
24-
class Student(BaseModel):
25-
name = CharField()
26-
age = IntegerField()
27-
28-
29-
class Score(BaseModel):
30-
name = CharField()
31-
date = DateField()
32-
chinese = IntegerField(null=True)
33-
english = IntegerField(null=True)
34-
math = IntegerField(null=True)
35-
36-
37-
def setup_module():
38-
# print("初始化数据")
39-
database = connect(dw_test_data_db_url)
40-
database_proxy.initialize(database)
41-
with database:
42-
database.drop_tables([Student, Score])
43-
database.create_tables([Student, Score])
44-
45-
fake = Faker(locale='zh_CN')
46-
students = []
47-
for i in range(NUM_OF_STUDENTS):
48-
student = Student.create(name=fake.unique.name(), age=fake.random_int(min=10, max=18))
49-
students.append(student)
50-
days = []
51-
for i in range(NUM_OF_DAY):
52-
date = (datetime.datetime.now() - datetime.timedelta(i)).date()
53-
days.append(date)
54-
random.shuffle(students)
55-
for j, student in enumerate(students):
56-
if j == 0:
57-
# 让一个学生缺考
58-
continue
59-
chinese = fake.random_int(min=0, max=100)
60-
english = fake.random_int(min=0, max=100)
61-
math = fake.random_int(min=0, max=100)
62-
if fake.random_int(min=1, max=10) == 1:
63-
chinese = None
64-
if fake.random_int(min=1, max=10) == 1:
65-
english = None
66-
if fake.random_int(min=1, max=10) == 1:
67-
math = None
68-
score = Score.create(
69-
name=student.name,
70-
date=date,
71-
chinese=chinese,
72-
english=english,
73-
math=math,
74-
)
75-
76-
77-
def teardown_module():
78-
# print('测试模块清理')
79-
pass
80-
81-
82-
@pytest.fixture
83-
def custom_macro_map():
84-
return {
85-
'today': {'impl': lambda: datetime.datetime.today().strftime("%Y-%m-%d")},
86-
'start_date': '2024-04-01',
87-
'column': 'name',
88-
}
89-
90-
91-
@pytest.fixture
92-
def db_svr():
93-
return DbServices(dw_backend_db_url)
94-
95-
96-
def test_demo_data():
1+
#!/usr/bin/env python
2+
# -*- coding: utf-8 -*-
3+
"""集成测试：完整流程 DataLoader → Watchtower → Validator → DbServices"""
4+
from data_watchtower import (
5+
Watchtower, DatabaseLoader,
6+
ExpectRowCountToBeBetween, ExpectColumnValuesToNotBeNull,
7+
)
8+
from conftest import Student, Score, NUM_OF_STUDENTS, NUM_OF_DAYS
9+
10+
11+
def test_demo_data(test_data_db):
12+
"""验证测试数据已正确生成"""
9713
assert Student.select().count() == NUM_OF_STUDENTS
98-
assert Score.select().count() == NUM_OF_DAY * (NUM_OF_STUDENTS - 1)
14+
assert Score.select().count() == NUM_OF_DAYS * (NUM_OF_STUDENTS - 1)
9915

10016

101-
def test_create_table(db_svr):
102-
db_svr.create_tables()
103-
assert len(db_svr.get_watchtowers()) >= 0
17+
def test_create_table(file_db_svr):
18+
file_db_svr.create_tables()
19+
assert len(file_db_svr.get_watchtowers()) >= 0
10420

10521

106-
def test_watchtower_crud(db_svr, custom_macro_map):
22+
def test_watchtower_crud(file_db_svr, test_data_db_url, custom_macro_map):
10723
wt_name = 'score of ${today}'
10824
query = "SELECT * FROM score where date='${today}'"
10925
params = dict(
11026
schedule="12:00",
11127
validator_success_method='all',
11228
success_method='all',
11329
)
114-
connection = dw_test_data_db_url
30+
connection = test_data_db_url
11531
# 先删除存在的
116-
db_svr.delete_watchtower(wt_name)
32+
file_db_svr.delete_watchtower(wt_name)
11733

11834
data_loader = DatabaseLoader(query=query, connection=connection)
11935
watchtower = Watchtower(name=wt_name, data_loader=data_loader, custom_macro_map=custom_macro_map, **params)
120-
db_svr.add_watchtower(watchtower)
121-
wt = db_svr.get_watchtower(wt_name)
36+
file_db_svr.add_watchtower(watchtower)
37+
wt = file_db_svr.get_watchtower(wt_name)
12238
assert wt['name'] == wt_name
12339
assert wt['data_loader']['__class__'] == DatabaseLoader.module_path()
12440
assert wt['data_loader']['query'] == query
@@ -133,30 +49,24 @@ def test_watchtower_crud(db_svr, custom_macro_map):
13349
query = "SELECT * FROM score where date='${today}' and 1=1 "
13450
data_loader = DatabaseLoader(query=query, connection=connection)
13551

136-
db_svr.update_watchtower(name=wt_name, data_loader=data_loader, params=params)
137-
wt = db_svr.get_watchtower(wt_name)
52+
file_db_svr.update_watchtower(name=wt_name, data_loader=data_loader, params=params)
53+
wt = file_db_svr.get_watchtower(wt_name)
13854
assert wt['name'] == wt_name
13955
assert wt['data_loader']['__class__'] == DatabaseLoader.module_path()
14056
assert wt['data_loader']['query'] == query
14157
assert wt['data_loader']['connection'] == connection
14258
assert wt['params'] == params
14359

144-
assert db_svr.delete_watchtower(wt_name) == 1
145-
wt = db_svr.get_watchtower(wt_name)
60+
assert file_db_svr.delete_watchtower(wt_name) == 1
61+
wt = file_db_svr.get_watchtower(wt_name)
14662
assert wt is None
14763

14864

149-
def test_case1(db_svr):
65+
def test_case1(file_db_svr, test_data_db, test_data_db_url, custom_macro_map):
15066
wt_name = 'score of ${today}'
151-
# 自定义宏模板
152-
custom_macro_map = {
153-
'today': {'impl': lambda: datetime.datetime.today().strftime("%Y-%m-%d")},
154-
'start_date': '2024-04-01',
155-
'column': 'name',
156-
}
15767
# 设置数据加载器,用来加载需要校验的数据
15868
query = "SELECT * FROM score where date='${today}'"
159-
data_loader = DatabaseLoader(query=query, connection=dw_test_data_db_url)
69+
data_loader = DatabaseLoader(query=query, connection=test_data_db_url)
16070
# 创建监控项
16171
watchtower = Watchtower(name=wt_name, data_loader=data_loader, custom_macro_map=custom_macro_map)
16272
# 添加校验器
@@ -173,11 +83,11 @@ def test_case1(db_svr):
17383
assert wt_name == watchtower.name
17484

17585
# 保存监控配置
176-
db_svr.add_watchtower(watchtower)
86+
file_db_svr.add_watchtower(watchtower)
17787
# 保存监控结果
178-
db_svr.save_result(watchtower, result)
88+
file_db_svr.save_result(watchtower, result)
17989

180-
item = db_svr.get_watchtower(wt_name)
90+
item = file_db_svr.get_watchtower(wt_name)
18191
watchtower = Watchtower.from_dict(item)
18292

18393
watchtower.set_custom_macro(**custom_macro_map)
@@ -193,4 +103,3 @@ def test_case1(db_svr):
193103
watchtower.macro_template.macro_config = result['macro_maps']
194104
assert result['name'] == watchtower.macro_template.apply_string(watchtower.name)
195105
assert wt_name == watchtower.name
196-
return

0 commit comments

Comments
 (0)