diff --git a/scripts/pling_upload.py b/scripts/pling_upload.py index cb7fb72..cbb842e 100755 --- a/scripts/pling_upload.py +++ b/scripts/pling_upload.py @@ -17,11 +17,11 @@ import time from dataclasses import dataclass from html.parser import HTMLParser from pathlib import Path -from typing import Dict, Iterable, Optional +from typing import Any, Iterable, Optional, Protocol, cast from urllib.parse import urljoin, urlparse try: - import niquests + import niquests # pyright: ignore[reportMissingImports] except ImportError as exc: # pragma: no cover - import guard print( "ERROR: missing dependency 'niquests'. Install it first, e.g. 'uv add niquests'.", @@ -63,7 +63,7 @@ class PlingUploaderError(RuntimeError): @dataclass(frozen=True) class LoginForm: action_url: str - hidden_fields: Dict[str, str] + hidden_fields: dict[str, str] @dataclass(frozen=True) @@ -91,13 +91,35 @@ class EditContext: file_server_owner_id: str +class ResponseLike(Protocol): + status_code: int | None + url: object + text: str | None + + def json(self) -> object: ... + + +class SessionLike(Protocol): + headers: dict[str, str] + cookies: object + + def request( + self, + method: str, + url: str, + *, + timeout: float, + **kwargs: Any, + ) -> ResponseLike: ... + + class _LoginFormParser(HTMLParser): def __init__(self) -> None: super().__init__() self._inside_form = False self._form_action: Optional[str] = None self._form_has_auth_fields = False - self._form_hidden_inputs: Dict[str, str] = {} + self._form_hidden_inputs: dict[str, str] = {} self.login_form: Optional[LoginForm] = None def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None: @@ -147,7 +169,7 @@ def _redact(key: str, value: object) -> object: def log_event(level: str, event: str, **fields: object) -> None: - payload = { + payload: dict[str, object] = { "level": level.upper(), "event": event, } @@ -163,22 +185,24 @@ def normalize_base_url(base_url: str) -> str: return normalized -def _safe_response_context(response: object) -> dict[str, object]: - status_code = getattr(response, "status_code", "unknown") - url = getattr(response, "url", "unknown") +def _safe_response_context(response: ResponseLike) -> dict[str, object]: + status_code: object = response.status_code + if status_code is None: + status_code = "unknown" + url = response.url return {"status_code": status_code, "url": str(url)} def request_with_retries( - session: "niquests.Session", + session: SessionLike, method: str, url: str, *, timeout: float, max_retries: int, retry_on_statuses: Iterable[int] = (429, 500, 502, 503, 504), - **kwargs: object, -): + **kwargs: Any, +) -> ResponseLike: attempt = 0 last_error: Optional[Exception] = None @@ -202,14 +226,15 @@ def request_with_retries( time.sleep(min(0.5 * attempt, 2.0)) continue - if response.status_code in retry_on_statuses and attempt <= max_retries: + status_code = response.status_code + if isinstance(status_code, int) and status_code in retry_on_statuses and attempt <= max_retries: log_event( "warning", "http_retryable_status", method=method, url=url, attempt=attempt, - status_code=response.status_code, + status_code=status_code, ) time.sleep(min(0.5 * attempt, 2.0)) continue @@ -224,16 +249,26 @@ def request_with_retries( raise PlingUploaderError(f"HTTP request failed after retries: {method} {url}") -def parse_json_response(response: object, *, context: str) -> dict[str, object]: +def parse_json_response(response: ResponseLike, *, context: str) -> dict[str, object]: try: - data = response.json() # type: ignore[attr-defined] + data = response.json() except Exception as exc: raise PlingUploaderError(f"{context}: expected JSON response.") from exc if not isinstance(data, dict): raise PlingUploaderError(f"{context}: response JSON shape is not an object.") - return data + normalized: dict[str, object] = {} + for key, value in data.items(): + normalized[str(key)] = value + return normalized + + +def get_response_text(response: ResponseLike, *, context: str) -> str: + text = response.text + if isinstance(text, str): + return text + raise PlingUploaderError(f"{context}: expected text response.") def parse_login_form(login_html: str, login_url: str) -> LoginForm: @@ -476,8 +511,8 @@ def resolve_runtime_config(args: argparse.Namespace) -> RuntimeConfig: ) -def create_session(base_url: str) -> "niquests.Session": - session = niquests.Session() +def create_session(base_url: str) -> SessionLike: + session = cast(SessionLike, cast(object, niquests.Session())) session.headers.update( { "User-Agent": ( @@ -489,12 +524,14 @@ def create_session(base_url: str) -> "niquests.Session": "Accept-Language": "en-US,en;q=0.9", } ) - session.cookies.set("verified", "1", domain=urlparse(base_url).hostname) + cookie_setter = getattr(session.cookies, "set", None) + if callable(cookie_setter): + cookie_setter("verified", "1", domain=urlparse(base_url).hostname) return session def discover_edit_context( - session: "niquests.Session", + session: SessionLike, config: RuntimeConfig, ) -> tuple[str, EditContext]: login_url = urljoin(config.base_url + "/", LOGIN_PATH.lstrip("/")) @@ -520,7 +557,8 @@ def discover_edit_context( ) log_event("info", "login_page_fetched", **_safe_response_context(login_page)) - login_form = parse_login_form(login_page.text, login_url) + login_page_html = get_response_text(login_page, context="login page") + login_form = parse_login_form(login_page_html, login_url) form_payload = dict(login_form.hidden_fields) form_payload.update({"email": config.username, "password": config.password}) @@ -534,7 +572,8 @@ def discover_edit_context( ) log_event("info", "login_submitted", **_safe_response_context(auth_response)) - if "incorrect login" in auth_response.text.lower(): + auth_response_html = get_response_text(auth_response, context="login response") + if "incorrect login" in auth_response_html.lower(): raise PlingUploaderError("Authentication failed: incorrect username or password.") edit_page = request_with_retries( @@ -546,7 +585,8 @@ def discover_edit_context( ) log_event("info", "edit_page_fetched", **_safe_response_context(edit_page)) - if looks_like_login_page(edit_page.text): + edit_page_html = get_response_text(edit_page, context="edit page") + if looks_like_login_page(edit_page_html): raise PlingUploaderError( "Authentication appears unsuccessful: redirected back to login page." ) @@ -554,7 +594,7 @@ def discover_edit_context( if str(edit_page.url).rstrip("/") == normalize_base_url(login_url).rstrip("/"): raise PlingUploaderError("Unable to access edit page; received login page URL.") - attrs = extract_required_data_attrs(edit_page.text, project_id=config.project_id) + attrs = extract_required_data_attrs(edit_page_html, project_id=config.project_id) resolved: dict[str, str] = {} for key, value in attrs.items(): @@ -564,9 +604,9 @@ def discover_edit_context( else: resolved[key] = replaced - client_id = extract_js_string_var(edit_page.text, "client_id") - file_uri = extract_js_string_var(edit_page.text, "fileUri") - owner_id = extract_form_append_literal(edit_page.text, "owner_id") + client_id = extract_js_string_var(edit_page_html, "client_id") + file_uri = extract_js_string_var(edit_page_html, "fileUri") + owner_id = extract_form_append_literal(edit_page_html, "owner_id") missing_runtime = [] if not client_id: @@ -581,6 +621,9 @@ def discover_edit_context( "Edit page is missing required upload runtime values: " + ", ".join(missing_runtime) ) + assert client_id is not None + assert file_uri is not None + assert owner_id is not None context = EditContext( add_file_url=resolved["data-addpploadfile-uri"], @@ -613,7 +656,7 @@ def discover_edit_context( def delete_all_existing_files( - session: "niquests.Session", + session: SessionLike, config: RuntimeConfig, edit_url: str, context: EditContext, @@ -646,7 +689,7 @@ def delete_all_existing_files( def upload_to_file_server( - session: "niquests.Session", + session: SessionLike, config: RuntimeConfig, context: EditContext, artifact_path: Path, @@ -691,7 +734,7 @@ def upload_to_file_server( def register_uploaded_file( - session: "niquests.Session", + session: SessionLike, config: RuntimeConfig, edit_url: str, context: EditContext, diff --git a/tests/test_pling_upload.py b/tests/test_pling_upload.py index 5974fac..34fd266 100644 --- a/tests/test_pling_upload.py +++ b/tests/test_pling_upload.py @@ -1,43 +1,62 @@ import argparse import importlib.util +import os import sys import tempfile import unittest +from collections.abc import Callable from pathlib import Path +from types import ModuleType +from typing import cast SCRIPT_PATH = Path(__file__).resolve().parents[1] / "scripts" / "pling_upload.py" SPEC = importlib.util.spec_from_file_location("pling_upload", SCRIPT_PATH) assert SPEC is not None and SPEC.loader is not None -MOD = importlib.util.module_from_spec(SPEC) +MOD = cast(ModuleType, importlib.util.module_from_spec(SPEC)) sys.modules[SPEC.name] = MOD SPEC.loader.exec_module(MOD) +resolve_cli_or_env = cast(Callable[..., str | None], getattr(MOD, "resolve_cli_or_env")) +extract_required_data_attrs = cast( + Callable[..., dict[str, str]], + getattr(MOD, "extract_required_data_attrs"), +) +resolve_runtime_config = cast(Callable[[argparse.Namespace], object], getattr(MOD, "resolve_runtime_config")) +run_upload_mode = cast(Callable[[object], int], getattr(MOD, "run_upload_mode")) +delete_all_existing_files = cast( + Callable[[object, object, str, object], None], + getattr(MOD, "delete_all_existing_files"), +) +RuntimeConfig = cast(Callable[..., object], getattr(MOD, "RuntimeConfig")) +EditContext = cast(Callable[..., object], getattr(MOD, "EditContext")) +PlingUploaderError = cast(type[Exception], getattr(MOD, "PlingUploaderError")) + class DummyResponse: - def __init__(self, payload): + def __init__(self, payload: dict[str, object]) -> None: self._payload = payload - def json(self): + def json(self) -> dict[str, object]: return self._payload class PlingUploadTests(unittest.TestCase): - def test_resolve_cli_or_env_prefers_cli(self): - old = MOD.os.environ.get("PLING_PROJECT_ID") - MOD.os.environ["PLING_PROJECT_ID"] = "from_env" + def test_resolve_cli_or_env_prefers_cli(self) -> None: + old = os.environ.get("PLING_PROJECT_ID") + os.environ["PLING_PROJECT_ID"] = "from_env" try: - value = MOD.resolve_cli_or_env( + value = resolve_cli_or_env( cli_value="from_cli", env_key="PLING_PROJECT_ID", ) self.assertEqual(value, "from_cli") finally: if old is None: - MOD.os.environ.pop("PLING_PROJECT_ID", None) + os.environ.pop("PLING_PROJECT_ID", None) else: - MOD.os.environ["PLING_PROJECT_ID"] = old + os.environ["PLING_PROJECT_ID"] = old - def test_extract_required_data_attrs_falls_back_product_id(self): + def test_extract_required_data_attrs_falls_back_product_id(self) -> None: html = ( '
' ) - attrs = MOD.extract_required_data_attrs(html, project_id="42") + attrs = extract_required_data_attrs(html, "42") self.assertEqual(attrs["data-product-id"], "42") - def test_resolve_runtime_config_requires_files_in_upload_mode(self): - old_env = dict(MOD.os.environ) + def test_resolve_runtime_config_requires_files_in_upload_mode(self) -> None: + old_env = dict(os.environ) try: - MOD.os.environ["PLING_PROJECT_ID"] = "1" - MOD.os.environ["PLING_USERNAME"] = "u" - MOD.os.environ["PLING_PASSWORD"] = "p" + os.environ["PLING_PROJECT_ID"] = "1" + os.environ["PLING_USERNAME"] = "u" + os.environ["PLING_PASSWORD"] = "p" args = argparse.Namespace( files=[], project_id=None, @@ -64,15 +83,15 @@ class PlingUploadTests(unittest.TestCase): max_retries=None, dry_run=False, ) - with self.assertRaises(MOD.PlingUploaderError): - MOD.resolve_runtime_config(args) + with self.assertRaises(PlingUploaderError): + resolve_runtime_config(args) finally: - MOD.os.environ.clear() - MOD.os.environ.update(old_env) + os.environ.clear() + os.environ.update(old_env) - def test_run_upload_mode_calls_steps_in_order(self): + def test_run_upload_mode_calls_steps_in_order(self) -> None: with tempfile.NamedTemporaryFile(suffix=".md") as tmp: - config = MOD.RuntimeConfig( + config = RuntimeConfig( project_id="1", base_url="https://example.com", username="u", @@ -82,7 +101,7 @@ class PlingUploadTests(unittest.TestCase): dry_run=False, artifact_paths=[Path(tmp.name)], ) - context = MOD.EditContext( + context = EditContext( add_file_url="https://example.com/add", update_file_url="https://example.com/update", delete_file_url="https://example.com/delete", @@ -94,45 +113,50 @@ class PlingUploadTests(unittest.TestCase): file_server_owner_id="4", ) - calls = [] - - create_session_old = MOD.create_session - discover_old = MOD.discover_edit_context - delete_old = MOD.delete_all_existing_files - upload_old = MOD.upload_to_file_server - register_old = MOD.register_uploaded_file + calls: list[str] = [] + create_session_old = getattr(MOD, "create_session") + discover_old = getattr(MOD, "discover_edit_context") + delete_old = getattr(MOD, "delete_all_existing_files") + upload_old = getattr(MOD, "upload_to_file_server") + register_old = getattr(MOD, "register_uploaded_file") try: - MOD.create_session = lambda _base: object() - MOD.discover_edit_context = lambda _s, _c: ("https://example.com/edit", context) + setattr(MOD, "create_session", lambda _base: object()) + setattr(MOD, "discover_edit_context", lambda _s, _c: ("https://example.com/edit", context)) - def fake_delete(_s, _cfg, _edit, _ctx): + def fake_delete(_s: object, _cfg: object, _edit: str, _ctx: object) -> None: calls.append("delete") - def fake_upload(_s, _cfg, _ctx, _file): + def fake_upload(_s: object, _cfg: object, _ctx: object, _file: Path) -> dict[str, object]: calls.append("upload") return {"id": "x"} - def fake_register(_s, _cfg, _edit, _ctx, _file): + def fake_register( + _s: object, + _cfg: object, + _edit: str, + _ctx: object, + _file: dict[str, object], + ) -> dict[str, object]: calls.append("register") return {"id": "x", "name": "test.md"} - MOD.delete_all_existing_files = fake_delete - MOD.upload_to_file_server = fake_upload - MOD.register_uploaded_file = fake_register + setattr(MOD, "delete_all_existing_files", fake_delete) + setattr(MOD, "upload_to_file_server", fake_upload) + setattr(MOD, "register_uploaded_file", fake_register) - rc = MOD.run_upload_mode(config) + rc = run_upload_mode(config) self.assertEqual(rc, 0) self.assertEqual(calls, ["delete", "upload", "register"]) finally: - MOD.create_session = create_session_old - MOD.discover_edit_context = discover_old - MOD.delete_all_existing_files = delete_old - MOD.upload_to_file_server = upload_old - MOD.register_uploaded_file = register_old + setattr(MOD, "create_session", create_session_old) + setattr(MOD, "discover_edit_context", discover_old) + setattr(MOD, "delete_all_existing_files", delete_old) + setattr(MOD, "upload_to_file_server", upload_old) + setattr(MOD, "register_uploaded_file", register_old) - def test_delete_all_existing_files_errors_on_non_ok(self): - config = MOD.RuntimeConfig( + def test_delete_all_existing_files_errors_on_non_ok(self) -> None: + config = RuntimeConfig( project_id="1", base_url="https://example.com", username="u", @@ -142,7 +166,7 @@ class PlingUploadTests(unittest.TestCase): dry_run=False, artifact_paths=[Path("/tmp/does-not-matter")], ) - context = MOD.EditContext( + context = EditContext( add_file_url="https://example.com/add", update_file_url="https://example.com/update", delete_file_url="https://example.com/delete", @@ -154,19 +178,19 @@ class PlingUploadTests(unittest.TestCase): file_server_owner_id="4", ) - old_request = MOD.request_with_retries + old_request = getattr(MOD, "request_with_retries") try: - MOD.request_with_retries = lambda *_a, **_kw: DummyResponse({"status": "error"}) - with self.assertRaises(MOD.PlingUploaderError): - MOD.delete_all_existing_files(object(), config, "https://example.com/edit", context) + setattr(MOD, "request_with_retries", lambda *_a, **_kw: DummyResponse({"status": "error"})) + with self.assertRaises(PlingUploaderError): + delete_all_existing_files(object(), config, "https://example.com/edit", context) finally: - MOD.request_with_retries = old_request + setattr(MOD, "request_with_retries", old_request) - def test_run_upload_mode_with_multiple_files(self): + def test_run_upload_mode_with_multiple_files(self) -> None: with tempfile.NamedTemporaryFile(suffix=".md") as tmp1, tempfile.NamedTemporaryFile( suffix=".md" ) as tmp2: - config = MOD.RuntimeConfig( + config = RuntimeConfig( project_id="1", base_url="https://example.com", username="u", @@ -176,7 +200,7 @@ class PlingUploadTests(unittest.TestCase): dry_run=False, artifact_paths=[Path(tmp1.name), Path(tmp2.name)], ) - context = MOD.EditContext( + context = EditContext( add_file_url="https://example.com/add", update_file_url="https://example.com/update", delete_file_url="https://example.com/delete", @@ -188,30 +212,37 @@ class PlingUploadTests(unittest.TestCase): file_server_owner_id="4", ) - calls = [] - create_session_old = MOD.create_session - discover_old = MOD.discover_edit_context - delete_old = MOD.delete_all_existing_files - upload_old = MOD.upload_to_file_server - register_old = MOD.register_uploaded_file + calls: list[str] = [] + create_session_old = getattr(MOD, "create_session") + discover_old = getattr(MOD, "discover_edit_context") + delete_old = getattr(MOD, "delete_all_existing_files") + upload_old = getattr(MOD, "upload_to_file_server") + register_old = getattr(MOD, "register_uploaded_file") try: - MOD.create_session = lambda _base: object() - MOD.discover_edit_context = lambda _s, _c: ("https://example.com/edit", context) - MOD.delete_all_existing_files = lambda *_a, **_kw: calls.append("delete") + setattr(MOD, "create_session", lambda _base: object()) + setattr(MOD, "discover_edit_context", lambda _s, _c: ("https://example.com/edit", context)) + setattr(MOD, "delete_all_existing_files", lambda *_a, **_kw: calls.append("delete")) - def fake_upload(_s, _cfg, _ctx, file_path): + def fake_upload(_s: object, _cfg: object, _ctx: object, file_path: Path) -> dict[str, object]: calls.append(f"upload:{Path(file_path).name}") return {"id": Path(file_path).name} - def fake_register(_s, _cfg, _edit, _ctx, file_payload): - calls.append(f"register:{file_payload['id']}") - return {"id": file_payload["id"], "name": file_payload["id"]} + def fake_register( + _s: object, + _cfg: object, + _edit: str, + _ctx: object, + file_payload: dict[str, object], + ) -> dict[str, object]: + file_id = str(file_payload["id"]) + calls.append(f"register:{file_id}") + return {"id": file_id, "name": file_id} - MOD.upload_to_file_server = fake_upload - MOD.register_uploaded_file = fake_register + setattr(MOD, "upload_to_file_server", fake_upload) + setattr(MOD, "register_uploaded_file", fake_register) - rc = MOD.run_upload_mode(config) + rc = run_upload_mode(config) self.assertEqual(rc, 0) self.assertEqual( calls, @@ -224,12 +255,12 @@ class PlingUploadTests(unittest.TestCase): ], ) finally: - MOD.create_session = create_session_old - MOD.discover_edit_context = discover_old - MOD.delete_all_existing_files = delete_old - MOD.upload_to_file_server = upload_old - MOD.register_uploaded_file = register_old + setattr(MOD, "create_session", create_session_old) + setattr(MOD, "discover_edit_context", discover_old) + setattr(MOD, "delete_all_existing_files", delete_old) + setattr(MOD, "upload_to_file_server", upload_old) + setattr(MOD, "register_uploaded_file", register_old) if __name__ == "__main__": - unittest.main() + _ = unittest.main()