91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.main as main
|
|
from app.config import settings
|
|
from app.db import Database
|
|
from app.scheduler import JobRunner
|
|
|
|
|
|
def test_capture_routes_are_scoped_to_the_browser_session(tmp_path, monkeypatch) -> None:
|
|
test_settings = replace(
|
|
settings,
|
|
data_dir=tmp_path,
|
|
database_path=tmp_path / "siteharbor.sqlite3",
|
|
work_dir=tmp_path / "work",
|
|
artifacts_dir=tmp_path / "artifacts",
|
|
reports_dir=tmp_path / "reports",
|
|
worker_enabled=False,
|
|
)
|
|
test_settings.ensure_directories()
|
|
test_database = Database(test_settings.database_path)
|
|
monkeypatch.setattr(main, "settings", test_settings)
|
|
monkeypatch.setattr(main, "database", test_database)
|
|
monkeypatch.setattr(main, "runner", JobRunner(test_database, test_settings))
|
|
|
|
owner_a = "a" * 32
|
|
owner_b = "b" * 32
|
|
with TestClient(main.app) as client:
|
|
homepage = client.get("/")
|
|
assert homepage.status_code == 200
|
|
assert "Copy the web." in homepage.text
|
|
assert client.get("/api/captures").status_code == 401
|
|
|
|
created = client.post(
|
|
"/api/captures",
|
|
headers={"X-SiteHarbor-Session": owner_a},
|
|
json={
|
|
"url": "https://example.com",
|
|
"parallel_connections": 2,
|
|
"download_asset_types": ["styles", "images"],
|
|
"skip_file_types": ["pdf, zip"],
|
|
},
|
|
)
|
|
assert created.status_code == 202
|
|
created_job = created.json()
|
|
job_id = created_job["id"]
|
|
assert created_job["options"]["parallel_connections"] == 2
|
|
assert created_job["options"]["download_asset_types"] == ["styles", "images"]
|
|
assert created_job["options"]["skip_file_types"] == ["pdf", "zip"]
|
|
|
|
assert client.get(
|
|
f"/api/captures/{job_id}", headers={"X-SiteHarbor-Session": owner_b}
|
|
).status_code == 404
|
|
assert client.post(
|
|
f"/api/captures/{job_id}/cancel", headers={"X-SiteHarbor-Session": owner_b}
|
|
).status_code == 404
|
|
|
|
owned_jobs = client.get("/api/captures", headers={"X-SiteHarbor-Session": owner_a})
|
|
assert owned_jobs.status_code == 200
|
|
assert [job["id"] for job in owned_jobs.json()["items"]] == [job_id]
|
|
|
|
stats = client.get("/api/stats")
|
|
assert stats.status_code == 200
|
|
assert stats.json()["active_crawls"] == 1
|
|
|
|
|
|
def test_run_forwards_server_options_to_uvicorn(monkeypatch) -> None:
|
|
calls: list[tuple[tuple[object, ...], dict[str, object]]] = []
|
|
|
|
def fake_run(*args: object, **kwargs: object) -> None:
|
|
calls.append((args, kwargs))
|
|
|
|
monkeypatch.setattr(main.uvicorn, "run", fake_run)
|
|
|
|
main.run(["--port", "9123", "--workers", "2"])
|
|
|
|
assert calls == [
|
|
(
|
|
("app.main:app",),
|
|
{
|
|
"host": settings.host,
|
|
"port": 9123,
|
|
"workers": 2,
|
|
"reload": False,
|
|
},
|
|
)
|
|
]
|