Differential D6521 Diff 23705 swh/objstorage/tests/test_objstorage_http.py

Changeset 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():
				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

				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"

				vlorentzUnsubmitted Done Inline Actions not covered? vlorentz: not covered?
				douarddaAuthorUnsubmitted Done Inline Actions indeed, 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
				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_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_not_implemented():
				sto_front, sto_back, objids = build_objstorage()

				with pytest.raises(NotImplementedError):
				sto_front.add(b"")
				with pytest.raises(NotImplementedError):
				sto_front.restore(b"")
				with pytest.raises(NotImplementedError):
				sto_front.delete(b"\x00" * 20)
				with pytest.raises(NotImplementedError):
				len(sto_front)
				with pytest.raises(NotImplementedError):
				iter(sto_front)