mirror of https://github.com/scrapy/scrapy.git
116 lines
3.7 KiB
Python
116 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import stat
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from scrapy.settings import Settings
|
|
from scrapy.utils._remote_control import (
|
|
JOB_FILE_VERSION,
|
|
job_files_dir,
|
|
new_job_file_name,
|
|
write_job_file,
|
|
)
|
|
|
|
|
|
def _write_job_file(path: Path) -> None:
|
|
write_job_file(
|
|
path,
|
|
spider="dummy",
|
|
project="testbot",
|
|
scrapy_version="2.17.0",
|
|
port=12345,
|
|
token="secret",
|
|
)
|
|
|
|
|
|
def test_write_job_file(tmp_path: Path) -> None:
|
|
path = tmp_path / "jobs" / f"{os.getpid()}-abc.json"
|
|
_write_job_file(path)
|
|
if sys.platform != "win32":
|
|
assert stat.S_IMODE(path.stat().st_mode) == 0o600
|
|
assert stat.S_IMODE(path.parent.stat().st_mode) == 0o700
|
|
record = json.loads(path.read_text(encoding="utf-8"))
|
|
assert record["version"] == JOB_FILE_VERSION
|
|
assert record["pid"] == os.getpid()
|
|
assert record["port"] == 12345
|
|
assert record["token"] == "secret"
|
|
assert record["spider"] == "dummy"
|
|
assert record["project"] == "testbot"
|
|
assert record["scrapy_version"] == "2.17.0"
|
|
assert isinstance(record["start_time"], float)
|
|
|
|
|
|
def test_write_job_file_leaves_no_temporary_file(tmp_path: Path) -> None:
|
|
name = f"{os.getpid()}-abc.json"
|
|
_write_job_file(tmp_path / name)
|
|
assert [path.name for path in tmp_path.iterdir()] == [name]
|
|
|
|
|
|
def test_new_job_file_name() -> None:
|
|
name = new_job_file_name()
|
|
assert name.endswith(".json")
|
|
pid, _, rest = name.removeprefix(".").partition("-")
|
|
assert rest
|
|
assert pid.isdigit()
|
|
assert int(pid) == os.getpid()
|
|
|
|
|
|
def test_write_job_file_refuses_an_existing_temporary_file(tmp_path: Path) -> None:
|
|
path = tmp_path / new_job_file_name()
|
|
planted = path.with_name(f".{path.name}.tmp")
|
|
planted.write_text("", encoding="utf-8")
|
|
with pytest.raises(FileExistsError):
|
|
_write_job_file(path)
|
|
assert not path.exists()
|
|
assert planted.read_text(encoding="utf-8") == "" # left untouched
|
|
|
|
|
|
def test_write_job_file_removes_the_temporary_file_after_a_failure(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
# A directory in the way makes the final rename fail, after the temporary
|
|
# file has already been created.
|
|
path = tmp_path / new_job_file_name()
|
|
path.mkdir()
|
|
# the specific exception differs between platforms
|
|
with pytest.raises(OSError): # noqa: PT011
|
|
_write_job_file(path)
|
|
assert list(tmp_path.iterdir()) == [path]
|
|
|
|
|
|
def test_write_job_file_ignores_a_failure_to_remove_the_temporary_file(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
def raise_runtime_error(*args: object, **kwargs: object) -> None:
|
|
raise RuntimeError("boom")
|
|
|
|
def raise_os_error(*args: object, **kwargs: object) -> None:
|
|
raise OSError("cannot remove")
|
|
|
|
monkeypatch.setattr(json, "dump", raise_runtime_error)
|
|
monkeypatch.setattr(Path, "unlink", raise_os_error)
|
|
path = tmp_path / new_job_file_name()
|
|
# the original error wins over the cleanup one
|
|
with pytest.raises(RuntimeError, match="boom"):
|
|
_write_job_file(path)
|
|
assert not path.exists()
|
|
assert path.with_name(f".{path.name}.tmp").exists() # could not be removed
|
|
|
|
|
|
def test_jobs_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
assert (
|
|
job_files_dir(Settings({"REMOTE_CONTROL_JOBS_DIR": str(tmp_path)})) == tmp_path
|
|
)
|
|
# platformdirs does not determine the user state folder from environment
|
|
# variables on every platform, hence the patching.
|
|
monkeypatch.setattr(
|
|
"scrapy.utils._remote_control.user_state_dir",
|
|
lambda *args, **kwargs: str(tmp_path),
|
|
)
|
|
assert job_files_dir(Settings()) == tmp_path / "job_files"
|