-
Notifications
You must be signed in to change notification settings - Fork 34
/
test_download.py
143 lines (119 loc) · 5.11 KB
/
test_download.py
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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
# SPDX-License-Identifier: LGPL-2.1-only
# SPDX-FileCopyrightText: 2021 Bastian Krause <[email protected]>, Pengutronix
import re
from helper import run
def test_download_inexistent_location(hawkbit, bundle_assigned, adjust_config):
"""
Assign bundle to target and test download to an inexistent location specified in config.
"""
location = '/tmp/does_not_exist/foo'
config = adjust_config(
{'client': {'bundle_download_location': location}}
)
out, err, exitcode = run(f'rauc-hawkbit-updater -c "{config}" -r')
assert 'New software ready for download' in out
# same warning from feedback() and from hawkbit_pull_cb()
assert err == \
f'WARNING: Failed to calculate free space for {location}: No such file or directory\n'*2
assert exitcode == 1
status = hawkbit.get_action_status()
assert status[0]['type'] == 'error'
assert f'Failed to calculate free space for {location}: No such file or directory' in \
status[0]['messages']
def test_download_unallowed_location(hawkbit, bundle_assigned, adjust_config):
"""
Assign bundle to target and test download to an unallowed location specified in config.
"""
location = '/root/foo'
config = adjust_config(
{'client': {'bundle_download_location': location}}
)
out, err, exitcode = run(f'rauc-hawkbit-updater -c "{config}" -r')
assert 'Start downloading' in out
assert err.strip() == \
f'WARNING: Download failed: Failed to open {location} for download: Permission denied'
assert exitcode == 1
status = hawkbit.get_action_status()
assert status[0]['type'] == 'error'
assert f'Download failed: Failed to open {location} for download: Permission denied' in \
status[0]['messages']
def test_download_too_slow(hawkbit, bundle_assigned, adjust_config, rate_limited_port):
"""Assign bundle to target and test too slow download of bundle."""
# limit to 50 bytes/s
port = rate_limited_port(50)
config = adjust_config({
'client': {
'hawkbit_server': f'{hawkbit.host}:{port}',
'low_speed_time': '3',
'low_speed_rate': '100',
}
})
out, err, exitcode = run(f'rauc-hawkbit-updater -c "{config}" -r', timeout=90)
assert 'Start downloading: ' in out
assert err.strip() == 'WARNING: Download failed: Timeout was reached'
assert exitcode == 1
def test_download_partials_without_resume(hawkbit, bundle_assigned, adjust_config,
partial_download_port):
"""
Assign bundle to target and test download of partial bundle parts without having
download resuming configured.
"""
config = adjust_config(
{'client': {'hawkbit_server': f'{hawkbit.host}:{partial_download_port}'}}
)
# ignore failing installation
out, err, exitcode = run(f'rauc-hawkbit-updater -c "{config}" -r')
assert 'Start downloading: ' in out
assert err.strip() == 'WARNING: Download failed: Transferred a partial file'
assert exitcode == 1
def test_download_partials_with_resume(hawkbit, bundle_assigned, adjust_config,
partial_download_port):
"""
Assign bundle to target and test download of partial bundle parts with download resuming
configured.
"""
config = adjust_config({
'client': {
'hawkbit_server': f'{hawkbit.host}:{partial_download_port}',
'resume_downloads': 'true',
}
})
# ignore failing installation
out, _, _ = run(f'rauc-hawkbit-updater -c "{config}" -r')
assert re.findall('Resuming download from offset [1-9]', out)
assert 'Download complete.' in out
assert 'File checksum OK.' in out
def test_download_slow_with_resume(hawkbit, bundle_assigned, adjust_config, rate_limited_port):
"""
Assign bundle to target and test slow download of bundle with download resuming enabled. That
should lead to resuming downloads.
"""
port = rate_limited_port(50000)
config = adjust_config({
'client': {
'hawkbit_server': f'{hawkbit.host}:{port}',
'resume_downloads': 'true',
'low_speed_time': '1',
'low_speed_rate': '100000',
}
})
# ignore failing installation
out, _, _ = run(f'rauc-hawkbit-updater -c "{config}" -r')
assert 'Timeout was reached, resuming download..' in out
assert 'Resuming download from offset' in out
assert 'Download complete.' in out
assert 'File checksum OK.' in out
def test_download_only(hawkbit, config, assign_bundle):
"""Test "downloadonly" deployment."""
assign_bundle(params={'type': 'downloadonly'})
out, err, exitcode = run(f'rauc-hawkbit-updater -c "{config}" -r')
assert 'Start downloading' in out
assert 'hawkBit requested to skip installation, not invoking RAUC yet.' in out
assert 'Download complete' in out
assert 'File checksum OK' in out
assert err == ''
assert exitcode == 0
status = hawkbit.get_action_status()
assert status[0]['type'] == 'downloaded'
# check last status message
assert 'File checksum OK.' in status[0]['messages']