hi
", "author_name": "someone"}' monkeypatch.setattr(ingest_mod, "safe_fetch_text", fake_fetch_text) ingest_mod._fetch_tweet(url, None, None) query = urllib.parse.parse_qs(urllib.parse.urlsplit(requested[0]).query) assert query["url"] == [expected] @pytest.mark.parametrize( ("first_url", "second_url"), [ ("https://example.org/report.pdf?version=1", "https://example.org/report.pdf?version=2"), ("https://example.org/photo.png?version=1", "https://example.org/photo.png?version=2"), ("https://example.org/report.pdf", "https://example.org/report.pdf"), ("https://example.org/report+a.pdf", "https://example.org/report_a.pdf"), ("https://example.org/" + "a" * 90 + "1.pdf", "https://example.org/" + "a" * 90 + "2.pdf"), ], ) def test_binary_captures_preserve_colliding_payloads(first_url, second_url, tmp_path, monkeypatch): payloads = iter([b"first capture", b"second capture"]) fetched = [] def fetch(url): fetched.append(url) return next(payloads) monkeypatch.setattr(ingest_mod, "validate_url", lambda url: url) monkeypatch.setattr(ingest_mod, "safe_fetch", fetch) first = ingest_mod.ingest(first_url, tmp_path) second = ingest_mod.ingest(second_url, tmp_path) assert first != second assert second.name == f"{first.stem}_1{first.suffix}" assert first.read_bytes() == b"first capture" assert second.read_bytes() == b"second capture" assert fetched == [first_url, second_url] assert set(tmp_path.iterdir()) == {first, second} @pytest.mark.parametrize( ("entry_kind", "numbered"), [("file", False), ("file", True), ("dangling_link", False), ("dangling_link", True), ("live_link", True), ("directory", False), ("directory", True)], ) def test_binary_captures_skip_existing_entries(entry_kind, numbered, tmp_path, monkeypatch): corpus = tmp_path / "corpus" corpus.mkdir() preferred = corpus / "example_org_report_pdf.pdf" occupied = preferred.with_name("example_org_report_pdf_1.pdf") if numbered else preferred sentinel = tmp_path / "sentinel" if numbered: preferred.write_bytes(b"retained base") if entry_kind == "file": occupied.write_bytes(b"retained file") elif entry_kind == "directory": occupied.mkdir() else: if entry_kind == "live_link": sentinel.write_bytes(b"outside sentinel") try: occupied.symlink_to(sentinel) except (OSError, NotImplementedError) as exc: pytest.skip(f"host cannot create symlinks: {exc}") # Windows reads the target back as \\?\C:\..., which never equals `sentinel`. link_target = occupied.readlink() monkeypatch.setattr(ingest_mod, "validate_url", lambda url: url) monkeypatch.setattr(ingest_mod, "safe_fetch", lambda url: b"new capture") result = ingest_mod.ingest("https://example.org/report.pdf", corpus) expected_number = 2 if numbered else 1 assert result == corpus / f"example_org_report_pdf_{expected_number}.pdf" assert result.read_bytes() == b"new capture" if numbered: assert preferred.read_bytes() == b"retained base" if entry_kind == "file": assert occupied.read_bytes() == b"retained file" elif entry_kind == "directory": assert occupied.is_dir() and not list(occupied.iterdir()) else: assert occupied.is_symlink() and occupied.readlink() == link_target if entry_kind == "live_link": assert sentinel.read_bytes() == b"outside sentinel" else: assert not sentinel.exists() @pytest.mark.parametrize("error_type", [PermissionError, OSError]) def test_binary_capture_inspection_error_precedes_fetch(error_type, tmp_path, monkeypatch): from pathlib import Path preferred = tmp_path / "example_org_report_pdf.pdf" original_lstat = Path.lstat inspected = [] fetched = [] def lstat(path, *args, **kwargs): if path == preferred: inspected.append(path) raise error_type("cannot inspect destination") return original_lstat(path, *args, **kwargs) monkeypatch.setattr(Path, "lstat", lstat) monkeypatch.setattr(ingest_mod, "safe_fetch", lambda url: fetched.append(url) or b"new capture") with pytest.raises(error_type, match="cannot inspect destination"): ingest_mod._download_binary("https://example.org/report.pdf", ".pdf", tmp_path) assert inspected == [preferred] assert fetched == [] assert not list(tmp_path.iterdir()) @pytest.mark.parametrize("suffix", [".pdf", ".png"]) def test_binary_capture_free_name_is_unchanged(suffix, tmp_path, monkeypatch): monkeypatch.setattr(ingest_mod, "validate_url", lambda url: url) monkeypatch.setattr(ingest_mod, "safe_fetch", lambda url: b"capture") result = ingest_mod.ingest(f"https://example.org/report{suffix}", tmp_path) assert result == tmp_path / f"example_org_report_{suffix[1:]}{suffix}" assert result.read_bytes() == b"capture" def test_binary_capture_fetch_failure_preserves_existing_file(tmp_path, monkeypatch): retained = tmp_path / "example_org_report_pdf.pdf" retained.write_bytes(b"retained capture") def fail_fetch(url): raise OSError("download failed") monkeypatch.setattr(ingest_mod, "validate_url", lambda url: url) monkeypatch.setattr(ingest_mod, "safe_fetch", fail_fetch) with pytest.raises(RuntimeError, match="download failed"): ingest_mod.ingest("https://example.org/report.pdf", tmp_path) assert retained.read_bytes() == b"retained capture" assert list(tmp_path.iterdir()) == [retained] @pytest.mark.parametrize("entry_kind", ["dangling_link", "file"]) def test_text_captures_skip_existing_entries(entry_kind, tmp_path, monkeypatch): """A webpage capture must not replace an existing entry either; a dangling symlink in particular would otherwise be written through to its target.""" corpus = tmp_path / "corpus" corpus.mkdir() occupied = corpus / "example_org_page.md" sentinel = tmp_path / "sentinel" if entry_kind == "file": occupied.write_text("retained file", encoding="utf-8") else: try: occupied.symlink_to(sentinel) except (OSError, NotImplementedError) as exc: pytest.skip(f"host cannot create symlinks: {exc}") monkeypatch.setattr(ingest_mod, "validate_url", lambda url: url) monkeypatch.setattr(ingest_mod, "safe_fetch_text", lambda url: "