Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/tests/test_objstorage_http.py
| Show All 20 Lines | def build_objstorage(): | ||||
| from the HTTPReadOnlyObjStorage to query the InMemoryStorage. | from the HTTPReadOnlyObjStorage to query the InMemoryStorage. | ||||
| Also fills the backend storage with a 100 objects. | Also fills the backend storage with a 100 objects. | ||||
| """ | """ | ||||
| sto_back = get_objstorage(cls="memory") | sto_back = get_objstorage(cls="memory") | ||||
| objids = [] | objids = [] | ||||
| for i in range(100): | for i in range(100): | ||||
| content = f"some content {i}".encode() | content = f"some content {i}".encode() | ||||
| objids.append(sto_back.add(content, obj_id=compute_hash(content))) | obj_id = compute_hash(content) | ||||
| objids.append(obj_id) | |||||
| sto_back.add(content, obj_id=obj_id) | |||||
| url = "http://127.0.0.1/content/" | url = "http://127.0.0.1/content/" | ||||
| sto_front = get_objstorage(cls="http", url=url) | sto_front = get_objstorage(cls="http", url=url) | ||||
| mock = fixture.Fixture() | mock = fixture.Fixture() | ||||
| mock.setUp() | mock.setUp() | ||||
| def get_cb(request, context): | def get_cb(request, context): | ||||
| dirname, basename = request.path.rsplit("/", 1) | dirname, basename = request.path.rsplit("/", 1) | ||||
| ▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Lines | |||||
| def test_http_objstorage_check(): | def test_http_objstorage_check(): | ||||
| sto_front, sto_back, objids = build_objstorage() | sto_front, sto_back, objids = build_objstorage() | ||||
| for objid in objids: | for objid in objids: | ||||
| assert sto_front.check(objid) is None # no Exception means OK | assert sto_front.check(objid) is None # no Exception means OK | ||||
| # create an invalid object in the in-memory objstorage | # create an invalid object in the in-memory objstorage | ||||
| invalid_content = b"p0wn3d content" | invalid_content = b"p0wn3d content" | ||||
| fake_objid = "\x01" * 20 | fake_objid = "\x01" * 20 | ||||
| id_added = sto_back.add(invalid_content, fake_objid) | sto_back.add(invalid_content, fake_objid) | ||||
| assert id_added == fake_objid | |||||
| # the http objstorage should report it as invalid | # the http objstorage should report it as invalid | ||||
| with pytest.raises(exc.Error): | with pytest.raises(exc.Error): | ||||
| sto_front.check(id_added) | sto_front.check(fake_objid) | ||||
| def test_http_objstorage_read_only(): | def test_http_objstorage_read_only(): | ||||
| sto_front, sto_back, objids = build_objstorage() | sto_front, sto_back, objids = build_objstorage() | ||||
| content = b"" | content = b"" | ||||
| obj_id = compute_hash(content) | obj_id = compute_hash(content) | ||||
| with pytest.raises(exc.ReadOnlyObjStorage): | with pytest.raises(exc.ReadOnlyObjStorage): | ||||
| Show All 20 Lines | |||||