from base64 import b64decode, b64encode import pytest from webtest import TestApp import paste.db from paste import application from paste.store import Auth DB = "file::memory:?cache=shared" @pytest.fixture def db(): with paste.db.connect(DB) as d: yield d @pytest.fixture def app(db): _ = db app = TestApp(application, extra_environ={"HTTP_HOST": "localhost", "PASTE_DB": DB}) yield app @pytest.fixture def token(db): return b64encode(Auth(db).generate_token()).decode() @pytest.mark.parametrize("method", ["put", "post", "delete"]) def test_without_apikey(app, method): res = getattr(app, method)( "/test_key", "Hello, World!", headers={"Content-Type": "text/plain"}, expect_errors=True, ) assert res.status == "401 Unauthorized" assert res.headers["WWW-Authenticate"] == "APIKey" @pytest.mark.parametrize("method", ["put", "post", "delete"]) def test_malformed_authorization_header(app, method): res = getattr(app, method)( "/test_key", "Hello, World!", headers={"Content-Type": "text/plain", "Authorization": "malformed"}, expect_errors=True, ) assert res.status == "401 Unauthorized" assert res.headers["WWW-Authenticate"] == "APIKey" @pytest.mark.parametrize("method", ["put", "post", "delete"]) def test_malformed_apikey(app, method): res = getattr(app, method)( "/test_key" "Hello, World!", headers={"Content-Type": "text/plain", "Authorization": "APIKey malformed"}, expect_errors=True, ) assert res.status == "401 Unauthorized" assert res.headers["WWW-Authenticate"] == "APIKey" @pytest.mark.parametrize("method", ["put", "post", "delete"]) def test_invalid_apikey(app, method, token): invalid = b64encode(bytes(b ^ 13 for b in b64decode(token))).decode() res = getattr(app, method)( "/test_key" "Hello, World!", headers={"Content-Type": "text/plain", "Authorization": f"APIKey {invalid}"}, expect_errors=True, ) assert res.status == "401 Unauthorized" assert res.headers["WWW-Authenticate"] == "APIKey" def test_put(app, token): res = app.put( "/test_key", "Hello, World!", headers={"Content-Type": "text/plain", "Authorization": f"APIKey {token}"}, ) assert res.status == "201 Created" assert res.headers["Location"] == res.request.url def test_put_twice(app, token): res = app.put( "/test_key", "Hello, World!", headers={"Content-Type": "text/plain", "Authorization": f"APIKey {token}"}, ) assert res.status == "201 Created" assert res.headers["Location"] == res.request.url res = app.put( "/test_key", "Hello, World!", headers={"Content-Type": "text/plain", "Authorization": f"APIKey {token}"}, ) assert res.status == "204 No Content" assert res.headers["Location"] == res.request.url @pytest.mark.parametrize("method", ["get", "head", "delete"]) def test_method_nonexistent_fails(app, method, token): headers = {} if method == "delete": headers = {"Authorization": f"APIKey {token}"} res = getattr(app, method)("/test_key", headers=headers, expect_errors=True) assert res.status == "404 Not Found" def test_put_then_get_then_head(app, token): BODY = "Hello, World!" res = app.put( "/test_key", BODY, headers={"Content-Type": "text/plain", "Authorization": f"APIKey {token}"}, ) assert res.status == "201 Created" assert res.headers["Location"] == res.request.url etag = res.headers["ETag"] res = app.get("/test_key") assert res.status == "200 OK" assert res.headers["ETag"] == etag assert res.text == BODY res = app.head("/test_key") assert res.status == "200 OK" assert res.headers["ETag"] == etag assert res.text == "" def test_if_none_match(app, token): res = app.put( "/test_key", "Hello, World!", headers={"Content-Type": "text/plain", "Authorization": f"APIKey {token}"}, ) assert res.status == "201 Created" assert res.headers["Location"] == res.request.url etag = res.headers["ETag"] res = app.get("/test_key", headers={"If-None-Match": etag}) assert res.status == "304 Not Modified" assert res.headers["ETag"] == etag def test_if_none_match_other_etag(app, token): BODY = "Hello, World!" res = app.put( "/test_key", BODY, headers={"Content-Type": "text/plain", "Authorization": f"APIKey {token}"}, ) assert res.status == "201 Created" assert res.headers["Location"] == res.request.url etag = res.headers["ETag"] res = app.get("/test_key", headers={"If-None-Match": '"not a real etag"'}) assert res.status == "200 OK" assert res.headers["ETag"] == etag assert res.text == BODY def test_if_none_match_malformed_etag(app, token): res = app.put( "/test_key", "Hello, World!", headers={"Content-Type": "text/plain", "Authorization": f"APIKey {token}"}, ) assert res.status == "201 Created" assert res.headers["Location"] == res.request.url res = app.get( "/test_key", headers={"If-None-Match": "malformed"}, expect_errors=True ) assert res.status == "400 Bad Request" @pytest.mark.parametrize( "etags", [ [None, "a"], ["a", None], [None, "a", "b"], ["a", None, "b"], ["a", "b", None], ], ) def test_if_none_match_list(app, etags, token): res = app.put( "/test_key", "Hello, World!", headers={"Content-Type": "text/plain", "Authorization": f"APIKey {token}"}, ) assert res.status == "201 Created" assert res.headers["Location"] == res.request.url etag = res.headers["ETag"] etags_str = ", ".join(f'"{e}"' if e else etag for e in etags) res = app.get("/test_key", headers={"If-None-Match": etags_str}) assert res.status == "304 Not Modified" assert res.headers["ETag"] == etag def test_put_update(app, token): res = app.put( "/test_key", "Hello, World!", headers={"Content-Type": "text/plain", "Authorization": f"APIKey {token}"}, ) assert res.status == "201 Created" assert res.headers["Location"] == res.request.url etag = res.headers["ETag"] res = app.put( "/test_key", "Hello, Updated World!", headers={"Content-Type": "text/plain", "Authorization": f"APIKey {token}"}, ) assert res.status == "204 No Content" assert res.headers["ETag"] != etag res = app.get("/test_key") assert res.status == "200 OK" assert res.text == "Hello, Updated World!" def test_delete(app, token): res = app.put( "/test_key", "Hello, World!", headers={"Content-Type": "text/plain", "Authorization": f"APIKey {token}"}, ) assert res.status == "201 Created" assert res.headers["Location"] == res.request.url res = app.delete("/test_key", expect_errors=True) assert res.status == "401 Unauthorized" res = app.delete("/test_key", headers={"Authorization": f"APIKey {token}"}) assert res.status == "204 No Content" res = app.delete( "/test_key", headers={"Authorization": f"APIKey {token}"}, expect_errors=True ) assert res.status == "404 Not Found" def test_post(app, token): res = app.post( "/test_key", "Hello, World!", headers={"Content-Type": "text/plain", "Authorization": f"APIKey {token}"}, ) assert res.status == "201 Created" assert res.headers["Location"] != res.request.url print(res.headers["Location"]) print(res.request.url) assert res.headers["Location"].startswith(res.request.url) etag = res.headers["ETag"] res = app.get(res.headers["Location"]) assert res.status == "200 OK" assert res.headers["ETag"] == etag