kde-6-widget-starter
Code Issues

fix(pling): pylance problems

Browse Source
This commit is contained in:
ruinivist
2026-04-13 18:13:19 +00:00
parent c108ccc8cf
commit 0928aa6d78
2 changed files with 181 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/pling_upload.py
+73 -30
View File
@@ -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,
tests/test_pling_upload.py
+108 -77
View File
@@ -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 = (
'<div data-addpploadfile-uri="/p/@@project_id@@/addpploadfile/" '
'data-updatepploadfile-uri="/p/@@project_id@@/updatepploadfile/" '
@@ -45,15 +64,15 @@ class PlingUploadTests(unittest.TestCase):
'data-deletepploadfiles-uri="/p/@@project_id@@/deletepploadfiles/" '
'data-product-id="" data-ppload-collection-id="123"></div>'
)
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()