Skip to content
Toggle navigation
Toggle navigation
This project
Loading...
Sign in
沈秋雨
/
import_hk_songs
Go to a project
Toggle navigation
Toggle navigation pinning
Projects
Groups
Snippets
Help
Project
Activity
Repository
Pipelines
Graphs
Issues
0
Merge Requests
0
Wiki
Network
Create a new issue
Builds
Commits
Issue Boards
Files
Commits
Network
Compare
Branches
Tags
Commit
179a08f8
...
179a08f8e141280b533af96ab18328a2a0c83bbc
authored
2026-07-15 10:56:51 +0800
by
沈秋雨
Browse Files
Options
Browse Files
Tag
Download
Email Patches
Plain Diff
fix
1 parent
0f853eb5
Hide whitespace changes
Inline
Side-by-side
Showing
1 changed file
with
89 additions
and
0 deletions
test_check_record_titles.py
test_check_record_titles.py
0 → 100644
View file @
179a08f
from
pathlib
import
Path
from
openpyxl
import
load_workbook
from
check_record_titles
import
combine_rows
,
fetch_crawler_rows
,
fetch_record_names
,
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"
,
"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_record_names_batches_queries
():
connection
=
Connection
([
[{
"id"
:
1
,
"record_name"
:
"甲"
},
{
"id"
:
2
,
"record_name"
:
"乙"
}],
[{
"id"
:
3
,
"record_name"
:
"丙"
}],
])
names
=
fetch_record_names
(
connection
,
[
3
,
1
,
2
,
2
],
batch_size
=
2
)
assert
names
==
{
1
:
"甲"
,
2
:
"乙"
,
3
:
"丙"
}
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"
,
"title"
:
"相同"
},
{
"record_id"
:
2
,
"platform"
:
"2"
,
"platform_song_id"
:
102
,
"recording_id"
:
"r2"
,
"title"
:
" 名称 "
},
{
"record_id"
:
3
,
"platform"
:
"4"
,
"platform_song_id"
:
103
,
"recording_id"
:
None
,
"title"
:
None
},
]
rows
=
combine_rows
(
crawler_rows
,
{
1
:
"相同"
,
2
:
"名称"
,
3
:
None
})
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"
,
"record_name"
,
"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
Please
register
or
sign in
to post a comment