test_oss.py
5.14 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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
from unittest.mock import MagicMock, patch
import math
import requests
from etl_to_crawler.oss import download_text_url, transfer_url, transfer_url_with_md5
ARCHIVE_URL = "https://archive-dev.oss-cn-beijing.aliyuncs.com/some/path.mp3"
OTHER_URL = "https://hikoon-data-platform.oss-cn-beijing.aliyuncs.com/qq-audio/abc.mp3"
BASE_URL = "https://archive-dev.oss-cn-beijing.aliyuncs.com"
def test_already_archive_dev_returns_as_is():
bucket = MagicMock()
result = transfer_url(ARCHIVE_URL, "any/key.mp3", bucket, BASE_URL)
assert result == ARCHIVE_URL
bucket.put_object.assert_not_called()
def test_download_text_url_reads_source_even_when_it_is_target_oss():
with patch('etl_to_crawler.oss._http_get') as mock_get:
mock_get.return_value.content = b'\xef\xbb\xbf[00:01.00]lyric'
mock_get.return_value.raise_for_status = MagicMock()
result = download_text_url(ARCHIVE_URL, BASE_URL)
assert result == '[00:01.00]lyric'
def test_empty_url_returns_empty():
bucket = MagicMock()
result = transfer_url('', "any/key.mp3", bucket, BASE_URL)
assert result == ''
bucket.put_object.assert_not_called()
def test_none_url_returns_empty():
bucket = MagicMock()
result = transfer_url(None, "any/key.mp3", bucket, BASE_URL)
assert result == ''
def test_invalid_nan_url_returns_empty_without_download():
bucket = MagicMock()
with patch('etl_to_crawler.oss._http_get') as mock_get:
assert transfer_url('nan', "any/key.mp3", bucket, BASE_URL) == ''
assert transfer_url(math.nan, "any/key.mp3", bucket, BASE_URL) == ''
assert transfer_url('not-a-url', "any/key.mp3", bucket, BASE_URL) == ''
assert transfer_url_with_md5('nan', "any/key.mp3", bucket, BASE_URL) == ('', '')
mock_get.assert_not_called()
bucket.put_object.assert_not_called()
def test_external_url_downloads_and_uploads():
bucket = MagicMock()
fake_content = b"audio_bytes"
with patch('etl_to_crawler.oss._http_get') as mock_get:
mock_get.return_value.content = fake_content
mock_get.return_value.raise_for_status = MagicMock()
result = transfer_url(OTHER_URL, "crawler/audio/qq/abc.mp3", bucket, BASE_URL)
bucket.put_object.assert_called_once_with("crawler/audio/qq/abc.mp3", fake_content)
assert result == f"{BASE_URL}/crawler/audio/qq/abc.mp3"
def test_external_url_can_rewrite_download_base_to_internal_endpoint():
bucket = MagicMock()
public_source = "https://source-bucket.oss-cn-hangzhou.aliyuncs.com/audio/abc.mp3"
internal_source_base = "https://source-bucket.oss-cn-hangzhou-internal.aliyuncs.com"
fake_content = b"audio_bytes"
with patch.dict('etl_to_crawler.oss.OSS_CONFIG', {
'download_rewrite_from_base_url': 'https://source-bucket.oss-cn-hangzhou.aliyuncs.com',
'download_base_url': internal_source_base,
}):
with patch('etl_to_crawler.oss._http_get') as mock_get:
mock_get.return_value.content = fake_content
mock_get.return_value.raise_for_status = MagicMock()
result = transfer_url(public_source, "crawler/audio/qq/abc.mp3", bucket, BASE_URL)
mock_get.assert_called_once_with(f"{internal_source_base}/audio/abc.mp3", timeout=30)
bucket.put_object.assert_called_once_with("crawler/audio/qq/abc.mp3", fake_content)
assert result == f"{BASE_URL}/crawler/audio/qq/abc.mp3"
def test_transfer_url_with_md5_hashes_downloaded_content_before_upload():
bucket = MagicMock()
fake_content = b"audio_bytes"
with patch('etl_to_crawler.oss._http_get') as mock_get:
mock_get.return_value.content = fake_content
mock_get.return_value.raise_for_status = MagicMock()
result, audio_md5 = transfer_url_with_md5(OTHER_URL, "crawler/audio/qq/abc.mp3", bucket, BASE_URL)
bucket.put_object.assert_called_once_with("crawler/audio/qq/abc.mp3", fake_content)
assert result == f"{BASE_URL}/crawler/audio/qq/abc.mp3"
assert audio_md5 == "04d43544b267629d9089eaed3b847a99"
def test_audio_transfer_retries_timeout_then_uploads():
bucket = MagicMock()
response = MagicMock(content=b'audio_bytes')
response.raise_for_status.return_value = None
with patch('etl_to_crawler.oss._http_get', side_effect=[requests.ReadTimeout('slow'), response]) as mock_get:
with patch('etl_to_crawler.oss.time.sleep') as mock_sleep:
result, _ = transfer_url_with_md5(OTHER_URL, 'crawler/audio/qq/abc.mp3', bucket, BASE_URL)
assert mock_get.call_count == 2
mock_sleep.assert_called_once()
assert result == f"{BASE_URL}/crawler/audio/qq/abc.mp3"
def test_transfer_does_not_retry_not_found_response():
bucket = MagicMock()
response = MagicMock(status_code=404)
error = requests.HTTPError(response=response)
with patch('etl_to_crawler.oss._http_get', side_effect=error) as mock_get:
with patch('etl_to_crawler.oss.time.sleep') as mock_sleep:
try:
transfer_url_with_md5(OTHER_URL, 'crawler/audio/qq/abc.mp3', bucket, BASE_URL)
except requests.HTTPError:
pass
else:
raise AssertionError('expected HTTPError')
mock_get.assert_called_once()
mock_sleep.assert_not_called()