From d0af5b52cbe08bb039fb85c6a7ba9c60ea89e16c Mon Sep 17 00:00:00 2001 From: Sahil Sunny Date: Thu, 9 Jul 2026 22:13:39 +0530 Subject: [PATCH 1/4] Validate and resolve file paths before API work MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary Fix path handling for --output-file, --output-dir, and --input-file so ~/ home shortcuts work, missing output directories are created, overwrite conflicts are detected early, and input files are verified before any scrape or batch API call. This prevents wasted credits from scraping first and failing on write, and gives clear REPL-friendly errors instead of literal '~/...' path failures. Details - cli_utils.py:323-325 — add resolve_output_path() to expand ~ via Path.expanduser() so all path helpers share one resolution step. - cli_utils.py:328-349 — add ensure_output_file_ready() to expand ~, mkdir parent dirs, and run overwrite checks before API work. - cli_utils.py:352-360 — add ensure_output_dir_ready() to expand ~ and create --output-dir before batch/crawl writes. - cli_utils.py:363-385 — add ensure_input_file_ready() to expand ~, verify the file exists/is readable, and pass stdin (-) through unchanged. - cli_utils.py:392-403 — update confirm_overwrite() to check the expanded path so ~/Desktop/foo.png correctly detects an existing Desktop file; REPL still raises UsageError with --overwrite hint instead of blocking on click.confirm(). - cli_utils.py:591-601 — call all ensure_* helpers at the end of store_common_options(), after flag validation but before commands hit the API. - cli_utils.py:1902-1923 — harden write_output() with expand/mkdir/open on the resolved path as a safety net when extensions are appended post-validation. - batch.py:220-221 — expand ~ in read_input_file() as a fallback for callers outside store_common_options(). - crawl.py:432,569 — ensure crawl output dirs (explicit --output-dir and default crawl_) exist before the spider starts. - export.py:96-101 — replace confirm_overwrite() with ensure_output_file_ready() so export validates the destination before reading input files. - test_v132_fixes.py:600-860 — unit tests for tilde expansion, early input/output validation, and REPL scrape guards that API must not run on bad paths. - test_repl_pty.py:320-345 — PTY regression that REPL shows overwrite error for existing ~/ output paths before scraping. - test_v132_fixes.py:955-1001 — update batch option tests to create real input files now that store_common_options validates --input-file up front. --- src/scrapingbee_cli/batch.py | 3 + src/scrapingbee_cli/cli_utils.py | 112 ++++++++-- src/scrapingbee_cli/commands/crawl.py | 7 +- src/scrapingbee_cli/commands/export.py | 8 +- tests/unit/test_repl_pty.py | 29 +++ tests/unit/test_v132_fixes.py | 288 ++++++++++++++++++++++++- 6 files changed, 425 insertions(+), 22 deletions(-) diff --git a/src/scrapingbee_cli/batch.py b/src/scrapingbee_cli/batch.py index 99b2b48..0f38fb7 100644 --- a/src/scrapingbee_cli/batch.py +++ b/src/scrapingbee_cli/batch.py @@ -217,6 +217,9 @@ def read_input_file(path: str, *, input_column: str | None = None) -> list[str]: If path ends with .csv, read as CSV using input_column (name or 0-based index).""" import sys as _sys + if path != "-": + path = str(Path(path).expanduser()) + if path == "-": lines = [line.strip() for line in _sys.stdin if line.strip()] elif path.lower().endswith(".csv"): diff --git a/src/scrapingbee_cli/cli_utils.py b/src/scrapingbee_cli/cli_utils.py index 4e5b48b..76131e7 100644 --- a/src/scrapingbee_cli/cli_utils.py +++ b/src/scrapingbee_cli/cli_utils.py @@ -6,6 +6,7 @@ import json import re import sys +from pathlib import Path from typing import Any import click @@ -243,7 +244,13 @@ def _batch_options(f: Any) -> Any: default=None, help="CSV input: column name or 0-based index.", )(f) - f = click.option("--output-dir", "output_dir", default=None, help="Batch output folder.")(f) + f = click.option( + "--output-dir", + "output_dir", + type=click.Path(), + default=None, + help="Batch output folder.", + )(f) f = click.option( "--output-format", "output_format", @@ -313,21 +320,86 @@ def _batch_options(f: Any) -> Any: return f +def resolve_output_path(path: str) -> str: + """Expand ``~`` in an output path.""" + return str(Path(path).expanduser()) + + +def ensure_output_file_ready( + path: str, + *, + overwrite: bool = False, + skip_overwrite_check: bool = False, +) -> str: + """Validate an output file path before any API work starts. + + Expands ``~``, creates parent directories, and checks overwrite policy. + Returns the resolved path. + """ + resolved = resolve_output_path(path) + parent = Path(resolved).parent + if str(parent) and not parent.exists(): + try: + parent.mkdir(parents=True, exist_ok=True) + except OSError as e: + click.echo(f"Cannot create directory '{parent}': {e.strerror}", err=True) + raise SystemExit(1) + if not skip_overwrite_check: + confirm_overwrite(resolved, overwrite) + return resolved + + +def ensure_output_dir_ready(path: str) -> str: + """Expand ``~`` and create an output directory before API work.""" + resolved = resolve_output_path(path) + try: + Path(resolved).mkdir(parents=True, exist_ok=True) + except OSError as e: + click.echo(f"Cannot create directory '{resolved}': {e.strerror}", err=True) + raise SystemExit(1) + return resolved + + +def ensure_input_file_ready(path: str) -> str: + """Validate an input file path before any API work starts. + + Expands ``~`` and ensures the file exists and is readable. Stdin (``-``) is + passed through unchanged. + """ + if path == "-": + return path + resolved = resolve_output_path(path) + p = Path(resolved) + if not p.is_file(): + if p.exists(): + click.echo(f"Not a file: '{resolved}'", err=True) + else: + click.echo(f"Input file not found: '{resolved}'", err=True) + raise SystemExit(1) + try: + with open(resolved, "rb"): + pass + except OSError as e: + click.echo(f"Cannot read from '{resolved}': {e.strerror}", err=True) + raise SystemExit(1) + return resolved + + def confirm_overwrite(path: str | None, overwrite: bool = False) -> None: """If path exists, prompt for confirmation unless --overwrite is set.""" if not path: return - from pathlib import Path + resolved = resolve_output_path(path) - if Path(path).exists() and not overwrite: + if Path(resolved).exists() and not overwrite: # In REPL mode, prompt_toolkit owns the TTY (full-screen / alt-buffer), # so click.confirm reads from sys.stdin and blocks forever. Surface # the conflict as an error and tell the user to pass --overwrite. if is_repl_mode(): raise click.UsageError( - f"'{path}' already exists. Re-run with --overwrite to replace it." + f"'{resolved}' already exists. Re-run with --overwrite to replace it." ) - if not click.confirm(f"'{path}' already exists. Overwrite?"): + if not click.confirm(f"'{resolved}' already exists. Overwrite?"): click.echo("Cancelled.", err=True) raise SystemExit(0) @@ -380,10 +452,6 @@ def store_common_options(obj: dict, **kwargs: Any) -> None: has_output_file = bool(obj.get("output_file")) has_output_dir = bool(obj.get("output_dir")) - # Check if output file already exists (skip for --update-csv which intentionally overwrites) - if has_output_file and not obj.get("update_csv"): - confirm_overwrite(obj["output_file"], obj.get("overwrite", False)) - # Mutual exclusion: --output-file and --output-dir if has_output_file and has_output_dir: click.echo( @@ -520,6 +588,18 @@ def store_common_options(obj: dict, **kwargs: Any) -> None: ) raise SystemExit(1) + # Resolve paths before any API work (expand ~, mkdir outputs, validate inputs). + if obj.get("input_file"): + obj["input_file"] = ensure_input_file_ready(obj["input_file"]) + if has_output_file: + obj["output_file"] = ensure_output_file_ready( + obj["output_file"], + overwrite=bool(obj.get("overwrite", False)), + skip_overwrite_check=bool(obj.get("update_csv")), + ) + if has_output_dir: + obj["output_dir"] = ensure_output_dir_ready(obj["output_dir"]) + def _parse_path(path: str) -> list[tuple[str, Any]]: """Parse a path expression into typed segments. @@ -1819,10 +1899,18 @@ def write_output( elif fields: data = _filter_fields(data, fields) if output_path: + resolved = resolve_output_path(output_path) + parent = Path(resolved).parent + if str(parent) and not parent.exists(): + try: + parent.mkdir(parents=True, exist_ok=True) + except OSError as e: + click.echo(f"Cannot create directory '{parent}': {e.strerror}", err=True) + raise SystemExit(1) try: - fh = open(output_path, "wb") + fh = open(resolved, "wb") except OSError as e: - click.echo(f"Cannot write to '{output_path}': {e.strerror}", err=True) + click.echo(f"Cannot write to '{resolved}': {e.strerror}", err=True) raise SystemExit(1) with fh: fh.write(data) @@ -1832,7 +1920,7 @@ def write_output( # stdout/pipes clean for non-REPL use. from .theme import BEE_DIM, BEE_YELLOW, err_console - err_console.print(f" [{BEE_DIM}]Saved to[/] [bold {BEE_YELLOW}]{output_path}[/]") + err_console.print(f" [{BEE_DIM}]Saved to[/] [bold {BEE_YELLOW}]{resolved}[/]") else: # In REPL mode, truncate large text dumps to a tidy preview and surface # a path to the full output. Non-REPL invocations (`scrapingbee scrape ...`) diff --git a/src/scrapingbee_cli/commands/crawl.py b/src/scrapingbee_cli/commands/crawl.py index 9d599ed..1b1313e 100644 --- a/src/scrapingbee_cli/commands/crawl.py +++ b/src/scrapingbee_cli/commands/crawl.py @@ -14,6 +14,7 @@ _validate_json_option, _validate_range, build_scrape_kwargs, + ensure_output_dir_ready, scrape_kwargs_to_api_params, store_common_options, ) @@ -427,7 +428,10 @@ def crawl_cmd( See https://www.scrapingbee.com/documentation/ for parameter details. """ store_common_options(obj, **kwargs) - obj["output_dir"] = output_dir or "" + if output_dir: + obj["output_dir"] = ensure_output_dir_ready(output_dir) + else: + obj["output_dir"] = "" obj["concurrency"] = concurrency or 0 obj["resume"] = resume obj["on_complete"] = on_complete @@ -562,6 +566,7 @@ def crawl_cmd( custom_headers[k.strip()] = v.strip() out_dir = (obj.get("output_dir") or "").strip() or None out_dir = out_dir or default_crawl_output_dir() + out_dir = ensure_output_dir_ready(out_dir) allowed_list: list[str] | None = None if allowed_domains: allowed_list = [d.strip() for d in allowed_domains.split(",") if d.strip()] diff --git a/src/scrapingbee_cli/commands/export.py b/src/scrapingbee_cli/commands/export.py index 70bb05c..5f68dec 100644 --- a/src/scrapingbee_cli/commands/export.py +++ b/src/scrapingbee_cli/commands/export.py @@ -93,10 +93,12 @@ def export_cmd( input_path = Path(input_dir).resolve() output_file = obj.get("output_file") - # Check if output file already exists - from ..cli_utils import confirm_overwrite + # Check if output file already exists and prepare destination before reading inputs. + from ..cli_utils import ensure_output_file_ready - confirm_overwrite(output_file, overwrite) + if output_file: + output_file = ensure_output_file_ready(output_file, overwrite=overwrite) + obj["output_file"] = output_file # Load manifest for URL → relative-path mapping (optional) # Supports both old format (string values) and new format (dict values with "file" key). diff --git a/tests/unit/test_repl_pty.py b/tests/unit/test_repl_pty.py index 0176fe8..648ea62 100644 --- a/tests/unit/test_repl_pty.py +++ b/tests/unit/test_repl_pty.py @@ -314,3 +314,32 @@ def test_session_default_no_skip_warning_for_supported_command(tmp_path): ) finally: child.close(force=True) + + +@needs_cli +def test_repl_tilde_output_file_validated_before_scrape(tmp_path): + """REPL expands ~/ output paths and rejects overwrites before any scrape work.""" + home = tmp_path / "home" + home.mkdir() + out_dir = home / "Desktop" / "sb-cli-test" / "ss" + out_dir.mkdir(parents=True) + (out_dir / "screenshot.png").write_bytes(b"old") + + child, screen, stream = _spawn(home) + try: + assert _pump_until(child, screen, stream, lambda s: "❯" in _text(s)), "no prompt" + child.send( + "scrape https://example.com --render-js false " + "--output-file ~/Desktop/sb-cli-test/ss/screenshot.png\r" + ) + assert _pump_until( + child, + screen, + stream, + lambda s: "already exists" in _text(s) and "--overwrite" in _text(s), + timeout=10.0, + ), "overwrite error not shown before scrape" + t = _text(screen) + assert "Cannot write to '~/" not in t + finally: + child.close(force=True) diff --git a/tests/unit/test_v132_fixes.py b/tests/unit/test_v132_fixes.py index 1bed75b..44545d9 100644 --- a/tests/unit/test_v132_fixes.py +++ b/tests/unit/test_v132_fixes.py @@ -592,6 +592,273 @@ def test_continues_when_user_confirms(self, tmp_path, monkeypatch): confirm_overwrite(str(path), overwrite=False) # should not raise +# ============================================================================= +# 8b. Output path resolution +# ============================================================================= + + +class TestOutputPathResolution: + """Tests for resolve_output_path / ensure_output_file_ready.""" + + def test_resolve_output_path_expands_tilde(self, monkeypatch): + from scrapingbee_cli.cli_utils import resolve_output_path + + monkeypatch.setenv("HOME", "/tmp/fakehome") + assert resolve_output_path("~/out.png") == "/tmp/fakehome/out.png" + + def test_ensure_output_file_ready_creates_parent_dirs(self, tmp_path): + from scrapingbee_cli.cli_utils import ensure_output_file_ready + + out = tmp_path / "nested" / "dir" / "shot.png" + resolved = ensure_output_file_ready(str(out)) + assert resolved == str(out) + assert out.parent.is_dir() + + def test_ensure_output_file_ready_checks_overwrite_before_return(self, tmp_path, monkeypatch): + from scrapingbee_cli.cli_utils import ensure_output_file_ready + + existing = tmp_path / "exists.png" + existing.write_bytes(b"old") + monkeypatch.setattr("click.confirm", lambda *a, **kw: False) + with pytest.raises(SystemExit): + ensure_output_file_ready(str(existing), overwrite=False) + + def test_store_common_options_prepares_tilde_output_file(self, tmp_path, monkeypatch): + from scrapingbee_cli.cli_utils import store_common_options + + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + out = "~/Desktop/sb-test/screenshot.png" + obj: dict = {} + store_common_options(obj, **TestStoreCommonOptionsBatchValidation()._make_obj(output_file=out)) + expected = str(home / "Desktop/sb-test/screenshot.png") + assert obj["output_file"] == expected + assert Path(expected).parent.is_dir() + + def test_store_common_options_prepares_output_dir(self, tmp_path, monkeypatch): + from scrapingbee_cli.cli_utils import store_common_options + + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + input_file = tmp_path / "urls.txt" + input_file.write_text("https://example.com\n", encoding="utf-8") + obj: dict = {} + store_common_options( + obj, + **TestStoreCommonOptionsBatchValidation()._make_obj( + input_file=str(input_file), + output_dir="~/batch-results", + ), + ) + expected = str(home / "batch-results") + assert obj["output_dir"] == expected + assert Path(expected).is_dir() + + +class TestInputPathResolution: + """Tests for ensure_input_file_ready / early --input-file validation.""" + + def test_ensure_input_file_ready_expands_tilde(self, tmp_path, monkeypatch): + from scrapingbee_cli.cli_utils import ensure_input_file_ready + + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + input_file = home / "urls.txt" + input_file.write_text("https://example.com\n", encoding="utf-8") + assert ensure_input_file_ready("~/urls.txt") == str(input_file) + + def test_ensure_input_file_ready_passes_stdin(self): + from scrapingbee_cli.cli_utils import ensure_input_file_ready + + assert ensure_input_file_ready("-") == "-" + + def test_ensure_input_file_ready_missing_file_exits(self, tmp_path, monkeypatch): + from scrapingbee_cli.cli_utils import ensure_input_file_ready + + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + with pytest.raises(SystemExit): + ensure_input_file_ready("~/missing.txt") + + def test_store_common_options_prepares_tilde_input_file(self, tmp_path, monkeypatch): + from scrapingbee_cli.cli_utils import store_common_options + + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + input_file = home / "urls.txt" + input_file.write_text("https://example.com\n", encoding="utf-8") + obj: dict = {} + store_common_options( + obj, + **TestStoreCommonOptionsBatchValidation()._make_obj(input_file="~/urls.txt"), + ) + assert obj["input_file"] == str(input_file) + + def test_read_input_file_expands_tilde(self, tmp_path, monkeypatch): + from scrapingbee_cli.batch import read_input_file + + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + input_file = home / "urls.txt" + input_file.write_text("https://example.com\n", encoding="utf-8") + assert read_input_file("~/urls.txt") == ["https://example.com"] + + +class TestReplOutputPathResolution: + """REPL-mode regressions for ~/ output paths (expand, mkdir, pre-scrape checks).""" + + def test_confirm_overwrite_repl_mode_uses_expanded_tilde_path(self, tmp_path, monkeypatch): + import click + + from scrapingbee_cli.cli_utils import confirm_overwrite + from scrapingbee_cli.theme import set_repl_mode + + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + existing = home / "Desktop" / "shot.png" + existing.parent.mkdir(parents=True) + existing.write_bytes(b"old") + + set_repl_mode(True) + try: + with pytest.raises(click.UsageError, match="already exists"): + confirm_overwrite("~/Desktop/shot.png", overwrite=False) + finally: + set_repl_mode(False) + + def test_scrape_repl_rejects_existing_tilde_output_before_api(self, tmp_path, monkeypatch): + from click.testing import CliRunner + + from scrapingbee_cli.commands.scrape import scrape_cmd + from scrapingbee_cli.theme import set_repl_mode + + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + out_dir = home / "Desktop" / "sb-cli-test" / "ss" + out_dir.mkdir(parents=True) + (out_dir / "screenshot.png").write_bytes(b"old") + + api_called = {"value": False} + + def _fail_if_api_runs(*_a, **_kw): + api_called["value"] = True + raise AssertionError("scrape API must not run when output path is invalid") + + monkeypatch.setattr("scrapingbee_cli.commands.scrape.get_api_key", _fail_if_api_runs) + + set_repl_mode(True) + try: + result = CliRunner().invoke( + scrape_cmd, + [ + "https://example.com", + "--render-js", + "false", + "--output-file", + "~/Desktop/sb-cli-test/ss/screenshot.png", + ], + obj={}, + ) + finally: + set_repl_mode(False) + + combined = (result.output or "") + (result.stderr or "") + assert api_called["value"] is False + assert result.exit_code != 0 + assert "already exists" in combined + assert "--overwrite" in combined + assert "Cannot write to '~/" not in combined + + def test_scrape_repl_rejects_missing_tilde_input_before_api(self, tmp_path, monkeypatch): + from click.testing import CliRunner + + from scrapingbee_cli.commands.scrape import scrape_cmd + from scrapingbee_cli.theme import set_repl_mode + + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + + api_called = {"value": False} + + def _fail_if_api_runs(*_a, **_kw): + api_called["value"] = True + raise AssertionError("scrape API must not run when input file is missing") + + monkeypatch.setattr("scrapingbee_cli.commands.scrape.get_api_key", _fail_if_api_runs) + + set_repl_mode(True) + try: + result = CliRunner().invoke( + scrape_cmd, + [ + "--input-file", + "~/urls.txt", + "--output-dir", + str(tmp_path / "out"), + ], + obj={}, + ) + finally: + set_repl_mode(False) + + combined = (result.output or "") + (result.stderr or "") + assert api_called["value"] is False + assert result.exit_code != 0 + assert "Input file not found" in combined + assert "~/urls.txt" not in combined or "home/urls.txt" in combined + + def test_scrape_repl_expands_tilde_output_path_on_success(self, tmp_path, monkeypatch): + from unittest.mock import AsyncMock, patch + + from click.testing import CliRunner + + from scrapingbee_cli.commands.scrape import scrape_cmd + from scrapingbee_cli.theme import set_repl_mode + + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + out_file = home / "Desktop" / "sb-cli-test" / "ss" / "screenshot.png" + + mock_client = AsyncMock() + mock_client.scrape.return_value = (b"png-bytes", {}, 200) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + + set_repl_mode(True) + try: + with patch("scrapingbee_cli.commands.scrape.get_api_key", return_value="fake"): + with patch("scrapingbee_cli.commands.scrape.Client", return_value=mock_client): + result = CliRunner().invoke( + scrape_cmd, + [ + "https://example.com", + "--render-js", + "false", + "--output-file", + "~/Desktop/sb-cli-test/ss/screenshot.png", + ], + obj={}, + ) + finally: + set_repl_mode(False) + + combined = (result.output or "") + (result.stderr or "") + assert result.exit_code == 0, combined + assert out_file.parent.is_dir() + assert out_file.read_bytes() == b"png-bytes" + assert "screenshot.png" in combined.replace("\n", "") + + # ============================================================================= # 9. store_common_options() — batch-only flags without --input-file # ============================================================================= @@ -688,16 +955,23 @@ def test_resume_without_input_file_shows_discovery_hint(self, capsys): # Should show bare scrapingbee --resume hint for discovery assert "scrapingbee --resume" in err - def test_negative_concurrency_exits(self): + def test_negative_concurrency_exits(self, tmp_path): from scrapingbee_cli.cli_utils import store_common_options + input_file = tmp_path / "urls.txt" + input_file.write_text("https://example.com\n", encoding="utf-8") obj = {} with pytest.raises(SystemExit): - store_common_options(obj, **self._make_obj(concurrency=-1, input_file="urls.txt")) + store_common_options( + obj, + **self._make_obj(concurrency=-1, input_file=str(input_file)), + ) - def test_output_file_and_output_dir_mutual_exclusion(self): + def test_output_file_and_output_dir_mutual_exclusion(self, tmp_path): from scrapingbee_cli.cli_utils import store_common_options + input_file = tmp_path / "urls.txt" + input_file.write_text("https://example.com\n", encoding="utf-8") obj = {} with pytest.raises(SystemExit): store_common_options( @@ -705,7 +979,7 @@ def test_output_file_and_output_dir_mutual_exclusion(self): **self._make_obj( output_file="/tmp/out.json", output_dir="/tmp/out/", - input_file="urls.txt", + input_file=str(input_file), ), ) @@ -715,14 +989,16 @@ def test_valid_single_url_options_pass(self): obj = {} store_common_options(obj, **self._make_obj()) # should not raise - def test_valid_batch_options_pass(self): + def test_valid_batch_options_pass(self, tmp_path): from scrapingbee_cli.cli_utils import store_common_options + input_file = tmp_path / "urls.txt" + input_file.write_text("https://example.com\n", encoding="utf-8") obj = {} store_common_options( obj, **self._make_obj( - input_file="urls.txt", + input_file=str(input_file), output_dir="/tmp/out", concurrency=5, deduplicate=True, From 4401a904e3fb8458f9af6f81b0f9af04399ed4fe Mon Sep 17 00:00:00 2001 From: Sahil Sunny Date: Thu, 9 Jul 2026 22:25:00 +0530 Subject: [PATCH 2/4] Fix ruff format and ty type errors in path validation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary CI lint failed on ruff format (test_v132_fixes.py) and ty reported invalid-argument-type for ensure_* calls passing Any | None from obj dict. Narrow path values with isinstance(str) before calling helpers. Details - cli_utils.py:592-603 — use isinstance(path, str) guards before ensure_input_file_ready / ensure_output_file_ready / ensure_output_dir_ready so ty accepts narrowed str arguments from the options dict. - test_v132_fixes.py:634 — ruff format wrap for store_common_options call. --- src/scrapingbee_cli/cli_utils.py | 15 +++++++++------ tests/unit/test_v132_fixes.py | 4 +++- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/src/scrapingbee_cli/cli_utils.py b/src/scrapingbee_cli/cli_utils.py index 76131e7..8db41ad 100644 --- a/src/scrapingbee_cli/cli_utils.py +++ b/src/scrapingbee_cli/cli_utils.py @@ -589,16 +589,19 @@ def store_common_options(obj: dict, **kwargs: Any) -> None: raise SystemExit(1) # Resolve paths before any API work (expand ~, mkdir outputs, validate inputs). - if obj.get("input_file"): - obj["input_file"] = ensure_input_file_ready(obj["input_file"]) - if has_output_file: + input_file_path = obj.get("input_file") + if isinstance(input_file_path, str): + obj["input_file"] = ensure_input_file_ready(input_file_path) + output_file_path = obj.get("output_file") + if isinstance(output_file_path, str): obj["output_file"] = ensure_output_file_ready( - obj["output_file"], + output_file_path, overwrite=bool(obj.get("overwrite", False)), skip_overwrite_check=bool(obj.get("update_csv")), ) - if has_output_dir: - obj["output_dir"] = ensure_output_dir_ready(obj["output_dir"]) + output_dir_path = obj.get("output_dir") + if isinstance(output_dir_path, str) and output_dir_path: + obj["output_dir"] = ensure_output_dir_ready(output_dir_path) def _parse_path(path: str) -> list[tuple[str, Any]]: diff --git a/tests/unit/test_v132_fixes.py b/tests/unit/test_v132_fixes.py index 44545d9..8b1b786 100644 --- a/tests/unit/test_v132_fixes.py +++ b/tests/unit/test_v132_fixes.py @@ -631,7 +631,9 @@ def test_store_common_options_prepares_tilde_output_file(self, tmp_path, monkeyp monkeypatch.setenv("HOME", str(home)) out = "~/Desktop/sb-test/screenshot.png" obj: dict = {} - store_common_options(obj, **TestStoreCommonOptionsBatchValidation()._make_obj(output_file=out)) + store_common_options( + obj, **TestStoreCommonOptionsBatchValidation()._make_obj(output_file=out) + ) expected = str(home / "Desktop/sb-test/screenshot.png") assert obj["output_file"] == expected assert Path(expected).parent.is_dir() From f7c503ccf6af28a0dce1fc03a5b32400426c5ec4 Mon Sep 17 00:00:00 2001 From: Sahil Sunny Date: Thu, 9 Jul 2026 22:34:28 +0530 Subject: [PATCH 3/4] Fix path tests on Windows by setting USERPROFILE MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary CI failed on all Windows matrix jobs: the new ~/ path tests set only HOME, but Path.expanduser() on Windows reads USERPROFILE, so ~ expanded to the real runner profile (C:\Users\runneradmin) instead of the pytest sandbox. Add a _set_home() helper that sets both variables and use it everywhere. Details - test_v132_fixes.py:28-37 — add _set_home() helper setting HOME and USERPROFILE so ~ expansion is sandboxed on POSIX and Windows alike. - test_v132_fixes.py:615-618 — test_resolve_output_path_expands_tilde now uses tmp_path instead of a hardcoded POSIX /tmp/fakehome expectation. - test_v132_fixes.py (10 call sites) — replace monkeypatch.setenv("HOME", ...) with _set_home(monkeypatch, home) in output/input/REPL path tests. Note: the macOS/Ubuntu failures on test_session_default_skip_warning_on_screen are a pre-existing flaky PTY test — it also failed on the main branch push (run 29008646319) before this branch existed. --- tests/unit/test_v132_fixes.py | 38 +++++++++++++++++++++++------------ 1 file changed, 25 insertions(+), 13 deletions(-) diff --git a/tests/unit/test_v132_fixes.py b/tests/unit/test_v132_fixes.py index 8b1b786..b3ce005 100644 --- a/tests/unit/test_v132_fixes.py +++ b/tests/unit/test_v132_fixes.py @@ -24,6 +24,18 @@ import click import pytest + +def _set_home(monkeypatch, path) -> None: + """Point the home directory at *path* on all platforms. + + ``Path.expanduser()`` reads ``HOME`` on POSIX but ``USERPROFILE`` on + Windows (``HOME`` is ignored there since Python 3.8), so tests must set + both for ``~`` expansion to land in the sandbox. + """ + monkeypatch.setenv("HOME", str(path)) + monkeypatch.setenv("USERPROFILE", str(path)) + + # ============================================================================= # 1. user_agent_headers() # ============================================================================= @@ -600,11 +612,11 @@ def test_continues_when_user_confirms(self, tmp_path, monkeypatch): class TestOutputPathResolution: """Tests for resolve_output_path / ensure_output_file_ready.""" - def test_resolve_output_path_expands_tilde(self, monkeypatch): + def test_resolve_output_path_expands_tilde(self, tmp_path, monkeypatch): from scrapingbee_cli.cli_utils import resolve_output_path - monkeypatch.setenv("HOME", "/tmp/fakehome") - assert resolve_output_path("~/out.png") == "/tmp/fakehome/out.png" + _set_home(monkeypatch, tmp_path) + assert resolve_output_path("~/out.png") == str(tmp_path / "out.png") def test_ensure_output_file_ready_creates_parent_dirs(self, tmp_path): from scrapingbee_cli.cli_utils import ensure_output_file_ready @@ -628,7 +640,7 @@ def test_store_common_options_prepares_tilde_output_file(self, tmp_path, monkeyp home = tmp_path / "home" home.mkdir() - monkeypatch.setenv("HOME", str(home)) + _set_home(monkeypatch, home) out = "~/Desktop/sb-test/screenshot.png" obj: dict = {} store_common_options( @@ -643,7 +655,7 @@ def test_store_common_options_prepares_output_dir(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() - monkeypatch.setenv("HOME", str(home)) + _set_home(monkeypatch, home) input_file = tmp_path / "urls.txt" input_file.write_text("https://example.com\n", encoding="utf-8") obj: dict = {} @@ -667,7 +679,7 @@ def test_ensure_input_file_ready_expands_tilde(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() - monkeypatch.setenv("HOME", str(home)) + _set_home(monkeypatch, home) input_file = home / "urls.txt" input_file.write_text("https://example.com\n", encoding="utf-8") assert ensure_input_file_ready("~/urls.txt") == str(input_file) @@ -682,7 +694,7 @@ def test_ensure_input_file_ready_missing_file_exits(self, tmp_path, monkeypatch) home = tmp_path / "home" home.mkdir() - monkeypatch.setenv("HOME", str(home)) + _set_home(monkeypatch, home) with pytest.raises(SystemExit): ensure_input_file_ready("~/missing.txt") @@ -691,7 +703,7 @@ def test_store_common_options_prepares_tilde_input_file(self, tmp_path, monkeypa home = tmp_path / "home" home.mkdir() - monkeypatch.setenv("HOME", str(home)) + _set_home(monkeypatch, home) input_file = home / "urls.txt" input_file.write_text("https://example.com\n", encoding="utf-8") obj: dict = {} @@ -706,7 +718,7 @@ def test_read_input_file_expands_tilde(self, tmp_path, monkeypatch): home = tmp_path / "home" home.mkdir() - monkeypatch.setenv("HOME", str(home)) + _set_home(monkeypatch, home) input_file = home / "urls.txt" input_file.write_text("https://example.com\n", encoding="utf-8") assert read_input_file("~/urls.txt") == ["https://example.com"] @@ -723,7 +735,7 @@ def test_confirm_overwrite_repl_mode_uses_expanded_tilde_path(self, tmp_path, mo home = tmp_path / "home" home.mkdir() - monkeypatch.setenv("HOME", str(home)) + _set_home(monkeypatch, home) existing = home / "Desktop" / "shot.png" existing.parent.mkdir(parents=True) existing.write_bytes(b"old") @@ -743,7 +755,7 @@ def test_scrape_repl_rejects_existing_tilde_output_before_api(self, tmp_path, mo home = tmp_path / "home" home.mkdir() - monkeypatch.setenv("HOME", str(home)) + _set_home(monkeypatch, home) out_dir = home / "Desktop" / "sb-cli-test" / "ss" out_dir.mkdir(parents=True) (out_dir / "screenshot.png").write_bytes(b"old") @@ -787,7 +799,7 @@ def test_scrape_repl_rejects_missing_tilde_input_before_api(self, tmp_path, monk home = tmp_path / "home" home.mkdir() - monkeypatch.setenv("HOME", str(home)) + _set_home(monkeypatch, home) api_called = {"value": False} @@ -828,7 +840,7 @@ def test_scrape_repl_expands_tilde_output_path_on_success(self, tmp_path, monkey home = tmp_path / "home" home.mkdir() - monkeypatch.setenv("HOME", str(home)) + _set_home(monkeypatch, home) out_file = home / "Desktop" / "sb-cli-test" / "ss" / "screenshot.png" mock_client = AsyncMock() From d62247038809db052ca8f54750ed3f0d293f5789 Mon Sep 17 00:00:00 2001 From: Sahil Sunny Date: Thu, 16 Jul 2026 18:22:44 +0530 Subject: [PATCH 4/4] Fix safe tilde expansion and final overwrite checks Restrict home expansion to current-user paths so literal tilde names cannot crash or target another user's home. Recheck extension-appended scrape destinations before writing to prevent silent clobbering. --- src/scrapingbee_cli/batch.py | 4 +- src/scrapingbee_cli/cli_utils.py | 27 ++++- src/scrapingbee_cli/commands/scrape.py | 22 ++++- src/scrapingbee_cli/interactive.py | 4 +- tests/unit/test_v132_fixes.py | 132 +++++++++++++++++++++++++ 5 files changed, 179 insertions(+), 10 deletions(-) diff --git a/src/scrapingbee_cli/batch.py b/src/scrapingbee_cli/batch.py index 0f38fb7..e88a720 100644 --- a/src/scrapingbee_cli/batch.py +++ b/src/scrapingbee_cli/batch.py @@ -218,7 +218,9 @@ def read_input_file(path: str, *, input_column: str | None = None) -> list[str]: import sys as _sys if path != "-": - path = str(Path(path).expanduser()) + from .cli_utils import resolve_output_path + + path = resolve_output_path(path) if path == "-": lines = [line.strip() for line in _sys.stdin if line.strip()] diff --git a/src/scrapingbee_cli/cli_utils.py b/src/scrapingbee_cli/cli_utils.py index 8db41ad..d7518f1 100644 --- a/src/scrapingbee_cli/cli_utils.py +++ b/src/scrapingbee_cli/cli_utils.py @@ -321,8 +321,16 @@ def _batch_options(f: Any) -> Any: def resolve_output_path(path: str) -> str: - """Expand ``~`` in an output path.""" - return str(Path(path).expanduser()) + """Expand a leading ``~/`` (current user only). + + Only ``~`` and paths starting with ``~/`` (or ``~\\`` on Windows) are + expanded. Bare filenames like ``~data.csv`` and foreign homes like + ``~root/x`` are left literal — ``Path.expanduser()`` would otherwise + raise ``RuntimeError`` or write into another user's home. + """ + if path == "~" or path.startswith("~/") or (sys.platform == "win32" and path.startswith("~\\")): + return str(Path(path).expanduser()) + return path def ensure_output_file_ready( @@ -333,7 +341,7 @@ def ensure_output_file_ready( ) -> str: """Validate an output file path before any API work starts. - Expands ``~``, creates parent directories, and checks overwrite policy. + Expands ``~/``, creates parent directories, and checks overwrite policy. Returns the resolved path. """ resolved = resolve_output_path(path) @@ -350,7 +358,7 @@ def ensure_output_file_ready( def ensure_output_dir_ready(path: str) -> str: - """Expand ``~`` and create an output directory before API work.""" + """Expand ``~/`` and create an output directory before API work.""" resolved = resolve_output_path(path) try: Path(resolved).mkdir(parents=True, exist_ok=True) @@ -363,7 +371,7 @@ def ensure_output_dir_ready(path: str) -> str: def ensure_input_file_ready(path: str) -> str: """Validate an input file path before any API work starts. - Expands ``~`` and ensures the file exists and is readable. Stdin (``-``) is + Expands ``~/`` and ensures the file exists and is readable. Stdin (``-``) is passed through unchanged. """ if path == "-": @@ -1831,6 +1839,8 @@ def write_output( fields: str | None = None, command: str | None = None, credit_cost: int | None = None, + overwrite: bool = False, + skip_overwrite_check: bool = True, ) -> None: """Write response data to file or stdout; optionally print verbose headers. @@ -1838,6 +1848,11 @@ def write_output( language. When *extract_field* is set, extract from JSON using a path expression. When *fields* is set, filter JSON to specified fields. Precedence: *smart_extract* > *extract_field* > *fields*. + + *skip_overwrite_check* defaults to ``True`` because callers normally run + ``ensure_output_file_ready`` on the same path earlier. Pass + ``skip_overwrite_check=False`` when the final path may differ from the + early-validated one (e.g. scrape auto-appends an extension). """ if verbose: if is_repl_mode(): @@ -1910,6 +1925,8 @@ def write_output( except OSError as e: click.echo(f"Cannot create directory '{parent}': {e.strerror}", err=True) raise SystemExit(1) + if not skip_overwrite_check: + confirm_overwrite(resolved, overwrite) try: fh = open(resolved, "wb") except OSError as e: diff --git a/src/scrapingbee_cli/commands/scrape.py b/src/scrapingbee_cli/commands/scrape.py index 5c97756..c6cdfe9 100644 --- a/src/scrapingbee_cli/commands/scrape.py +++ b/src/scrapingbee_cli/commands/scrape.py @@ -774,12 +774,24 @@ async def _single() -> None: raise SystemExit(1) data = _apply_chunking(url or "", data, chunk_size, chunk_overlap) # Force .ndjson extension when chunking - output_path = obj["output_file"] + early_path = obj["output_file"] + output_path = early_path if output_path and "." not in os.path.basename(output_path): output_path = output_path.rstrip("/") + ".ndjson" - write_output(data, resp_headers, status_code, output_path, obj["verbose"]) + write_output( + data, + resp_headers, + status_code, + output_path, + obj["verbose"], + overwrite=bool(obj.get("overwrite", False)), + # Re-check when extension append changes the path after the + # early ensure_output_file_ready validation. + skip_overwrite_check=output_path == early_path, + ) return - output_path = obj["output_file"] + early_path = obj["output_file"] + output_path = early_path if output_path: if force_extension: if "." not in os.path.basename(output_path): @@ -795,6 +807,10 @@ async def _single() -> None: status_code, output_path, obj["verbose"], + overwrite=bool(obj.get("overwrite", False)), + # Re-check when extension append changes the path after the + # early ensure_output_file_ready validation. + skip_overwrite_check=output_path == early_path, ) asyncio.run(_single()) diff --git a/src/scrapingbee_cli/interactive.py b/src/scrapingbee_cli/interactive.py index 6741f09..6f47ab8 100644 --- a/src/scrapingbee_cli/interactive.py +++ b/src/scrapingbee_cli/interactive.py @@ -2154,7 +2154,9 @@ def _handle_meta( target_path = crawl_log missing_msg = "no crawl log yet — run `crawl ...` first" else: - target_path = Path(target_arg).expanduser() + from .cli_utils import resolve_output_path + + target_path = Path(resolve_output_path(target_arg)) missing_msg = f"file not found: {target_arg}" if not target_path.exists(): err_console.print(f" [{BEE_DIM}]{missing_msg}[/]") diff --git a/tests/unit/test_v132_fixes.py b/tests/unit/test_v132_fixes.py index b3ce005..aa34711 100644 --- a/tests/unit/test_v132_fixes.py +++ b/tests/unit/test_v132_fixes.py @@ -618,6 +618,43 @@ def test_resolve_output_path_expands_tilde(self, tmp_path, monkeypatch): _set_home(monkeypatch, tmp_path) assert resolve_output_path("~/out.png") == str(tmp_path / "out.png") + def test_resolve_output_path_expands_bare_tilde(self, tmp_path, monkeypatch): + from scrapingbee_cli.cli_utils import resolve_output_path + + _set_home(monkeypatch, tmp_path) + assert resolve_output_path("~") == str(tmp_path) + + def test_resolve_output_path_leaves_tilde_filename_literal(self, tmp_path, monkeypatch): + """``~data.csv`` is a legal filename — must not expanduser/crash.""" + from scrapingbee_cli.cli_utils import resolve_output_path + + _set_home(monkeypatch, tmp_path) + assert resolve_output_path("~data.csv") == "~data.csv" + + def test_resolve_output_path_leaves_foreign_home_literal(self, tmp_path, monkeypatch): + """``~root/...`` must not expand into another user's home.""" + from scrapingbee_cli.cli_utils import resolve_output_path + + _set_home(monkeypatch, tmp_path) + assert resolve_output_path("~root/x.png") == "~root/x.png" + + @pytest.mark.skipif(sys.platform == "win32", reason="POSIX path semantics") + def test_resolve_output_path_leaves_backslash_tilde_literal_on_posix( + self, tmp_path, monkeypatch + ): + """A backslash after ``~`` is not a path separator on POSIX.""" + from scrapingbee_cli.cli_utils import resolve_output_path + + _set_home(monkeypatch, tmp_path) + assert resolve_output_path(r"~\data.csv") == r"~\data.csv" + + @pytest.mark.skipif(sys.platform != "win32", reason="Windows path semantics") + def test_resolve_output_path_expands_backslash_tilde_on_windows(self, tmp_path, monkeypatch): + from scrapingbee_cli.cli_utils import resolve_output_path + + _set_home(monkeypatch, tmp_path) + assert resolve_output_path(r"~\data.csv") == str(tmp_path / "data.csv") + def test_ensure_output_file_ready_creates_parent_dirs(self, tmp_path): from scrapingbee_cli.cli_utils import ensure_output_file_ready @@ -626,6 +663,15 @@ def test_ensure_output_file_ready_creates_parent_dirs(self, tmp_path): assert resolved == str(out) assert out.parent.is_dir() + def test_ensure_output_file_ready_tilde_filename_literal(self, tmp_path, monkeypatch): + """``--output-file ~data.csv`` must not raise RuntimeError from expanduser.""" + from scrapingbee_cli.cli_utils import ensure_output_file_ready + + _set_home(monkeypatch, tmp_path) + monkeypatch.chdir(tmp_path) + resolved = ensure_output_file_ready("~data.csv", overwrite=True) + assert resolved == "~data.csv" + def test_ensure_output_file_ready_checks_overwrite_before_return(self, tmp_path, monkeypatch): from scrapingbee_cli.cli_utils import ensure_output_file_ready @@ -635,6 +681,29 @@ def test_ensure_output_file_ready_checks_overwrite_before_return(self, tmp_path, with pytest.raises(SystemExit): ensure_output_file_ready(str(existing), overwrite=False) + def test_write_output_checks_overwrite_when_path_changed(self, tmp_path, monkeypatch): + """Extension-append case: final path must not silently clobber.""" + from scrapingbee_cli.cli_utils import write_output + from scrapingbee_cli.theme import set_repl_mode + + existing = tmp_path / "report.html" + existing.write_bytes(b"old") + set_repl_mode(True) + try: + with pytest.raises(click.UsageError, match="already exists"): + write_output( + b"", + {}, + 200, + str(existing), + verbose=False, + overwrite=False, + skip_overwrite_check=False, + ) + finally: + set_repl_mode(False) + assert existing.read_bytes() == b"old" + def test_store_common_options_prepares_tilde_output_file(self, tmp_path, monkeypatch): from scrapingbee_cli.cli_utils import store_common_options @@ -670,6 +739,18 @@ def test_store_common_options_prepares_output_dir(self, tmp_path, monkeypatch): assert obj["output_dir"] == expected assert Path(expected).is_dir() + def test_store_common_options_tilde_filename_literal(self, tmp_path, monkeypatch): + """``--output-file ~data.csv`` is accepted as a literal cwd-relative name.""" + from scrapingbee_cli.cli_utils import store_common_options + + _set_home(monkeypatch, tmp_path) + monkeypatch.chdir(tmp_path) + obj: dict = {} + store_common_options( + obj, **TestStoreCommonOptionsBatchValidation()._make_obj(output_file="~data.csv") + ) + assert obj["output_file"] == "~data.csv" + class TestInputPathResolution: """Tests for ensure_input_file_ready / early --input-file validation.""" @@ -723,6 +804,15 @@ def test_read_input_file_expands_tilde(self, tmp_path, monkeypatch): input_file.write_text("https://example.com\n", encoding="utf-8") assert read_input_file("~/urls.txt") == ["https://example.com"] + def test_read_input_file_leaves_tilde_filename_literal(self, tmp_path, monkeypatch): + from scrapingbee_cli.batch import read_input_file + + _set_home(monkeypatch, tmp_path) + monkeypatch.chdir(tmp_path) + input_file = tmp_path / "~urls.txt" + input_file.write_text("https://example.com\n", encoding="utf-8") + assert read_input_file("~urls.txt") == ["https://example.com"] + class TestReplOutputPathResolution: """REPL-mode regressions for ~/ output paths (expand, mkdir, pre-scrape checks).""" @@ -872,6 +962,48 @@ def test_scrape_repl_expands_tilde_output_path_on_success(self, tmp_path, monkey assert out_file.read_bytes() == b"png-bytes" assert "screenshot.png" in combined.replace("\n", "") + def test_scrape_repl_extension_append_does_not_clobber(self, tmp_path, monkeypatch): + """``--output-file report`` must not silently overwrite existing report.html.""" + from unittest.mock import AsyncMock, patch + + from click.testing import CliRunner + + from scrapingbee_cli.commands.scrape import scrape_cmd + from scrapingbee_cli.theme import set_repl_mode + + monkeypatch.chdir(tmp_path) + existing = tmp_path / "report.html" + existing.write_bytes(b"keep-me") + + mock_client = AsyncMock() + mock_client.scrape.return_value = (b"new", {"Content-Type": "text/html"}, 200) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + + set_repl_mode(True) + try: + with patch("scrapingbee_cli.commands.scrape.get_api_key", return_value="fake"): + with patch("scrapingbee_cli.commands.scrape.Client", return_value=mock_client): + result = CliRunner().invoke( + scrape_cmd, + [ + "https://example.com", + "--render-js", + "false", + "--output-file", + "report", + ], + obj={}, + ) + finally: + set_repl_mode(False) + + combined = (result.output or "") + (result.stderr or "") + assert result.exit_code != 0, combined + assert "already exists" in combined + assert "--overwrite" in combined + assert existing.read_bytes() == b"keep-me" + # ============================================================================= # 9. store_common_options() — batch-only flags without --input-file