92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import threading
|
|
import zipfile
|
|
from dataclasses import replace
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
from app.capture import CaptureService
|
|
from app.config import settings
|
|
from app.db import Database
|
|
|
|
|
|
class MinimalSiteHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None: # noqa: N802
|
|
if self.path == "/":
|
|
body = (
|
|
b"<html><head><link rel='stylesheet' href='/site.css'></head>"
|
|
b"<body>Home</body></html>"
|
|
)
|
|
content_type = "text/html"
|
|
status = 200
|
|
elif self.path == "/site.css":
|
|
body = b"body { color: #123456; }"
|
|
content_type = "text/css"
|
|
status = 200
|
|
else:
|
|
body = b"not found"
|
|
content_type = "text/plain"
|
|
status = 404
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, format: str, *args) -> None: # type: ignore[override]
|
|
return
|
|
|
|
|
|
def test_capture_service_creates_private_archive(tmp_path) -> None:
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), MinimalSiteHandler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
port = server.server_address[1]
|
|
|
|
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",
|
|
allow_private_networks=True,
|
|
allow_nonstandard_ports=True,
|
|
respect_robots=False,
|
|
)
|
|
test_settings.ensure_directories()
|
|
database = Database(test_settings.database_path)
|
|
database.initialize()
|
|
source_url = f"http://127.0.0.1:{port}/"
|
|
job = database.create_job(
|
|
source_url,
|
|
source_url,
|
|
{
|
|
"include_external_assets": False,
|
|
"max_pages": 10,
|
|
"max_depth": 2,
|
|
"max_bytes": 2_000_000,
|
|
"max_duration_seconds": 30,
|
|
"retention_days": 1,
|
|
},
|
|
)
|
|
claimed = database.claim_next("test-worker", lease_seconds=120)
|
|
assert claimed is not None
|
|
|
|
try:
|
|
CaptureService(database, test_settings, "test-worker").run(claimed)
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
completed = database.get_job(job["id"])
|
|
assert completed is not None
|
|
assert completed["state"] == "ready"
|
|
assert completed["artifact_path"]
|
|
assert completed["report_path"]
|
|
|
|
with zipfile.ZipFile(completed["artifact_path"]) as archive:
|
|
names = archive.namelist()
|
|
assert any(name.endswith("siteharbor-manifest.json") for name in names)
|
|
assert any(name.endswith("site.css") for name in names)
|