test_check_record_titles.py
2.83 KB
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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
from pathlib import Path
from openpyxl import load_workbook
from check_record_titles import combine_rows, fetch_crawler_rows, fetch_spider_titles, write_excel
class Cursor:
def __init__(self, rows):
self.rows = rows
self.calls = []
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
return False
def execute(self, sql, params=None):
self.calls.append((sql, params))
def fetchall(self):
return self.rows
class Connection:
def __init__(self, cursor_rows):
self.cursors = [Cursor(rows) for rows in cursor_rows]
def cursor(self):
return self.cursors.pop(0)
def test_fetch_crawler_rows_joins_all_platform_tables():
connection = Connection([[(10, "1", 101, "recording-1", "同名")]])
cursor = connection.cursors[0]
rows = fetch_crawler_rows(connection, "yinyan_song_records")
assert rows == [{
"record_id": 10,
"platform": "1",
"platform_song_id": 101,
"recording_id": "recording-1",
"crawler_title": "同名",
}]
assert "WHERE ysr.is_archive_push = TRUE" in cursor.calls[0][0]
def test_fetch_crawler_rows_rejects_unknown_table():
connection = Connection([])
try:
fetch_crawler_rows(connection, "unsafe_table")
except ValueError as error:
assert "不支持" in str(error)
else:
raise AssertionError("应拒绝非白名单表")
def test_fetch_spider_titles_batches_queries_by_platform():
connection = Connection([
[{"id": 101, "title": "甲"}],
[{"id": 102, "title": "乙"}],
])
crawler_rows = [
{"platform": "1", "platform_song_id": 101},
{"platform": "2", "platform_song_id": 102},
]
titles = fetch_spider_titles(connection, crawler_rows, batch_size=2)
assert titles == {("1", 101): "甲", ("2", 102): "乙"}
def test_excel_marks_only_exact_non_null_matches_green(tmp_path: Path):
crawler_rows = [
{"record_id": 1, "platform": "1", "platform_song_id": 101, "recording_id": "r1", "crawler_title": "相同"},
{"record_id": 2, "platform": "2", "platform_song_id": 102, "recording_id": "r2", "crawler_title": " 名称 "},
{"record_id": 3, "platform": "4", "platform_song_id": 103, "recording_id": None, "crawler_title": None},
]
rows = combine_rows(crawler_rows, {("1", 101): "相同", ("2", 102): "名称"})
output = tmp_path / "check.xlsx"
write_excel(rows, output)
worksheet = load_workbook(output).active
assert [worksheet.cell(1, column).value for column in range(1, 7)] == [
"record_id", "spider_title", "crawler_title", "platform", "platform_song_id", "recording_id",
]
assert worksheet["A2"].fill.fgColor.rgb == "00C6EFCE"
assert worksheet["A3"].fill.fill_type is None
assert worksheet["A4"].fill.fill_type is None