297 lines
11 KiB
Python
297 lines
11 KiB
Python
import json
|
|
from xml.etree import ElementTree as etree
|
|
|
|
import pytest
|
|
|
|
from utils import get_tests_file_content
|
|
|
|
from petale.utils import etag
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def test_authentication_failure(app):
|
|
resp = app.get('/api/partner/cut/', status=401)
|
|
json.loads(resp.content)
|
|
|
|
|
|
def test_resource_not_found(app, partner_southpark, cut_kevin_uuid, acl):
|
|
app.authorization = ('Basic', ('family', 'family'))
|
|
|
|
# test with invalid partner
|
|
resp = app.get('/api/partner/12345/whatever/', status=403)
|
|
assert resp.json['error'] == 'access-forbidden'
|
|
|
|
# test with invalid id cut
|
|
resp = app.get('/api/southpark/12345/invoices/', status=404)
|
|
assert resp.json['error'] == 'key-not-found'
|
|
|
|
# test with invalid key
|
|
url = '/api/southpark/%s/invoices/' % cut_kevin_uuid
|
|
resp = app.get(url, status=404)
|
|
assert resp.json['error'] == 'key-not-found'
|
|
|
|
app.head(url, status=404)
|
|
|
|
|
|
def test_access_control_list(app, partner_southpark, cut_kevin_uuid,
|
|
petal_books, petal_invoice, acl):
|
|
app.authorization = ('Basic', ('arkham', 'arkham'))
|
|
|
|
# test permission on requested partner
|
|
resp = app.get('/api/southpark/%s/' % cut_kevin_uuid, status=200)
|
|
|
|
# test permission on method
|
|
app.authorization = ('Basic', ('library', 'library'))
|
|
resp = app.put('/api/southpark/%s/books/' % cut_kevin_uuid, status=403)
|
|
assert resp.json['error'] == 'access-forbidden'
|
|
|
|
# test permission on key
|
|
resp = app.get('/api/southpark/%s/invoices/' % cut_kevin_uuid, status=403)
|
|
assert resp.json['error'] == 'access-forbidden'
|
|
|
|
#
|
|
app.authorization = ('Basic', ('library', 'library'))
|
|
payload = json.loads(get_tests_file_content('books.json'))
|
|
url = '/api/southpark/%s/loans/' % cut_kevin_uuid
|
|
resp = app.put_json(url, params=payload, headers={'If-None-Match': '*'}, status=201)
|
|
|
|
app.authorization = ('Basic', ('arkham', 'arkham'))
|
|
resp = app.get(url, status=403)
|
|
assert resp.json['error'] == 'access-forbidden'
|
|
|
|
|
|
def test_simple_api(app, partner_southpark, cut_kevin_uuid, acl, petal_invoice):
|
|
app.authorization = ('Basic', ('library', 'library'))
|
|
|
|
payload = json.loads(get_tests_file_content('books.json'))
|
|
url = '/api/southpark/%s/loans/' % cut_kevin_uuid
|
|
|
|
# test create key without content-type
|
|
resp = app.put(
|
|
url, params=json.dumps(payload),
|
|
headers={
|
|
'If-None-Match': '*',
|
|
'Content-Type': '',
|
|
}, status=400)
|
|
assert resp.json['error'] == 'missing-content-type'
|
|
|
|
# test create key
|
|
resp = app.put_json(url, params=payload, headers={'If-None-Match': '*'}, status=201)
|
|
assert resp.headers['ETag'] == etag(json.dumps(payload))
|
|
cached_etag = resp.headers['Etag']
|
|
|
|
# test head
|
|
resp = app.head(url, status=200)
|
|
assert resp.headers['ETag'] == etag(json.dumps(payload))
|
|
assert resp.headers['Content-Type'] == 'application/json'
|
|
assert resp.headers['Content-Length'] == '639'
|
|
|
|
# test get key by sending etag
|
|
resp = app.get(url, headers={'If-None-Match': cached_etag}, status=304)
|
|
|
|
# test get key
|
|
resp = app.get(url, status=200)
|
|
assert resp.headers['Content-Length'] == '639'
|
|
assert resp.headers['Content-Type'] == 'application/json'
|
|
assert resp.json == payload
|
|
|
|
cut_keys_url = '/api/southpark/%s/' % cut_kevin_uuid
|
|
# test get all keys
|
|
resp = app.head(cut_keys_url, status=405)
|
|
|
|
resp = app.get(cut_keys_url, status=200)
|
|
assert set(resp.json['keys']) == set(['loans'])
|
|
|
|
# test get all keys and prefix
|
|
resp = app.get('%s?prefix=invoice' % cut_keys_url, status=200)
|
|
assert resp.json['keys'] == []
|
|
|
|
resp = app.get('%s?prefix=whatever' % cut_keys_url, status=200)
|
|
assert resp.json['keys'] == []
|
|
|
|
# test key deletion
|
|
resp = app.delete(url, status=204)
|
|
assert resp.content == ''
|
|
app.get(url, status=404)
|
|
|
|
|
|
def test_binary_data(app, partner_southpark, cut_kevin_uuid, acl):
|
|
app.authorization = ('Basic', ('library', 'library'))
|
|
|
|
# test create xml data
|
|
url = '/api/southpark/%s/profile-friends/' % cut_kevin_uuid
|
|
content_type = 'text/xml'
|
|
resp = app.put(url, params=get_tests_file_content('users.xml'),
|
|
headers={'If-None-Match': '*', 'Content-Type': content_type}, status=201)
|
|
resp = app.get(url)
|
|
assert resp.headers.get('Content-Type') == content_type
|
|
xml_data = etree.fromstring(resp.content)
|
|
assert xml_data.tag == 'friends'
|
|
assert len(xml_data.getchildren()) == 4
|
|
|
|
# test update by changing format
|
|
payload = {"friends": [{"name": "Token", "age": 10}, {"name": "Kenny", "age": 10}]}
|
|
resp = app.put_json(url, params=payload, status=200)
|
|
resp = app.get(url, status=200)
|
|
data = json.loads(resp.content)
|
|
for datum in data['friends']:
|
|
assert datum['name'] in ('Token', 'Kenny')
|
|
assert datum['age'] == 10
|
|
|
|
# test create binary data
|
|
partner_southpark.hard_global_max_size = 1000000000
|
|
partner_southpark.hard_per_key_max_size = 1000000000
|
|
partner_southpark.save()
|
|
url = '/api/southpark/%s/profile-picture/' % cut_kevin_uuid
|
|
content_type = 'application/octet-stream'
|
|
content = get_tests_file_content('fg.jpg') * 100
|
|
resp = app.put(url, params=content,
|
|
headers={'If-None-Match': '*', 'Content-Type': content_type}, status=201)
|
|
resp = app.get(url)
|
|
assert resp.headers.get('Content-Type') == content_type
|
|
assert resp.content == content
|
|
|
|
# test create binary data
|
|
url = '/api/southpark/%s/profile-invoice/' % cut_kevin_uuid
|
|
content_type = 'application/pdf'
|
|
resp = app.put(url, params=get_tests_file_content('invoice.pdf'),
|
|
headers={'If-None-Match': '*', 'Content-Type': content_type}, status=201)
|
|
resp = app.get(url)
|
|
assert resp.headers.get('Content-Type') == content_type
|
|
|
|
|
|
def test_caching(app, partner_southpark, cut_kevin_uuid, acl):
|
|
payload = {
|
|
"favourites": [
|
|
{
|
|
"name": "bus stations",
|
|
"url": "https://southpark.com/bus/station",
|
|
"items": ["33", "42"],
|
|
},
|
|
{
|
|
"name": "weather",
|
|
"url": "https://southpark.com/weather/",
|
|
},
|
|
]
|
|
}
|
|
|
|
url = '/api/southpark/%s/favourites/' % cut_kevin_uuid
|
|
app.authorization = ('Basic', ('cityhall', 'cityhall'))
|
|
resp = app.put_json(url, params=payload, headers={'If-None-Match': '*'}, status=201)
|
|
cache = resp.headers['ETag']
|
|
|
|
# try to create the same data
|
|
app.authorization = ('Basic', ('cityhall', 'cityhall'))
|
|
resp = app.put_json(url, params=payload, headers={'If-None-Match': '*'}, status=412)
|
|
data = json.loads(resp.content)
|
|
assert resp.json['error'] == 'concurrent-access'
|
|
|
|
# try to create by sending list of eatgs
|
|
app.authorization = ('Basic', ('cityhall', 'cityhall'))
|
|
etags = [
|
|
'"sha1:5feceb66ffc86f38d952786c6d696c79c2dbc239dd4e91b46729d73a27fb57e9"',
|
|
'"sha1:c75ac194bf3ed4ef3f3e14343585c2fd7ed9b06bbdfbf0eb9f817b7337b966ea" ',
|
|
' "sha1:6b86b273ff34fce19d6b804eff5a3f5747ada4eaa22f1d49c01e52ddb7875b4b"'
|
|
]
|
|
etags.append(cache)
|
|
resp = app.put_json(url, params=payload, headers={'If-None-Match': ','.join(etags)}, status=412)
|
|
data = json.loads(resp.content)
|
|
assert resp.json['error'] == 'concurrent-access'
|
|
|
|
# update from the library service
|
|
app.authorization = ('Basic', ('cityhall', 'cityhall'))
|
|
resp = app.get(url, status=200)
|
|
data = json.loads(resp.content)
|
|
data['favourites'].append({
|
|
"name": "best books",
|
|
"url": "https://southpark.com/library",
|
|
"items": ["Guide To Life by E. Cartman", "Gingers Have Sools by Kyle"]
|
|
})
|
|
|
|
etag = resp.headers['Etag']
|
|
resp = app.put_json(url, params=data, headers={'If-Match': etag}, status=200)
|
|
|
|
# update attempts from the cityhall
|
|
app.authorization = ('Basic', ('cityhall', 'cityhall'))
|
|
new_item = {
|
|
"name": "Green zones",
|
|
"url": "https://southpark.com/parks",
|
|
"items": ["Main Street", "Emo Kids Street"]
|
|
}
|
|
payload['favourites'].append(new_item)
|
|
resp = app.put_json(url, params=payload, headers={'If-Match': cache}, status=412)
|
|
assert resp.json['error'] == 'concurrent-access'
|
|
|
|
resp = app.get(url, status=200)
|
|
data = json.loads(resp.content)
|
|
data['favourites'].append(new_item)
|
|
etag = resp.headers['Etag']
|
|
resp = app.put_json(url, params=payload, headers={'If-Match': etag}, status=200)
|
|
|
|
|
|
def test_partner_size_limit(app, cut_kevin_uuid, acl, petal_invoice, petal_books, mailoutbox):
|
|
app.authorization = ('Basic', ('arkham', 'arkham'))
|
|
payload = json.loads(get_tests_file_content('taxe.json'))
|
|
|
|
# test sending data sized above key limits
|
|
payload['phonenumber'] = '+18855776644'
|
|
payload['birthdate'] = '1980/06/21'
|
|
payload['birthplace'] = 'Chelsea, Quebec'
|
|
url = '/api/gotham/%s/taxes-fail/' % cut_kevin_uuid
|
|
resp = app.put_json(url, params=payload, headers={'If-None-Match': '*'}, status=500)
|
|
assert resp.json['error'] == 'key-space-exhausted'
|
|
|
|
# test sending data sized within soft and hard key limit
|
|
payload.pop('birthplace')
|
|
resp = app.put_json(url, params=payload, headers={'If-None-Match': '*'}, status=201)
|
|
assert len(mailoutbox) == 1
|
|
sent_mail = mailoutbox[0]
|
|
assert sent_mail.to[0] == 'b.wayne@gotham.gov'
|
|
assert 'taxes-fail' in sent_mail.subject
|
|
assert 'gotham' in sent_mail.subject
|
|
assert '395' in sent_mail.message().as_string()
|
|
assert '400' in sent_mail.message().as_string()
|
|
|
|
payload.pop('birthdate')
|
|
for i in range(4):
|
|
url = '/api/gotham/%s/taxes-%d/' % (cut_kevin_uuid, i)
|
|
app.put_json(url, params=payload, headers={'If-None-Match': '*'}, status=201)
|
|
|
|
assert len(mailoutbox) == 2
|
|
sent_mail = mailoutbox[1]
|
|
assert sent_mail.to[0] == 'b.wayne@gotham.gov'
|
|
assert 'gotham' in sent_mail.subject
|
|
assert '1867' in sent_mail.message().as_string()
|
|
assert '2048' in sent_mail.message().as_string()
|
|
|
|
url = '/api/gotham/%s/taxes-4/' % cut_kevin_uuid
|
|
resp = app.put_json(url, params=payload, headers={'If-None-Match': '*'}, status=500)
|
|
assert resp.json['error'] == 'global-space-exhausted'
|
|
|
|
|
|
def test_api_logging(caplog, app, cut_kevin_uuid, acl):
|
|
app.authorization = ('Basic', ('library', 'library'))
|
|
payload = {"friends": [{"name": "Token", "age": 10}, {"name": "Kenny", "age": 10}]}
|
|
url = '/api/southpark/%s/profile-friends/' % cut_kevin_uuid
|
|
resp = app.put_json(url, params=payload, status=201, headers={'If-None-Match': '*'})
|
|
|
|
records = [record for record in caplog.records() if record.name == 'petale']
|
|
assert len(records) == 5
|
|
for record in records:
|
|
assert record.name == 'petale'
|
|
if getattr(record, 'request_url', None):
|
|
assert url in record.request_url
|
|
if getattr(record, 'request_headers', None):
|
|
assert 'http_if_none_match' in record.request_headers.lower()
|
|
if getattr(record, 'request_body', None):
|
|
assert 'Token' in record.request_body
|
|
if getattr(record, 'response_headers', None):
|
|
assert resp.headers['Etag'] in record.response_headers
|
|
if getattr(record, 'response_body', None):
|
|
assert '{}' in record.response_body
|
|
if getattr(record, 'response_status_code', None):
|
|
record.response_status_code == 201
|
|
|
|
app.get('/api/southpark/%s/' % cut_kevin_uuid)
|