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
|
# -*- coding: utf-8 -*-
import json
import responses
from tests.conftest import IaRequestsMock, load_test_data_file, ia_call
PROTOCOL = 'https:'
STATUS_CHECK_RESPONSE = load_test_data_file('s3_status_check.json')
def test_ia_upload(tmpdir_ch, caplog):
with open('test.txt', 'w') as fh:
fh.write('foo')
with IaRequestsMock() as rsps:
rsps.add_metadata_mock('nasa')
rsps.add(responses.PUT, '{0}//s3.us.archive.org/nasa/test.txt'.format(PROTOCOL),
body='',
content_type='text/plain')
ia_call(['ia', '--log', 'upload', 'nasa', 'test.txt'])
assert ('uploaded test.txt to {0}//s3.us.archive.org/nasa/'
'test.txt'.format(PROTOCOL)) in caplog.text
def test_ia_upload_status_check(capsys):
with IaRequestsMock() as rsps:
rsps.add(responses.GET, '{0}//s3.us.archive.org'.format(PROTOCOL),
body=STATUS_CHECK_RESPONSE,
content_type='application/json')
ia_call(['ia', 'upload', 'nasa', '--status-check'])
out, err = capsys.readouterr()
assert 'success: nasa is accepting requests.' in out
j = json.loads(STATUS_CHECK_RESPONSE)
j['over_limit'] = 1
rsps.add(responses.GET, '{0}//s3.us.archive.org'.format(PROTOCOL),
body=json.dumps(j),
content_type='application/json')
ia_call(['ia', 'upload', 'nasa', '--status-check'], expected_exit_code=1)
out, err = capsys.readouterr()
assert ('warning: nasa is over limit, and not accepting requests. '
'Expect 503 SlowDown errors.') in err
def test_ia_upload_debug(capsys, tmpdir_ch, nasa_mocker):
with open('test.txt', 'w') as fh:
fh.write('foo')
ia_call(['ia', 'upload', '--debug', 'nasa', 'test.txt'])
out, err = capsys.readouterr()
assert 'User-Agent' in out
assert 's3.us.archive.org/nasa/test.txt' in out
assert 'Accept:*/*' in out
assert 'Authorization:LOW ' in out
assert 'Connection:close' in out
assert 'Content-Length:3' in out
assert 'Accept-Encoding:gzip, deflate' in out
def test_ia_upload_403(capsys):
s3_error = ('<Error>'
'<Code>SignatureDoesNotMatch</Code>'
'<Message>The request signature we calculated does not match '
'the signature you provided. Check your AWS Secret Access Key '
'and signing method. For more information, see REST '
'Authentication and SOAP Authentication for details.</Message>'
"<Resource>'PUT\n\n\n\n/iacli-test-item60/test-replace.txt'</Resource>"
'<RequestId>18a9c5ea-088f-42f5-9fcf-70651cc085ca</RequestId>'
'</Error>')
with IaRequestsMock() as rsps:
rsps.add_metadata_mock('nasa')
rsps.add(responses.PUT,
'{0}//s3.us.archive.org/nasa/test_ia_upload.py'.format(PROTOCOL),
body=s3_error,
status=403,
content_type='text/plain')
ia_call(['ia', 'upload', 'nasa', __file__], expected_exit_code=1)
out, err = capsys.readouterr()
assert 'error uploading test_ia_upload.py' in err
def test_ia_upload_invalid_cmd(capsys):
ia_call(['ia', 'upload', 'nasa', 'nofile.txt'], expected_exit_code=1)
out, err = capsys.readouterr()
assert '<file> should be a readable file or directory.' in err
def test_ia_upload_size_hint(capsys, tmpdir_ch, nasa_mocker):
with open('test.txt', 'w') as fh:
fh.write('foo')
ia_call(['ia', 'upload', '--debug', 'nasa', '--size-hint', '30', 'test.txt'])
out, err = capsys.readouterr()
assert 'User-Agent' in out
assert 's3.us.archive.org/nasa/test.txt' in out
assert 'x-archive-size-hint:30' in out
assert 'Accept:*/*' in out
assert 'Authorization:LOW ' in out
assert 'Connection:close' in out
assert 'Content-Length:3' in out
assert 'Accept-Encoding:gzip, deflate' in out
def test_ia_upload_unicode(tmpdir_ch, caplog):
with open('தமிழ் - baz ∆.txt', 'w') as fh:
fh.write('unicode foo')
efname = '%E0%AE%A4%E0%AE%AE%E0%AE%BF%E0%AE%B4%E0%AF%8D%20-%20baz%20%E2%88%86.txt'
with IaRequestsMock(assert_all_requests_are_fired=False) as rsps:
rsps.add_metadata_mock('nasa')
rsps.add(responses.PUT,
'{0}//s3.us.archive.org/nasa/{1}'.format(PROTOCOL, efname),
body='',
content_type='text/plain')
ia_call(['ia', '--log', 'upload', 'nasa', 'தமிழ் - baz ∆.txt',
'--metadata', 'foo:∆'])
assert (u'uploaded தமிழ் - baz ∆.txt to {0}//s3.us.archive.org/nasa/'
u'%E0%AE%A4%E0%AE%AE%E0%AE%BF%E0%AE%B4%E0%AF%8D%20-%20'
u'baz%20%E2%88%86.txt'.format(PROTOCOL)) in caplog.text
def test_ia_upload_remote_name(tmpdir_ch, caplog):
with open('test.txt', 'w') as fh:
fh.write('foo')
with IaRequestsMock() as rsps:
rsps.add_metadata_mock('nasa')
rsps.add(responses.PUT, '{0}//s3.us.archive.org/nasa/hi.txt'.format(PROTOCOL),
body='',
content_type='text/plain')
ia_call(['ia', '--log', 'upload', 'nasa', 'test.txt', '--remote-name',
'hi.txt'])
assert ('uploaded hi.txt to {0}//s3.us.archive.org/nasa/'
'hi.txt'.format(PROTOCOL)) in caplog.text
|
