test_writer.py
1.6 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
from unittest.mock import MagicMock, call
from etl_to_crawler.writer import (
upsert_qq_singers,
upsert_qq_singer_songs,
upsert_yinyan_song_records,
)
def test_upsert_qq_singers_executes_insert():
cur = MagicMock()
singers = [{
'id': 1, 'mid': 'abc', 'name': '歌手', 'avatar': 'https://archive-dev.oss-cn-beijing.aliyuncs.com/a.jpg',
'sex': 'M', 'area': '华语', 'index': 'G', 'intro': None, 'home_url': None
}]
upsert_qq_singers(cur, singers)
assert cur.executemany.called
args = cur.executemany.call_args
assert 'crawler_qqmusic_singers' in args[0][0]
assert 'ON CONFLICT' in args[0][0]
def test_upsert_qq_singers_empty_does_nothing():
cur = MagicMock()
upsert_qq_singers(cur, [])
cur.executemany.assert_not_called()
def test_upsert_qq_singer_songs():
cur = MagicMock()
upsert_qq_singer_songs(cur, [(1, 'uuid-abc'), (2, 'uuid-def')])
assert cur.executemany.called
sql = cur.executemany.call_args[0][0]
assert 'crawler_qqmusic_singer_songs' in sql
def test_upsert_yinyan_song_records_replaces_existing_song_relation():
cur = MagicMock()
upsert_yinyan_song_records(cur, [(10, 100), (11, 101)])
assert cur.executemany.call_count == 2
delete_sql, delete_rows = cur.executemany.call_args_list[0][0]
insert_sql, insert_rows = cur.executemany.call_args_list[1][0]
assert 'DELETE FROM yinyan_song_records' in delete_sql
assert 'WHERE song_id = %s' in delete_sql
assert delete_rows == [(10,), (11,)]
assert 'INSERT INTO yinyan_song_records' in insert_sql
assert insert_rows == [(10, 100), (11, 101)]