aptly/system/t12_api/files.py

import inspect
import os
import shutil
import tempfile

from api_lib import APITest
from lib import BaseTest


class FilesAPITestUpload(APITest):
    """
    POST /files/:dir
    """

    def check(self):
        d = self.random_name()
        resp = self.upload("/api/files/" + d, "pyspi_0.6.1-1.3.dsc")
        self.check_equal(resp.status_code, 200)
        self.check_equal(resp.json(), [d + '/pyspi_0.6.1-1.3.dsc'])
        self.check_exists("upload/" + d + '/pyspi_0.6.1-1.3.dsc')


class FilesAPITestUploadMulti(APITest):
    """
    POST /files/:dir, GET /files/:dir multi files
    """

    def check(self):
        d = self.random_name()

        self.check_equal(self.get("/api/files/" + d).status_code, 404)

        resp = self.upload("/api/files/" + d, "pyspi_0.6.1-1.3.dsc", "pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz")
        self.check_equal(resp.status_code, 200)
        self.check_equal(sorted(resp.json()),
                         [d + '/pyspi_0.6.1-1.3.diff.gz', d + '/pyspi_0.6.1-1.3.dsc', d + '/pyspi_0.6.1.orig.tar.gz'])
        self.check_exists("upload/" + d + '/pyspi_0.6.1-1.3.dsc')
        self.check_exists("upload/" + d + '/pyspi_0.6.1-1.3.diff.gz')
        self.check_exists("upload/" + d + '/pyspi_0.6.1.orig.tar.gz')

        resp = self.get("/api/files/" + d)
        self.check_equal(resp.status_code, 200)
        self.check_equal(sorted(resp.json()),
                         ['pyspi_0.6.1-1.3.diff.gz', 'pyspi_0.6.1-1.3.dsc', 'pyspi_0.6.1.orig.tar.gz'])


class FilesAPITestList(APITest):
    """
    GET /files/
    """

    def check(self):
        d1, d2, d3 = self.random_name(), self.random_name(), self.random_name()

        resp = self.get("/api/files")
        self.check_equal(resp.status_code, 200)
        self.check_equal(resp.json(), [])

        self.check_equal(self.upload("/api/files/" + d1, "pyspi_0.6.1-1.3.dsc").status_code, 200)
        self.check_equal(self.upload("/api/files/" + d2, "pyspi_0.6.1-1.3.dsc").status_code, 200)
        self.check_equal(self.upload("/api/files/" + d3, "pyspi_0.6.1-1.3.dsc").status_code, 200)

        resp = self.get("/api/files")
        self.check_equal(resp.status_code, 200)
        self.check_equal(sorted(resp.json()), sorted([d1, d2, d3]))


class FilesAPITestDelete(APITest):
    """
    DELETE /files/:dir, DELETE /files/:dir/:name
    """

    def check(self):
        d1, d2 = self.random_name(), self.random_name()

        self.check_equal(self.get("/api/files").json(), [])
        self.check_equal(self.delete("/api/files/" + d1).status_code, 200)
        self.check_equal(self.delete("/api/files/" + d1 + "/" + "pyspi_0.6.1-1.3.dsc").status_code, 200)

        self.check_equal(self.upload("/api/files/" + d1, "pyspi_0.6.1-1.3.dsc").status_code, 200)
        self.check_equal(self.upload("/api/files/" + d2, "pyspi_0.6.1-1.3.dsc").status_code, 200)

        self.check_equal(self.delete("/api/files/" + d1).status_code, 200)
        self.check_equal(self.get("/api/files").json(), [d2])

        self.check_equal(self.delete("/api/files/" + d2 + "/" + "no-such-file").status_code, 200)
        self.check_equal(self.get("/api/files/" + d2).json(), ["pyspi_0.6.1-1.3.dsc"])

        self.check_equal(self.delete("/api/files/" + d2 + "/" + "pyspi_0.6.1-1.3.dsc").status_code, 200)
        self.check_equal(self.get("/api/files").json(), [d2])
        self.check_equal(self.get("/api/files/" + d2).json(), [])


class FilesAPITestSecurity(APITest):
    """
    delete & upload security
    """

    def check(self):
        self.check_equal(self.delete("/api/files/.").status_code, 404)
        self.check_equal(self.delete("/api/files").status_code, 404)
        self.check_equal(self.delete("/api/files/").status_code, 404)
        self.check_equal(self.delete("/api/files/../.").status_code, 404)
        self.check_equal(self.delete("/api/files/./..").status_code, 404)
        self.check_equal(self.delete("/api/files/dir/..").status_code, 404)


class FilesAPITestDputUpload(APITest):
    """
    PUT /api/files/:dir/:file via dput, then POST /api/repos/:name/include/:dir

    Uses the real dput binary to upload a .changes file and all its referenced
    files to the aptly API, then imports them into a local repo via include.
    Skipped if dput is not installed.
    """

    def fixture_available(self):
        return super().fixture_available() and shutil.which("dput") is not None

    def check(self):
        d = self.random_name()
        repo_name = self.random_name()

        # Create target repo
        self.check_equal(
            self.post("/api/repos", json={"Name": repo_name}).status_code, 201)

        changes_dir = os.path.join(
            os.path.dirname(inspect.getsourcefile(BaseTest)), "changes")
        changes_file = os.path.join(changes_dir, "hardlink_0.2.1_amd64.changes")

        # dput strips leading/trailing slashes from 'incoming' then prepends /,
        # producing: PUT http://{fqdn}/api/files/{d}/{filename}
        # fqdn includes host:port so dput connects directly to the test API server.
        dput_cf = (
            "[aptly]\n"
            f"fqdn     = {self.base_url}\n"
            "method   = http\n"
            f"incoming = api/files/{d}\n"
            "login    = *\n"
            "allow_unsigned_uploads = 1\n"
            "allow_dcut = 0\n"
        )

        tmpdir = tempfile.mkdtemp()
        try:
            dput_cf_path = os.path.join(tmpdir, "dput.cf")
            with open(dput_cf_path, "w") as f:
                f.write(dput_cf)

            # dput -U: allow unsigned uploads (skip local GPG check)
            # dput reads the .changes and PUTs every file listed in Files: + the .changes itself
            self.run_cmd(["dput", "-c", dput_cf_path, "-U", "aptly", changes_file])
        finally:
            shutil.rmtree(tmpdir)

        # All files referenced in the .changes must now be present in the upload dir
        self.check_exists(f"upload/{d}/hardlink_0.2.1_amd64.changes")
        self.check_exists(f"upload/{d}/hardlink_0.2.1.dsc")
        self.check_exists(f"upload/{d}/hardlink_0.2.1.tar.gz")
        self.check_exists(f"upload/{d}/hardlink_0.2.1_amd64.deb")

        # Import via the .changes file into the repo
        resp = self.post_task(
            f"/api/repos/{repo_name}/include/{d}",
            params={"ignoreSignature": 1})
        self.check_task(resp)

        output = self.get(f"/api/tasks/{resp.json()['ID']}/output")
        self.check_in(b"Added: hardlink_0.2.1_source added, hardlink_0.2.1_amd64 added", output.content)

        # Packages must be in the repo
        self.check_equal(
            sorted(self.get(f"/api/repos/{repo_name}/packages").json()),
            ["Pamd64 hardlink 0.2.1 daf8fcecbf8210ad", "Psource hardlink 0.2.1 8f72df429d7166e5"])

        # include cleans up the upload dir
        self.check_not_exists(f"upload/{d}")