458 lines
18 KiB
Python
458 lines
18 KiB
Python
import os
|
|
import json
|
|
from xml.etree import ElementTree as etree
|
|
from multiprocessing.pool import ThreadPool
|
|
|
|
import pytest
|
|
import mock
|
|
|
|
from utils import get_tests_file_content, FakedResponse
|
|
|
|
from petale.utils import etag
|
|
from petale.models import CUT, Petal
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def test_authentication_failure(app):
|
|
resp = app.get('/api/partner/12345abcde12345abcde12345abcde12/', status=401)
|
|
json.loads(resp.content)
|
|
|
|
|
|
def test_resource_not_found(app, partner_southpark, cut_kevin_uuid, acl):
|
|
app.authorization = ('Basic', ('family', 'family'))
|
|
cut_uuid = 'abcd1234' * 31
|
|
|
|
# test with invalid partner
|
|
resp = app.get('/api/partner/%s/whatever/' % cut_uuid, status=403)
|
|
assert resp.json['error'] == 'access-forbidden'
|
|
|
|
# test with cut uuid with length over 255
|
|
app.get('/api/southpark/%s11111111111/whatever/' % cut_uuid, status=404)
|
|
|
|
# test with invalid id cut
|
|
resp = app.get('/api/southpark/%s/invoices/' % cut_uuid, status=404)
|
|
assert resp.json['error'] == 'cut-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 with cut uuid length over 32
|
|
app.put_json('/api/southaprk/%s12/whatever/' % cut_kevin_uuid,
|
|
params=json.dumps(payload), status=404)
|
|
|
|
# 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 with If-Match
|
|
resp = app.delete(url, status=412, headers={'If-Match': '"xxx"'})
|
|
assert resp.json['error'] == 'concurrent-access'
|
|
|
|
# test key deletion with cut uuid length over 32
|
|
app.delete('/api/southpark/%s12/whatever/' % cut_kevin_uuid, status=404)
|
|
|
|
# test key deletion
|
|
resp = app.delete(url, status=204)
|
|
assert resp.content == b''
|
|
app.get(url, status=404)
|
|
|
|
# test keys purge
|
|
for idx in range(4):
|
|
url = '/api/southpark/%s/loans-%d/' % (cut_kevin_uuid, idx)
|
|
app.put_json(url, params=payload, headers={'If-None-Match': '*'}, status=201)
|
|
|
|
resp = app.get('/api/southpark/%s/' % cut_kevin_uuid, status=200)
|
|
assert len(resp.json['keys']) == 4
|
|
app.delete('/api/southpark/%s/' % cut_kevin_uuid, status=204)
|
|
resp = app.get('/api/southpark/%s/' % cut_kevin_uuid, status=200)
|
|
assert len(resp.json['keys']) == 0
|
|
|
|
|
|
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(list(xml_data)) == 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):
|
|
# gotham sizes:
|
|
# Global hard max: 20Ko
|
|
# Global soft max: 10Ko
|
|
# Per key hard max: 2Ko
|
|
# Per key soft max: 1Ko
|
|
app.authorization = ('Basic', ('arkham', 'arkham'))
|
|
|
|
# test sending data sized above per key hard max size
|
|
url = '/api/gotham/%s/taxes-fail/' % cut_kevin_uuid
|
|
resp = app.put(url,
|
|
params='a' * 3 * 1024, # 3ko
|
|
headers={'If-None-Match': '*'},
|
|
content_type='text/plain',
|
|
status=500)
|
|
assert resp.json['error'] == 'key-space-exhausted'
|
|
|
|
# test sending data sized within soft and hard key limit
|
|
resp = app.put(url,
|
|
params='a' * (1024 + 1), # 1ko + 1
|
|
headers={'If-None-Match': '*'},
|
|
content_type='text/plain',
|
|
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 '1025' in sent_mail.message().as_string()
|
|
assert '2048' in sent_mail.message().as_string()
|
|
|
|
for i in range(18):
|
|
url = '/api/gotham/%s/taxes-%d/' % (cut_kevin_uuid, i)
|
|
app.put(url,
|
|
params='a' * 1024,
|
|
headers={'If-None-Match': '*'},
|
|
content_type='text/plain',
|
|
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 '10241' in sent_mail.message().as_string()
|
|
assert str(1024 * 20) in sent_mail.message().as_string()
|
|
|
|
url = '/api/gotham/%s/taxes-100/' % cut_kevin_uuid
|
|
resp = app.put(url,
|
|
params='a' * (1024 * 2 - 1),
|
|
headers={'If-None-Match': '*'},
|
|
content_type='text/plain',
|
|
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)
|
|
|
|
|
|
@mock.patch('petale.authentication.requests.post')
|
|
def test_idp_based_partner_authentication(mocked_post, app, cut_kevin_uuid, acl):
|
|
app.authorization = ('Basic', ('library', 'whatever'))
|
|
payload = {"friends": [{"name": "Token", "age": 10}, {"name": "Kenny", "age": 10}]}
|
|
url = '/api/southpark/%s/profile-whatever/' % cut_kevin_uuid
|
|
# failure
|
|
response = {
|
|
"result": 0, "errors": ["Invalid username/password."]
|
|
}
|
|
mocked_post.return_value = FakedResponse(content=json.dumps(response))
|
|
app.put_json(url, params=payload, status=401, headers={'If-None-Match': '*'})
|
|
response = {"result": 1, "errors": []}
|
|
# test with RP
|
|
client_id = client_secret = 'a1b2' * 8
|
|
app.authorization = ('Basic', (client_id, client_secret))
|
|
mocked_post.return_value = FakedResponse(content=json.dumps(response))
|
|
app.put_json(url, params=payload, status=201, headers={'If-None-Match': '*'})
|
|
# success
|
|
app.authorization = ('Basic', ('library', 'whatever'))
|
|
mocked_post.return_value = FakedResponse(content=json.dumps(response))
|
|
url = '/api/southpark/%s/profile-whatever2/' % cut_kevin_uuid
|
|
app.put_json(url, params=payload, status=201, headers={'If-None-Match': '*'})
|
|
|
|
|
|
@mock.patch('petale.api_views.requests.post')
|
|
def test_cut_uuid_idp_checking(mocked_post, settings, app, acl):
|
|
cut_uuid = 'a1b2' * 8
|
|
app.authorization = ('Basic', ('library', 'library'))
|
|
payload = {"username": "t0k3n", "email": "t0k3n@deadsec.org"}
|
|
headers = {'Content-Type': 'application/json'}
|
|
|
|
# failure with AUTHENTIC_PETALE improperly defined
|
|
settings.PETALE_AUTHENTIC_URL = ''
|
|
resp = app.put_json('/api/southpark/%s/profile/' % cut_uuid, params=payload,
|
|
headers=headers, status=404)
|
|
assert resp.json["error"] == "cut-not-found"
|
|
# failure when cut uuid doesn't exist on idp
|
|
response = {"unknown_uuids": [cut_uuid], "result": 1}
|
|
mocked_post.return_value = FakedResponse(content=json.dumps(response),
|
|
status_code=200)
|
|
resp = app.put_json('/api/southpark/%s/profile/' % cut_uuid, params=payload,
|
|
headers=headers, status=404)
|
|
assert resp.json["error"] == "cut-not-found"
|
|
|
|
resp = app.get('/api/southpark/%s/' % cut_uuid, headers=headers, status=404)
|
|
assert resp.json["error"] == "cut-not-found"
|
|
|
|
resp = app.get('/api/southpark/%s/profile/' % cut_uuid, headers=headers, status=404)
|
|
assert resp.json["error"] == "cut-not-found"
|
|
|
|
# sucess
|
|
settings.PETALE_AUTHENTIC_URL = 'http://example.net/idp/'
|
|
response = {"unknown_uuids": [], "result": 1}
|
|
mocked_post.return_value = FakedResponse(content=json.dumps(response),
|
|
status_code=200)
|
|
|
|
resp = app.get('/api/southpark/%s/profile/' % cut_uuid, headers=headers, status=404)
|
|
assert resp.json["error"] == "key-not-found"
|
|
|
|
resp = app.get('/api/southpark/%s/' % cut_uuid, headers=headers)
|
|
assert resp.json["keys"] == []
|
|
|
|
resp = app.put_json('/api/southpark/%s/profile/' % cut_uuid, params=payload,
|
|
headers=headers, status=201)
|
|
assert CUT.objects.get(uuid=cut_uuid)
|
|
assert Petal.objects.get(name='profile', cut__uuid=cut_uuid, partner__name='southpark')
|
|
|
|
|
|
def test_storage_error(app, partner_southpark, cut_kevin_uuid, acl, caplog):
|
|
app.authorization = ('Basic', ('library', 'library'))
|
|
|
|
payload = json.loads(get_tests_file_content('books.json'))
|
|
url = '/api/southpark/%s/loans/' % cut_kevin_uuid
|
|
|
|
app.put_json(url, params=payload, headers={'If-None-Match': '*'}, status=201)
|
|
|
|
os.unlink(Petal.objects.get().data.path)
|
|
|
|
app.get(url, status=500)
|
|
|
|
|
|
def test_concurrent_put(app, transactional_db):
|
|
'''Test concurrent PUT to the same key'''
|
|
from utils import create_cut, create_partner, create_service, create_acl_record
|
|
|
|
uuid = create_cut('a' * 255).uuid
|
|
southpark = create_partner('southpark', hg=20240, hk=19728)
|
|
library = create_service('library')
|
|
create_acl_record(1, southpark, library, 'loans', methods='GET,HEAD,PUT,DELETE')
|
|
|
|
app.authorization = ('Basic', ('library', 'library'))
|
|
|
|
payload = json.loads(get_tests_file_content('books.json'))
|
|
url = '/api/%s/%s/loans/' % (southpark.name, uuid)
|
|
|
|
pool_count = 40
|
|
put_count = 60
|
|
|
|
pool = ThreadPool(pool_count)
|
|
|
|
def f(i):
|
|
from django.db import connection
|
|
if i % 2 == 0:
|
|
headers = {'If-None-Match': '*'}
|
|
else:
|
|
headers = {}
|
|
|
|
response = app.put_json(url, params=payload, headers=headers, status='*')
|
|
assert response.status_code in (412, 201, 200)
|
|
connection.close()
|
|
return i, response.status_code
|
|
|
|
l = pool.map(f, range(put_count))
|
|
assert len(l) == put_count
|