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
|
import os
import responses
import internetarchive.session
from internetarchive import __version__
from tests.conftest import NASA_METADATA_PATH, PROTOCOL, IaRequestsMock
CONFIG = {
's3': {
'access': 'test_access',
'secret': 'test_secret',
},
'cookies': {
'logged-in-user': 'test%40example.com; path=/; domain=.archive.org',
'logged-in-sig': 'testsig; path=/; domain=.archive.org',
},
'logging': {
'level': 'INFO',
'file': 'test.log',
},
}
def test_archive_session(tmpdir):
tmpdir.chdir()
s = internetarchive.session.ArchiveSession(CONFIG)
assert os.path.isfile('test.log')
assert CONFIG == s.config
assert set(s.cookies.keys()) == set(CONFIG['cookies'].keys())
assert s.secure is True
assert s.protocol == PROTOCOL
assert s.access_key == 'test_access'
assert s.secret_key == 'test_secret'
assert s.headers['user-agent'].startswith(f'internetarchive/{__version__}')
def test_get_item(tmpdir):
tmpdir.chdir()
with open(NASA_METADATA_PATH) as fh:
item_metadata = fh.read().strip()
with responses.RequestsMock() as rsps:
rsps.add(responses.GET, f'{PROTOCOL}//archive.org/metadata/nasa',
body=item_metadata,
content_type='application/json')
s = internetarchive.session.ArchiveSession()
item = s.get_item('nasa')
assert item.exists is True
assert item.identifier == 'nasa'
with responses.RequestsMock() as rsps:
rsps.add(responses.GET, f'{PROTOCOL}//archive.org/metadata/nasa',
body=item_metadata,
status=400,
content_type='application/json')
s = internetarchive.session.ArchiveSession(CONFIG)
try:
item = s.get_item('nasa')
except Exception:
with open('test.log') as fh:
assert '400 Client Error' in fh.read()
def test_s3_is_overloaded():
test_body = """{
"accesskey": "test_access",
"bucket": "nasa",
"detail": {
"accesskey_ration": 74,
"accesskey_tasks_queued": 0,
"bucket_ration": 24,
"bucket_tasks_queued": 0,
"limit_reason": "",
"rationing_engaged": 0,
"rationing_level": 1399,
"total_global_limit": 1799,
"total_tasks_queued": 308
},
"over_limit": 0
}"""
with IaRequestsMock() as rsps:
rsps.add(responses.GET, f'{PROTOCOL}//s3.us.archive.org',
body=test_body,
content_type='application/json')
s = internetarchive.session.ArchiveSession(CONFIG)
r = s.s3_is_overloaded('nasa')
assert r is False
test_body = """{
"accesskey": "test_access",
"bucket": "nasa",
"detail": {
"accesskey_ration": 74,
"accesskey_tasks_queued": 0,
"bucket_ration": 24,
"bucket_tasks_queued": 0,
"limit_reason": "",
"rationing_engaged": 0,
"rationing_level": 1399,
"total_global_limit": 1799,
"total_tasks_queued": 308
},
"over_limit": 1
}"""
with responses.RequestsMock() as rsps:
rsps.add(responses.GET, f'{PROTOCOL}//s3.us.archive.org',
body=test_body,
content_type='application/json')
s = internetarchive.session.ArchiveSession(CONFIG)
r = s.s3_is_overloaded('nasa')
assert r is True
def test_cookies():
with responses.RequestsMock() as rsps:
rsps.add(responses.GET, f'{PROTOCOL}//archive.org')
s = internetarchive.session.ArchiveSession(CONFIG)
r = s.get(f'{PROTOCOL}//archive.org')
assert 'logged-in-sig' in r.request.headers['Cookie']
assert 'logged-in-user' in r.request.headers['Cookie']
for c in s.cookies:
if c.name.startswith('logged-in-'):
assert c.domain == '.archive.org'
with responses.RequestsMock() as rsps:
rsps.add(responses.GET, f'{PROTOCOL}//example.com')
s = internetarchive.session.ArchiveSession(CONFIG)
r = s.get(f'{PROTOCOL}//example.com')
assert 'logged-in-sig' not in r.request.headers.get('Cookie', '')
assert 'logged-in-user' not in r.request.headers.get('Cookie', '')
assert '.archive.org' not in r.request.headers.get('Cookie', '')
|
