diff --git a/src/scrapingbee_cli/batch.py b/src/scrapingbee_cli/batch.py index 99b2b48..e88a720 100644 --- a/src/scrapingbee_cli/batch.py +++ b/src/scrapingbee_cli/batch.py @@ -217,6 +217,11 @@ 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 != "-": + 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()] elif path.lower().endswith(".csv"): diff --git a/src/scrapingbee_cli/cli_utils.py b/src/scrapingbee_cli/cli_utils.py index 4e5b48b..d7518f1 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,94 @@ def _batch_options(f: Any) -> Any: return f +def resolve_output_path(path: str) -> str: + """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( + 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 +460,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 +596,21 @@ def store_common_options(obj: dict, **kwargs: Any) -> None: ) raise SystemExit(1) + # Resolve paths before any API work (expand ~, mkdir outputs, validate inputs). + 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( + output_file_path, + overwrite=bool(obj.get("overwrite", False)), + skip_overwrite_check=bool(obj.get("update_csv")), + ) + 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]]: """Parse a path expression into typed segments. @@ -1748,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. @@ -1755,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(): @@ -1819,10 +1917,20 @@ 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) + if not skip_overwrite_check: + confirm_overwrite(resolved, overwrite) 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 +1940,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/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_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..aa34711 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() # ============================================================================= @@ -592,6 +604,407 @@ 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, tmp_path, monkeypatch): + from scrapingbee_cli.cli_utils import resolve_output_path + + _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 + + 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_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 + + 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_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 + + home = tmp_path / "home" + home.mkdir() + _set_home(monkeypatch, 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() + _set_home(monkeypatch, 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() + + 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.""" + + 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() + _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) + + 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() + _set_home(monkeypatch, 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() + _set_home(monkeypatch, 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() + _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"] + + 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).""" + + 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() + _set_home(monkeypatch, 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() + _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") + + 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() + _set_home(monkeypatch, 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() + _set_home(monkeypatch, 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", "") + + 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 # ============================================================================= @@ -688,16 +1101,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 +1125,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 +1135,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,