Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/api/test_deposit_private_read_archive.py
| # Copyright (C) 2017-2019 The Software Heritage developers | # Copyright (C) 2017-2021 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import io | import io | ||||
| from os.path import exists, join | |||||
| import zipfile | import zipfile | ||||
| from django.urls import reverse | from django.urls import reverse | ||||
| from rest_framework import status | from rest_framework import status | ||||
| from swh.deposit.api.private.deposit_read import aggregate_tarballs | |||||
| from swh.deposit.config import EM_IRI, PRIVATE_GET_RAW_CONTENT | from swh.deposit.config import EM_IRI, PRIVATE_GET_RAW_CONTENT | ||||
| from swh.deposit.tests.common import create_arborescence_archive | from swh.deposit.tests.common import create_arborescence_archive | ||||
| PRIVATE_GET_RAW_CONTENT_NC = PRIVATE_GET_RAW_CONTENT + "-nc" | PRIVATE_GET_RAW_CONTENT_NC = PRIVATE_GET_RAW_CONTENT + "-nc" | ||||
| def private_get_raw_url_endpoints(collection, deposit): | def private_get_raw_url_endpoints(collection, deposit): | ||||
| """There are 2 endpoints to check (one with collection, one without)""" | """There are 2 endpoints to check (one with collection, one without)""" | ||||
| ▲ Show 20 Lines • Show All 59 Lines • ▼ Show 20 Lines | for url in private_get_raw_url_endpoints(deposit_collection, deposit): | ||||
| assert r._headers["content-type"][1] == "application/zip" | assert r._headers["content-type"][1] == "application/zip" | ||||
| # read the stream | # read the stream | ||||
| data = b"".join(r.streaming_content) | data = b"".join(r.streaming_content) | ||||
| # extract the file from the zip | # extract the file from the zip | ||||
| zfile = zipfile.ZipFile(io.BytesIO(data)) | zfile = zipfile.ZipFile(io.BytesIO(data)) | ||||
| assert set(zfile.namelist()) == {"file1", "file2"} | assert set(zfile.namelist()) == {"file1", "file2"} | ||||
| assert zfile.open("file1").read() == b"some content in file" | assert zfile.open("file1").read() == b"some content in file" | ||||
| assert zfile.open("file2").read() == b"some other content in file" | assert zfile.open("file2").read() == b"some other content in file" | ||||
| def test_aggregate_tarballs_with_strange_archive(datadir, tmp_path): | |||||
| archive = join(datadir, "archives", "single-artifact-package.tar.gz") | |||||
| with aggregate_tarballs(tmp_path, [archive]) as tarball_path: | |||||
| assert exists(tarball_path) | |||||