fix(env): export dest 末尾 / のファイル名自動補完 (#24)

lib/devbase/env/io_export.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import getpass # noqa: F401 (tests monkey-patch devbase.env.io_export.getpass)
+import os
 import re
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -44,11 +45,42 @@ class ExportOptions:
  unsafe_allow_unencrypted_bucket: bool = False
 
 
-def _default_dest(force_unencrypted: bool) -> str:
+def _generate_default_filename(force_unencrypted: bool) -> str:
+ """既定ファイル名 (prefix なし) を生成する共通ヘルパー。
+ `_default_dest` / `_default_filename` の両方から呼ぶことで結合度を下げる
+ (PR #25 cross-review round 1 gemini 指摘)。"""
  # microsecond まで含めて衝突を回避する (PR #22 codex round 3 指摘)
  ts = datetime.now().strftime('%Y%m%d-%H%M%S-%f')
  suffix = '.dbenv.tar.gz' if force_unencrypted else '.dbenv'
- return f'./devbase-env-{ts}{suffix}'
+ return f'devbase-env-{ts}{suffix}'
+
+
+def _default_dest(force_unencrypted: bool) -> str:
+ return f'./{_generate_default_filename(force_unencrypted)}'
+
+
+def _default_filename(force_unencrypted: bool) -> str:
+ """既定ファイル名 (prefix なし) を返す。dest がディレクトリ的なときに append する用途。"""
+ return _generate_default_filename(force_unencrypted)
+
+
+def _complete_dir_dest(dest: str, force_unencrypted: bool) -> str:
+ """dest が「ディレクトリ的」なら既定ファイル名を補完する (`aws s3 cp` 互換、#24)。
+
+ - S3 URI で末尾が `/`: `s3://bucket/prefix/` → `s3://bucket/prefix/<default>`
+ - ローカルで既存ディレクトリ: `/tmp/out/` (または末尾 `/` なし) → `/tmp/out/<default>`
+ - それ以外 (フルキー / 通常ファイルパス / stdio `-`) はそのまま返す。
+ """
+ if _storage.is_stdio(dest):
+ return dest
+ name = _default_filename(force_unencrypted)
+ if _storage.is_s3(dest):
+ return dest + name if dest.endswith('/') else dest
+ # ローカル: 既存ディレクトリか末尾 `/` ならディレクトリ扱い
+ p = Path(dest)
+ if dest.endswith(('/', os.sep)) or p.is_dir():
+ return str(p / name)
+ return dest
 
 
 def _read_passphrase(opts: ExportOptions) -> Optional[str]:
@@ -163,6 +195,14 @@ def export(devbase_root: Path, opts: ExportOptions) -> int:
  logger.debug("暗号化後サイズ: %d bytes", len(payload))
 
  dest = opts.dest or _default_dest(opts.force_unencrypted)
+ # dest が「ディレクトリ的」なら `aws s3 cp` 互換でファイル名を自動補完する (#24)。
+ # 末尾 `/` の S3 URI で空キーオブジェクトが作られる事故と、ローカル既存
+ # ディレクトリへの OSError fail-fast の両方を救う。
+ if opts.dest:
+ completed = _complete_dir_dest(dest, opts.force_unencrypted)
+ if completed != dest:
+ logger.info("dest がディレクトリ的なためファイル名を補完: %s", completed)
+ dest = completed
  # 既定名 (opts.dest 未指定) かつローカルパスの場合、既存ファイルの上書きを拒否する
  # (microsecond 精度でも理論上は衝突しうるため防御的にチェック)
  if not opts.dest and not _storage.is_s3(dest) and not _storage.is_stdio(dest):

tests/cli/test_env_export.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import io
+import os
 from pathlib import Path
 
 import pyrage
@@ -343,3 +344,101 @@ def test_export_empty_dest_rejects_existing_file(
  dest="", # 空文字 — None と同様に既定名ガードが効くこと
  recipients=[f"@{pub_file}"],
  ))
+
+
+# --- dest 末尾 `/` のファイル名自動補完 (#24, PLAN03-2) ---
+
+
+def test_complete_dir_dest_s3_trailing_slash(monkeypatch):
+ """S3 URI が末尾 `/` のときは既定ファイル名を append する (`aws s3 cp` 互換)"""
+ from devbase.env.io_export import _complete_dir_dest
+ monkeypatch.setattr(
+ "devbase.env.io_export._default_filename",
+ lambda fu: "devbase-env-FIXED.dbenv",
+ )
+ assert _complete_dir_dest("s3://bucket/prefix/", False) == \
+ "s3://bucket/prefix/devbase-env-FIXED.dbenv"
+
+
+def test_complete_dir_dest_s3_full_key_unchanged(monkeypatch):
+ """S3 フルキー (末尾 `/` なし) はそのまま返す (回帰防止)"""
+ from devbase.env.io_export import _complete_dir_dest
+ monkeypatch.setattr(
+ "devbase.env.io_export._default_filename",
+ lambda fu: "devbase-env-FIXED.dbenv",
+ )
+ assert _complete_dir_dest("s3://bucket/prefix/foo.dbenv", False) == \
+ "s3://bucket/prefix/foo.dbenv"
+
+
+def test_complete_dir_dest_local_existing_dir(tmp_path, monkeypatch):
+ """ローカル既存ディレクトリのときも補完する"""
+ from devbase.env.io_export import _complete_dir_dest
+ monkeypatch.setattr(
+ "devbase.env.io_export._default_filename",
+ lambda fu: "devbase-env-FIXED.dbenv",
+ )
+ d = tmp_path / "outdir"
+ d.mkdir()
+ result = _complete_dir_dest(str(d), False)
+ assert result == str(d / "devbase-env-FIXED.dbenv")
+
+
+def test_complete_dir_dest_local_trailing_slash(tmp_path, monkeypatch):
+ """末尾 `/` のローカルパスは (ディレクトリが存在しなくても) 補完する"""
+ from devbase.env.io_export import _complete_dir_dest
+ monkeypatch.setattr(
+ "devbase.env.io_export._default_filename",
+ lambda fu: "devbase-env-FIXED.dbenv",
+ )
+ target_dir = tmp_path / "nodir"
+ target_path_with_slash = str(target_dir) + os.sep
+ expected_path = str(target_dir / "devbase-env-FIXED.dbenv")
+ assert _complete_dir_dest(target_path_with_slash, False) == expected_path
+
+
+def test_complete_dir_dest_local_normal_file_unchanged(tmp_path, monkeypatch):
+ """通常のファイルパスは補完しない"""
+ from devbase.env.io_export import _complete_dir_dest
+ monkeypatch.setattr(
+ "devbase.env.io_export._default_filename",
+ lambda fu: "devbase-env-FIXED.dbenv",
+ )
+ target = str(tmp_path / "out.dbenv")
+ assert _complete_dir_dest(target, False) == target
+
+
+def test_complete_dir_dest_stdio_unchanged():
+ """stdio (`-`) は補完しない"""
+ from devbase.env.io_export import _complete_dir_dest
+ assert _complete_dir_dest("-", False) == "-"
+
+
+def test_complete_dir_dest_plaintext_suffix(monkeypatch):
+ """force_unencrypted=True のときは `.dbenv.tar.gz` で補完される"""
+ from devbase.env.io_export import _complete_dir_dest, _default_filename
+ name = _default_filename(True)
+ assert name.endswith(".dbenv.tar.gz")
+ result = _complete_dir_dest("s3://bucket/prefix/", True)
+ assert result.endswith(".dbenv.tar.gz")
+ assert result.startswith("s3://bucket/prefix/")
+
+
+def test_export_local_dir_completes_filename(fake_root, age_keys, tmp_path):
+ """end-to-end: ローカル既存ディレクトリへの export でファイル名が補完される"""
+ pub_file, id_file = age_keys
+ outdir = tmp_path / "outdir"
+ outdir.mkdir()
+ rc = export(fake_root, ExportOptions(
+ dest=str(outdir),
+ recipients=[f"@{pub_file}"],
+ ))
+ assert rc == 0
+ # 補完されたファイルが 1 つだけ生成されている
+ files = list(outdir.iterdir())
+ assert len(files) == 1
+ assert files[0].name.startswith("devbase-env-")
+ assert files[0].name.endswith(".dbenv")
+ # 内容を復号できる
+ decrypted = cipher.decrypt(files[0].read_bytes(), identities=[str(id_file)])
+ assert decrypted