Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/tests/test_objstorage_http.py
- This file was added.
# Copyright (C) 2021 The Software Heritage developers | |||||
# See the AUTHORS file at the top-level directory of this distribution | |||||
# License: GNU General Public License version 3, or any later version | |||||
# See top-level LICENSE file for more information | |||||
import pytest | |||||
import requests_mock | |||||
from requests_mock.contrib import fixture | |||||
from swh.objstorage import exc | |||||
from swh.objstorage.factory import get_objstorage | |||||
def build_objstorage(): | |||||
"""Build an HTTPReadOnlyObjStorage suitable for tests | |||||
this instancaite 2 ObjStorage, one HTTPReadOnlyObjStorage (the "front" one | |||||
being under test), and one InMemoryObjStorage (which actually stores the | |||||
test content), and install a request mock fixture to route HTTP requests | |||||
from the HTTPReadOnlyObjStorage to query the InMemoryStorage. | |||||
Also fills the backend storage with a 100 objects. | |||||
""" | |||||
sto_back = get_objstorage(cls="memory") | |||||
objids = [] | |||||
for i in range(100): | |||||
objids.append(sto_back.add(f"some content {i}".encode())) | |||||
url = "http://127.0.0.1/content/" | |||||
sto_front = get_objstorage(cls="http", url=url) | |||||
mock = fixture.Fixture() | |||||
mock.setUp() | |||||
def get_cb(request, context): | |||||
dirname, basename = request.path.rsplit("/", 1) | |||||
objid = bytes.fromhex(basename) | |||||
if dirname == "/content" and objid in sto_back: | |||||
return sto_back.get(objid) | |||||
context.status_code = 404 | |||||
vlorentz: not covered? | |||||
Done Inline Actionsindeed, it's covered for the HEAD query but not the GET. Will fix douardda: indeed, it's covered for the HEAD query but not the GET. Will fix | |||||
def head_cb(request, context): | |||||
dirname, basename = request.path.rsplit("/", 1) | |||||
objid = bytes.fromhex(basename) | |||||
if dirname != "/content" or objid not in sto_back: | |||||
context.status_code = 404 | |||||
return b"Not Found" | |||||
return b"Found" | |||||
mock.register_uri(requests_mock.GET, requests_mock.ANY, content=get_cb) | |||||
mock.register_uri(requests_mock.HEAD, requests_mock.ANY, content=head_cb) | |||||
return sto_front, sto_back, objids | |||||
def test_http_objstorage(): | |||||
sto_front, sto_back, objids = build_objstorage() | |||||
for objid in objids: | |||||
assert objid in sto_front | |||||
assert sto_front.get(objid) == sto_back.get(objid) | |||||
assert sto_front.get(objid).decode().startswith("some content ") | |||||
def test_http_objstorage_missing(): | |||||
sto_front, sto_back, objids = build_objstorage() | |||||
assert b"\x00" * 20 not in sto_front | |||||
def test_http_objstorage_get_missing(): | |||||
sto_front, sto_back, objids = build_objstorage() | |||||
with pytest.raises(exc.ObjNotFoundError): | |||||
sto_front.get(b"\x00" * 20) | |||||
def test_http_objstorage_check(): | |||||
sto_front, sto_back, objids = build_objstorage() | |||||
for objid in objids: | |||||
assert sto_front.check(objid) is None # no Exception means OK | |||||
# create an invalid object in the in-memory objstorage | |||||
invalid_content = b"p0wn3d content" | |||||
fake_objid = "\x01" * 20 | |||||
id_added = sto_back.add(invalid_content, fake_objid) | |||||
assert id_added == fake_objid | |||||
# the http objstorage should report it as invalid | |||||
with pytest.raises(exc.Error): | |||||
sto_front.check(id_added) | |||||
def test_http_objstorage_read_only(): | |||||
sto_front, sto_back, objids = build_objstorage() | |||||
with pytest.raises(exc.ReadOnlyObjStorage): | |||||
sto_front.add(b"") | |||||
with pytest.raises(exc.ReadOnlyObjStorage): | |||||
sto_front.restore(b"") | |||||
with pytest.raises(exc.ReadOnlyObjStorage): | |||||
sto_front.delete(b"\x00" * 20) | |||||
def test_http_objstorage_not_iterable(): | |||||
sto_front, sto_back, objids = build_objstorage() | |||||
with pytest.raises(exc.NonIterableObjStorage): | |||||
len(sto_front) | |||||
with pytest.raises(exc.NonIterableObjStorage): | |||||
iter(sto_front) |
not covered?