OpenConceptLab/ocl_issues#2864 | export download saves under the API's name by default

README.md

@@ -191,8 +191,9 @@ ocl repo export status OWNER REPO VERSION --type source|collection
 # Trigger export creation (if not already cached)
 ocl repo export create OWNER REPO VERSION --type source|collection
 
-# Download to a local file
-ocl repo export download OWNER REPO VERSION --type source|collection -o FILENAME
+# Download, saved under the API's name; never replaces an existing file
+# (-o FILE saves it as FILE, overwriting; -o DIR saves it in that existing directory)
+ocl repo export download OWNER REPO VERSION --type source|collection [-o FILE|DIR]
 
 # Delete a cached export
 ocl repo export delete OWNER REPO VERSION --type source|collection
@@ -202,9 +203,12 @@ Example — download the latest CIEL release export:
 
 ```bash
 ocl repo export status CIEL CIEL v2026-03-23 --type source
-ocl repo export download CIEL CIEL v2026-03-23 --type source -o CIEL_v2026-03-23.zip
+ocl repo export download CIEL CIEL v2026-03-23 --type source
+# Saved to orgs_CIEL_sources_CIEL_v2026-03-23_<lastUpdated>.zip
 ```
 
+The API names exports `[owner-type]_[owner]_[repo-type]_[repo]_[version]_[expansion]_[lastUpdated].zip`, with the expansion when the export has one. An older export the API can't name keeps its storage name; if no usable name comes back at all, the command asks for `-o FILE`.
+
 ### Concepts
 
 ```bash

src/ocl_cli/commands/export.py

@@ -1,6 +1,9 @@
 """Export commands: status, create, delete, download."""
 
-import sys
+import errno
+import os
+import re
+from urllib.parse import parse_qs, unquote, urlsplit
 
 import click
 
@@ -80,26 +83,111 @@ def delete(ctx, owner, repo, version, repo_type, owner_type):
         handle_api_error(e)
 
 
+# OCL names (and so export and storage names) use only these characters; a leading '.' is never a name.
+SAFE_FILENAME = re.compile(r"[A-Za-z0-9_@-][A-Za-z0-9._@-]*")
+WINDOWS_DEVICE_NAMES = {"CON", "PRN", "AUX", "NUL", *(f"COM{i}" for i in range(1, 10)),
+                        *(f"LPT{i}" for i in range(1, 10))}
+DISPOSITION_PARAM = re.compile(r'\s*;\s*([^\s=;]+)\s*=\s*("(?:[^"\\]|\\.)*"|[^;]*)')
+
+
+def _disposition_params(value):
+    """Parameters of a Content-Disposition value, lower-cased names, quoted values unescaped."""
+    params = {}
+    pos = value.find(";")
+    while 0 <= pos < len(value):
+        match = DISPOSITION_PARAM.match(value, pos)
+        if not match:
+            break
+        raw = match.group(2).strip()
+        if raw.startswith('"') and raw.endswith('"') and len(raw) > 1:
+            raw = re.sub(r"\\(.)", r"\1", raw[1:-1])
+        params.setdefault(match.group(1).lower(), raw)
+        pos = match.end()
+    return params
+
+
+def _filename_from_content_disposition(value):
+    """The filename in a Content-Disposition value, preferring filename* (RFC 6266), or None."""
+    params = _disposition_params(value or "")
+    extended = params.get("filename*", "")
+    if extended.count("'") >= 2:
+        charset, _, encoded = extended.split("'", 2)
+        try:
+            return unquote(encoded, encoding=charset or "utf-8", errors="strict")
+        except (LookupError, UnicodeDecodeError):
+            pass
+    return params.get("filename")
+
+
+def _usable_filename(name):
+    if not name or not SAFE_FILENAME.fullmatch(name):
+        return False
+    return name.split(".")[0].upper() not in WINDOWS_DEVICE_NAMES
+
+
+def export_filename(response):
+    """The name the API gives the export, or None.
+
+    Tried in order: the Content-Disposition header, the signed URL's
+    response-content-disposition, then the URL's last path segment. A name
+    must be a plain file name made of the characters the API uses.
+    """
+    url = urlsplit(str(response.url))
+    candidates = (
+        lambda: _filename_from_content_disposition(response.headers.get("content-disposition")),
+        lambda: _filename_from_content_disposition(
+            parse_qs(url.query).get("response-content-disposition", [None])[0]),
+        lambda: unquote(url.path.rsplit("/", 1)[-1]),
+    )
+    for candidate in candidates:
+        try:
+            name = candidate()
+        except ValueError:
+            continue
+        if _usable_filename(name):
+            return name
+    return None
+
+
 @export.command()
 @export_args
 @click.option(
     "-o", "--output", "output_path",
     type=click.Path(),
-    help="Output file path (required).",
-    required=True,
+    help="Output file, or an existing directory (default: the API's name for the export, in the current directory).",
 )
 @click.pass_context
 def download(ctx, owner, repo, version, repo_type, owner_type, output_path):
-    """Download an export file to a local path."""
+    """Download an export file, named as the API names it unless -o gives a file."""
     client = ctx.obj["client"]
     try:
-        click.echo(f"Downloading export...", err=True)
+        click.echo("Downloading export...", err=True)
         response = client.export_download(
             owner, repo, version, owner_type=owner_type, repo_type=repo_type,
         )
 
-        with open(output_path, "wb") as f:
-            f.write(response.content)
+        # An explicit -o FILE is overwritten, as before. A name the API chose never replaces an existing file.
+        mode = "wb"
+        if not output_path or os.path.isdir(output_path):
+            filename = export_filename(response)
+            if not filename:
+                raise click.ClickException("The API didn't name the export. Pass -o FILE to choose a name.")
+            output_path = os.path.join(output_path or "", filename)
+            mode = "xb"
+        elif output_path.endswith(("/", os.sep)):
+            raise click.ClickException(f"Directory {output_path} doesn't exist.")
+
+        try:
+            with open(output_path, mode) as f:
+                f.write(response.content)
+        except FileExistsError:
+            raise click.ClickException(f"{output_path} already exists. Remove it, or pass -o FILE to overwrite.") from None
+        except OSError as e:
+            if mode != "xb" or e.errno != errno.ENAMETOOLONG:
+                raise
+            raise click.ClickException(
+                "The export's name is too long for this filesystem. Pass -o FILE to choose a name."
+            ) from None
 
         size = len(response.content)
         click.echo(f"Saved to {output_path} ({size:,} bytes)", err=True)

tests/test_export_commands.py

@@ -0,0 +1,192 @@
+import os
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import httpx
+from click.testing import CliRunner
+
+from ocl_cli.main import cli
+
+API_NAME = "orgs_PIH_sources_PIH_1.8.24_2026-09-30_123456.zip"
+SIGNED_URL = "https://exports.example.test/orgs/PIH/PIH_PIH_v1.8.24.2026-09-30_123456.zip?X-Amz-Signature=abc"
+
+
+def export_response(url=SIGNED_URL, headers=None):
+    return httpx.Response(200, headers=headers, content=b"zip-bytes", request=httpx.Request("GET", url))
+
+
+def make_runner():
+    # Click 8.1 mixes stderr into stdout unless told not to; 8.2 dropped mix_stderr.
+    try:
+        return CliRunner(mix_stderr=False)
+    except TypeError:
+        return CliRunner()
+
+
+class FakeConfig:
+    def get_server(self, server_id):
+        return SimpleNamespace(base_url="https://api.example.test")
+
+    def resolve_token(self, server, token_override=None):
+        return token_override
+
+
+class FakeExportClient:
+    def __init__(self, response):
+        self.response = response
+
+    def export_download(self, *args, **kwargs):
+        return self.response
+
+    def close(self):
+        pass
+
+
+class ExportDownloadCommandTest(unittest.TestCase):
+    def download(self, response, *args, exit_code=0):
+        runner = make_runner()
+        with (
+            runner.isolated_filesystem(),
+            patch("ocl_cli.main.CLIConfig.load", return_value=FakeConfig()),
+            patch("ocl_cli.main.OCLAPIClient", return_value=FakeExportClient(response)),
+        ):
+            os.mkdir("downloads")
+            result = runner.invoke(
+                cli, ["repo", "export", "download", "PIH", "PIH", "1.8.24", "--type", "source", *args]
+            )
+            self.assertEqual(result.exit_code, exit_code, result.output)
+            files = sorted(
+                os.path.join(root, name) for root, _, names in os.walk(".") for name in names
+            )
+        return files, result
+
+    def test_saves_under_the_content_disposition_name_by_default(self):
+        response = export_response(headers={"content-disposition": f'attachment; filename="{API_NAME}"'})
+
+        files, _ = self.download(response)
+
+        self.assertEqual(files, [f"./{API_NAME}"])
+
+    def test_reads_the_name_from_the_signed_url_without_a_header(self):
+        url = f"{SIGNED_URL}&response-content-disposition=attachment%3B%20filename%3D%22{API_NAME}%22"
+
+        files, _ = self.download(export_response(url=url))
+
+        self.assertEqual(files, [f"./{API_NAME}"])
+
+    def test_falls_back_to_the_storage_name(self):
+        files, _ = self.download(export_response())
+
+        self.assertEqual(files, ["./PIH_PIH_v1.8.24.2026-09-30_123456.zip"])
+
+    def test_output_directory_keeps_the_api_name(self):
+        response = export_response(headers={"content-disposition": f'attachment; filename="{API_NAME}"'})
+
+        files, _ = self.download(response, "-o", "downloads")
+
+        self.assertEqual(files, [f"./downloads/{API_NAME}"])
+
+    def test_output_file_overrides_the_api_name(self):
+        response = export_response(headers={"content-disposition": f'attachment; filename="{API_NAME}"'})
+
+        files, _ = self.download(response, "-o", "mine.zip")
+
+        self.assertEqual(files, ["./mine.zip"])
+
+    def test_a_path_in_the_name_is_rejected(self):
+        response = export_response(headers={"content-disposition": 'attachment; filename="../../evil.zip"'})
+
+        files, _ = self.download(response, "-o", "downloads")
+
+        self.assertEqual(files, ["./downloads/PIH_PIH_v1.8.24.2026-09-30_123456.zip"])
+
+    def test_filename_star_wins_over_filename(self):
+        disposition = f"attachment; filename=\"fallback.zip\"; filename*=UTF-8''{API_NAME}"
+
+        files, _ = self.download(export_response(headers={"content-disposition": disposition}))
+
+        self.assertEqual(files, [f"./{API_NAME}"])
+
+    def test_unusable_header_name_falls_back_to_the_signed_url(self):
+        url = f"{SIGNED_URL}&response-content-disposition=attachment%3B%20filename%3D%22{API_NAME}%22"
+        for disposition in ("attachment; filename*=UTF-8''evil%00.zip", 'attachment; filename="NUL.zip"',
+                            'attachment; filename="victim.txt:stream.zip"', "attachment; filename*=UTF-8''%FF.zip"):
+            with self.subTest(disposition=disposition):
+                files, _ = self.download(export_response(url=url, headers={"content-disposition": disposition}))
+
+                self.assertEqual(files, [f"./{API_NAME}"])
+
+    def test_quoted_parameters_can_contain_semicolons(self):
+        disposition = 'attachment; note="x; filename=wrong.zip; y"; filename="right.zip"'
+
+        files, _ = self.download(export_response(headers={"content-disposition": disposition}))
+
+        self.assertEqual(files, ["./right.zip"])
+
+    def test_storage_name_may_start_with_an_underscore(self):
+        url = "https://exports.example.test/orgs/_PIH/_PIH_PIH_v1.8.24.zip?X-Amz-Signature=abc"
+
+        files, _ = self.download(export_response(url=url))
+
+        self.assertEqual(files, ["./_PIH_PIH_v1.8.24.zip"])
+
+    def test_an_existing_file_is_not_overwritten(self):
+        response = export_response(headers={"content-disposition": f'attachment; filename="{API_NAME}"'})
+        runner = make_runner()
+        with (
+            runner.isolated_filesystem(),
+            patch("ocl_cli.main.CLIConfig.load", return_value=FakeConfig()),
+            patch("ocl_cli.main.OCLAPIClient", return_value=FakeExportClient(response)),
+        ):
+            with open("important.txt", "w") as f:
+                f.write("keep")
+            os.symlink("important.txt", API_NAME)
+            result = runner.invoke(cli, ["repo", "export", "download", "PIH", "PIH", "1.8.24", "--type", "source"])
+
+            self.assertEqual(result.exit_code, 1, result.output)
+            self.assertIn("already exists", result.stderr)
+            with open("important.txt") as f:
+                self.assertEqual(f.read(), "keep")
+
+    def test_a_name_too_long_for_the_filesystem_asks_for_output(self):
+        long_name = "orgs_PIH_sources_" + "x" * 300 + ".zip"
+        response = export_response(headers={"content-disposition": f'attachment; filename="{long_name}"'})
+
+        files, result = self.download(response, exit_code=1)
+
+        self.assertEqual(files, [])
+        self.assertIn("too long", result.stderr)
+
+    def test_a_missing_output_directory_is_an_error(self):
+        files, result = self.download(export_response(), "-o", "missing/", exit_code=1)
+
+        self.assertEqual(files, [])
+        self.assertIn("doesn't exist", result.stderr)
+
+    def test_unnamed_export_asks_for_output(self):
+        response = export_response(url="https://api.example.test/orgs/PIH/sources/PIH/1.8.24/export/")
+
+        files, result = self.download(response, exit_code=1)
+
+        self.assertEqual(files, [])
+        self.assertIn("Pass -o FILE", result.stderr)
+
+
+def test_export_download_follows_the_redirect_to_the_named_file(httpx_mock):
+    from ocl_cli.api_client import OCLAPIClient
+    from ocl_cli.commands.export import export_filename
+
+    httpx_mock.add_response(
+        method="GET", url="https://api.example.test/orgs/PIH/sources/PIH/1.8.24/export/",
+        status_code=302, headers={"location": SIGNED_URL},
+    )
+    httpx_mock.add_response(
+        method="GET", url=SIGNED_URL, content=b"zip-bytes",
+        headers={"content-disposition": f'attachment; filename="{API_NAME}"'},
+    )
+
+    response = OCLAPIClient(base_url="https://api.example.test", token="t").export_download("PIH", "PIH", "1.8.24")
+
+    assert response.content == b"zip-bytes"
+    assert export_filename(response) == API_NAME