From bec2ef438263b4803783fa5dcbedc0a933645c4b Mon Sep 17 00:00:00 2001 From: smkc Date: Wed, 25 Feb 2026 22:32:42 +0100 Subject: [PATCH 1/3] feat(lighthouse): add full latarnia SDK+CLI support with models, docs, and tests --- README.md | 9 ++ docs/README.md | 1 + docs/api/README.md | 1 + docs/api/client.md | 1 + docs/api/lighthouse.md | 27 +++++ docs/cli/README.md | 31 ++++- docs/configuration.md | 21 ++++ src/ksef_client/__init__.py | 13 +- src/ksef_client/cli/app.py | 2 + .../cli/commands/lighthouse_cmd.py | 105 ++++++++++++++++ src/ksef_client/cli/sdk/adapters.py | 40 +++++++ src/ksef_client/cli/sdk/factory.py | 11 +- src/ksef_client/client.py | 31 ++++- src/ksef_client/clients/__init__.py | 3 + src/ksef_client/clients/lighthouse.py | 52 ++++++++ src/ksef_client/config.py | 23 ++++ src/ksef_client/models.py | 112 +++++++++++++++++- tests/cli/integration/test_lighthouse.py | 66 +++++++++++ tests/cli/unit/test_core_coverage.py | 1 + tests/cli/unit/test_sdk_adapters.py | 91 +++++++++++++- tests/test_client.py | 18 ++- tests/test_clients.py | 48 ++++++++ tests/test_config.py | 42 ++++++- tests/test_lighthouse_openapi_coverage.py | 102 ++++++++++++++++ tests/test_models.py | 52 ++++++++ 25 files changed, 892 insertions(+), 11 deletions(-) create mode 100644 docs/api/lighthouse.md create mode 100644 src/ksef_client/cli/commands/lighthouse_cmd.py create mode 100644 src/ksef_client/clients/lighthouse.py create mode 100644 tests/cli/integration/test_lighthouse.py create mode 100644 tests/test_lighthouse_openapi_coverage.py diff --git a/README.md b/README.md index 712711e..58a71d0 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,7 @@ Aktualna kompatybilność: **KSeF API `v2.1.2`** ([api-changelog.md](https://git - Uwierzytelnianie: token KSeF i podpis XAdES, w tym `XadesKeyPair` dla PKCS#12 lub PEM. - Workflows wysyłki: sesje online i batch z ZIP, partiami i pre-signed URL. - Eksport/pobieranie: obsługa paczek i narzędzi do odszyfrowania/rozpakowania. +- Latarnia: publiczne endpointy dostępności KSeF (`client.lighthouse`, `ksef lighthouse ...`). - Narzędzia pomocnicze: AES/ZIP/Base64Url, linki weryfikacyjne, QR. - CLI `ksef`: szybka ścieżka od konfiguracji do pierwszych operacji: `init -> auth -> invoice/send/upo`. @@ -98,6 +99,7 @@ Najważniejsze grupy komend: - auth: `auth login-token`, `auth login-xades`, `auth status`, `auth refresh`, `auth logout` - operacje: `invoice ...`, `send ...`, `upo ...`, `export ...` - diagnostyka: `health check` +- latarnia: `lighthouse status`, `lighthouse messages` Pełna specyfikacja CLI: [`docs/cli/README.md`](docs/cli/README.md) @@ -187,6 +189,13 @@ session_reference_number = BatchSessionWorkflow(client.sessions, client.http_cli ) ``` +### Odczyt statusu Latarni + +```python +status = client.lighthouse.get_status() +messages = client.lighthouse.get_messages() +``` + ## 📚 Dokumentacja Dokumentacja SDK znajduje się w `docs/`: diff --git a/docs/README.md b/docs/README.md index c0bef42..1e69c1e 100644 --- a/docs/README.md +++ b/docs/README.md @@ -47,6 +47,7 @@ Biblioteka udostępnia dwa poziomy użycia: - [`client.auth`](api/auth.md) - [`client.sessions`](api/sessions.md) - [`client.invoices`](api/invoices.md) +- [`client.lighthouse`](api/lighthouse.md) - [`client.permissions`](api/permissions.md) - [`client.certificates`](api/certificates.md) - [`client.tokens`](api/tokens.md) diff --git a/docs/api/README.md b/docs/api/README.md index bdcac76..471688a 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -12,6 +12,7 @@ Dla `AsyncKsefClient` zestaw metod jest taki sam – różni się tylko sposób - [`client.auth`](auth.md) - [`client.sessions`](sessions.md) - [`client.invoices`](invoices.md) +- [`client.lighthouse`](lighthouse.md) - [`client.permissions`](permissions.md) - [`client.certificates`](certificates.md) - [`client.tokens`](tokens.md) diff --git a/docs/api/client.md b/docs/api/client.md index 187f952..f2e12af 100644 --- a/docs/api/client.md +++ b/docs/api/client.md @@ -7,6 +7,7 @@ - `client.auth` - `client.sessions` - `client.invoices` +- `client.lighthouse` - `client.permissions` - `client.certificates` - `client.tokens` diff --git a/docs/api/lighthouse.md b/docs/api/lighthouse.md new file mode 100644 index 0000000..604cf7f --- /dev/null +++ b/docs/api/lighthouse.md @@ -0,0 +1,27 @@ +# Latarnia (`client.lighthouse`) + +`client.lighthouse` udostępnia publiczne endpointy Latarni KSeF (bez autoryzacji). + +## `get_status()` + +Endpoint: `GET /status` (API Latarni) + +Zwraca `LighthouseStatusResponse`: +- `status`: `AVAILABLE`, `MAINTENANCE`, `FAILURE`, `TOTAL_FAILURE` +- `messages`: komunikaty powiązane z aktualnym statusem (lub brak dla pełnej dostępności) + +## `get_messages()` + +Endpoint: `GET /messages` (API Latarni) + +Zwraca `list[LighthouseMessage]` z opublikowanymi komunikatami. + +## Środowiska + +Domyślne mapowanie dla `KsefClientOptions.base_url`: +- `TEST` -> `https://api-latarnia-test.ksef.mf.gov.pl` +- `DEMO` -> `https://api-latarnia-test.ksef.mf.gov.pl` +- `PROD` -> `https://api-latarnia.ksef.mf.gov.pl` + +Nadpisanie mapowania: +- `KsefClientOptions(base_lighthouse_url="https://...")` diff --git a/docs/cli/README.md b/docs/cli/README.md index 3b5d357..f73fa6c 100644 --- a/docs/cli/README.md +++ b/docs/cli/README.md @@ -20,6 +20,9 @@ CLI ma skracac droge od instalacji do pierwszej realnej operacji KSeF: - `auth logout` - diagnostyka: - `health check` +- latarnia: + - `lighthouse status` + - `lighthouse messages` - faktury i UPO: - `invoice list`, `invoice download` - `send online`, `send batch`, `send status` @@ -80,6 +83,9 @@ ksef logout health check + lighthouse + status + messages invoice list download @@ -117,7 +123,7 @@ Zachowanie profilu: - gdy nie podasz `--base-url`, CLI bierze kolejno: CLI option -> `KSEF_BASE_URL` -> `profile.base_url` -> DEMO. Brak aktywnego profilu: -- komendy `auth`, `health`, `invoice`, `send`, `upo`, `export` wymagaja aktywnego profilu, +- komendy `auth`, `health`, `lighthouse`, `invoice`, `send`, `upo`, `export` wymagaja aktywnego profilu, - jesli profil nie jest ustawiony, CLI zwraca czytelny blad z podpowiedzia: - `ksef init --set-active` - `ksef profile use --name ` @@ -292,6 +298,29 @@ Options: --base-url TEXT ``` +## `ksef lighthouse status` + +```text +Usage: ksef lighthouse status [OPTIONS] + +Options: + --base-url TEXT +``` + +Uwagi: +- komenda korzysta z publicznego API Latarni (bez tokenu dostepowego), +- domyslnie base URL Latarni mapowany jest z profilu KSeF (`TEST`/`DEMO` -> latarnia TEST, `PROD` -> latarnia PROD), +- `--base-url` lub `KSEF_LIGHTHOUSE_BASE_URL` pozwala nadpisac endpoint Latarni. + +## `ksef lighthouse messages` + +```text +Usage: ksef lighthouse messages [OPTIONS] + +Options: + --base-url TEXT +``` + ## invoice / send / upo / export ## `ksef invoice list` diff --git a/docs/configuration.md b/docs/configuration.md index 159a68b..22559a1 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -19,6 +19,7 @@ options = KsefClientOptions( allowed_presigned_hosts=None, allow_private_network_presigned_urls=False, base_qr_url=None, + base_lighthouse_url=None, ) ``` @@ -44,6 +45,26 @@ Używane przez `VerificationLinkService` do budowania linków pod QR. - Jeśli `base_qr_url` nie jest ustawione, biblioteka dobiera je na podstawie `base_url` (TEST/DEMO/PROD). - Dla niestandardowego `base_url` wymagane jest ustawienie `base_qr_url` jawnie. +### `base_lighthouse_url` + +Używane przez `client.lighthouse` (`/status`, `/messages` API Latarni). + +- Jeśli `base_lighthouse_url` nie jest ustawione, biblioteka mapuje je z `base_url`: + - KSeF `TEST` -> Latarnia `TEST` + - KSeF `DEMO` -> Latarnia `TEST` + - KSeF `PROD` -> Latarnia `PROD` +- Dla niestandardowego `base_url` można ustawić `base_lighthouse_url` jawnie. + +Gotowe stałe: + +```python +from ksef_client import KsefLighthouseEnvironment + +KsefLighthouseEnvironment.TEST.value +KsefLighthouseEnvironment.PROD.value +KsefLighthouseEnvironment.PRD.value # alias +``` + ### `timeout_seconds` Timeout dla pojedynczego żądania HTTP (httpx). W przypadku wysyłki partów i pobierania paczek eksportu może być wymagane zwiększenie wartości (duże paczki, wolniejsze łącza). diff --git a/src/ksef_client/__init__.py b/src/ksef_client/__init__.py index 58dc0fa..e819877 100644 --- a/src/ksef_client/__init__.py +++ b/src/ksef_client/__init__.py @@ -1,7 +1,7 @@ """KSeF Python SDK.""" from .client import AsyncKsefClient, KsefClient -from .config import KsefClientOptions, KsefEnvironment +from .config import KsefClientOptions, KsefEnvironment, KsefLighthouseEnvironment from .exceptions import KsefApiError, KsefHttpError, KsefRateLimitError from .models import ( AuthenticationChallengeResponse, @@ -11,6 +11,11 @@ InvoiceExportStatusResponse, InvoicePackage, InvoicePackagePart, + LighthouseKsefStatus, + LighthouseMessage, + LighthouseMessageCategory, + LighthouseMessageType, + LighthouseStatusResponse, StatusInfo, TokenInfo, ) @@ -18,6 +23,7 @@ __all__ = [ "KsefClientOptions", "KsefEnvironment", + "KsefLighthouseEnvironment", "KsefClient", "AsyncKsefClient", "KsefApiError", @@ -27,6 +33,11 @@ "AuthenticationInitResponse", "AuthenticationTokensResponse", "AuthenticationTokenRefreshResponse", + "LighthouseKsefStatus", + "LighthouseMessageCategory", + "LighthouseMessageType", + "LighthouseMessage", + "LighthouseStatusResponse", "TokenInfo", "StatusInfo", "InvoicePackage", diff --git a/src/ksef_client/cli/app.py b/src/ksef_client/cli/app.py index ce5ceed..26082a0 100644 --- a/src/ksef_client/cli/app.py +++ b/src/ksef_client/cli/app.py @@ -10,6 +10,7 @@ health_cmd, init_cmd, invoice_cmd, + lighthouse_cmd, profile_cmd, send_cmd, upo_cmd, @@ -79,6 +80,7 @@ def main( app.add_typer(profile_cmd.app, name="profile") app.add_typer(auth_cmd.app, name="auth") app.add_typer(health_cmd.app, name="health") +app.add_typer(lighthouse_cmd.app, name="lighthouse") app.add_typer(invoice_cmd.app, name="invoice") app.add_typer(send_cmd.app, name="send") app.add_typer(upo_cmd.app, name="upo") diff --git a/src/ksef_client/cli/commands/lighthouse_cmd.py b/src/ksef_client/cli/commands/lighthouse_cmd.py new file mode 100644 index 0000000..b86ad4f --- /dev/null +++ b/src/ksef_client/cli/commands/lighthouse_cmd.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import os + +import typer + +from ksef_client.exceptions import KsefApiError, KsefHttpError, KsefRateLimitError + +from ..auth.manager import resolve_base_url +from ..context import profile_label, require_context, require_profile +from ..errors import CliError +from ..exit_codes import ExitCode +from ..output import get_renderer +from ..sdk.adapters import get_lighthouse_messages, get_lighthouse_status + +app = typer.Typer(help="Read public KSeF Lighthouse status and messages.") + + +def _render_error(ctx: typer.Context, command: str, exc: Exception) -> None: + cli_ctx = require_context(ctx) + renderer = get_renderer(cli_ctx) + + if isinstance(exc, CliError): + renderer.error( + command=command, + profile=profile_label(cli_ctx), + code=exc.code.name, + message=exc.message, + hint=exc.hint, + ) + raise typer.Exit(int(exc.code)) + + if isinstance(exc, KsefRateLimitError): + hint = f"Retry-After: {exc.retry_after}" if exc.retry_after else "Wait and retry." + renderer.error( + command=command, + profile=profile_label(cli_ctx), + code="RATE_LIMIT", + message=str(exc), + hint=hint, + ) + raise typer.Exit(int(ExitCode.RETRY_EXHAUSTED)) + + if isinstance(exc, (KsefApiError, KsefHttpError)): + renderer.error( + command=command, + profile=profile_label(cli_ctx), + code="API_ERROR", + message=str(exc), + hint="Check Lighthouse response and request parameters.", + ) + raise typer.Exit(int(ExitCode.API_ERROR)) + + renderer.error( + command=command, + profile=profile_label(cli_ctx), + code="UNEXPECTED", + message=str(exc), + hint="Run with -v and inspect logs.", + ) + raise typer.Exit(int(ExitCode.CONFIG_ERROR)) + + +@app.command("status") +def lighthouse_status( + ctx: typer.Context, + base_url: str | None = typer.Option( + None, "--base-url", help="Override Lighthouse base URL for this command." + ), +) -> None: + cli_ctx = require_context(ctx) + renderer = get_renderer(cli_ctx) + profile = profile_label(cli_ctx) + try: + profile = require_profile(cli_ctx) + result = get_lighthouse_status( + profile=profile, + base_url=resolve_base_url(os.getenv("KSEF_BASE_URL"), profile=profile), + lighthouse_base_url=base_url or os.getenv("KSEF_LIGHTHOUSE_BASE_URL"), + ) + except Exception as exc: + _render_error(ctx, "lighthouse.status", exc) + renderer.success(command="lighthouse.status", profile=profile, data=result) + + +@app.command("messages") +def lighthouse_messages( + ctx: typer.Context, + base_url: str | None = typer.Option( + None, "--base-url", help="Override Lighthouse base URL for this command." + ), +) -> None: + cli_ctx = require_context(ctx) + renderer = get_renderer(cli_ctx) + profile = profile_label(cli_ctx) + try: + profile = require_profile(cli_ctx) + result = get_lighthouse_messages( + profile=profile, + base_url=resolve_base_url(os.getenv("KSEF_BASE_URL"), profile=profile), + lighthouse_base_url=base_url or os.getenv("KSEF_LIGHTHOUSE_BASE_URL"), + ) + except Exception as exc: + _render_error(ctx, "lighthouse.messages", exc) + renderer.success(command="lighthouse.messages", profile=profile, data=result) diff --git a/src/ksef_client/cli/sdk/adapters.py b/src/ksef_client/cli/sdk/adapters.py index 1e118bf..cabac93 100644 --- a/src/ksef_client/cli/sdk/adapters.py +++ b/src/ksef_client/cli/sdk/adapters.py @@ -427,6 +427,46 @@ def _wait_for_export_status( ) +def get_lighthouse_status( + *, + profile: str, + base_url: str, + lighthouse_base_url: str | None = None, +) -> dict[str, Any]: + _ = profile + with create_client( + base_url, + base_lighthouse_url=lighthouse_base_url, + ) as client: + status = client.lighthouse.get_status() + + messages = [message.to_dict() for message in (status.messages or [])] + return { + "status": status.status.value, + "messages": messages, + } + + +def get_lighthouse_messages( + *, + profile: str, + base_url: str, + lighthouse_base_url: str | None = None, +) -> dict[str, Any]: + _ = profile + with create_client( + base_url, + base_lighthouse_url=lighthouse_base_url, + ) as client: + messages = client.lighthouse.get_messages() + + items = [message.to_dict() for message in messages] + return { + "count": len(items), + "items": items, + } + + def list_invoices( *, profile: str, diff --git a/src/ksef_client/cli/sdk/factory.py b/src/ksef_client/cli/sdk/factory.py index bb63239..575c9d2 100644 --- a/src/ksef_client/cli/sdk/factory.py +++ b/src/ksef_client/cli/sdk/factory.py @@ -3,5 +3,12 @@ from ksef_client import KsefClient, KsefClientOptions -def create_client(base_url: str, access_token: str | None = None) -> KsefClient: - return KsefClient(KsefClientOptions(base_url=base_url), access_token=access_token) +def create_client( + base_url: str, + access_token: str | None = None, + base_lighthouse_url: str | None = None, +) -> KsefClient: + return KsefClient( + KsefClientOptions(base_url=base_url, base_lighthouse_url=base_lighthouse_url), + access_token=access_token, + ) diff --git a/src/ksef_client/client.py b/src/ksef_client/client.py index 4ebb1f1..025f814 100644 --- a/src/ksef_client/client.py +++ b/src/ksef_client/client.py @@ -3,6 +3,7 @@ from .clients.auth import AsyncAuthClient, AuthClient from .clients.certificates import AsyncCertificatesClient, CertificatesClient from .clients.invoices import AsyncInvoicesClient, InvoicesClient +from .clients.lighthouse import AsyncLighthouseClient, LighthouseClient from .clients.limits import AsyncLimitsClient, LimitsClient from .clients.peppol import AsyncPeppolClient, PeppolClient from .clients.permissions import AsyncPermissionsClient, PermissionsClient @@ -18,9 +19,19 @@ class KsefClient: def __init__(self, options: KsefClientOptions, access_token: str | None = None) -> None: self._http = BaseHttpClient(options, access_token=access_token) + self._lighthouse_http = BaseHttpClient(options, access_token=None) + lighthouse_base_url = "" + try: + lighthouse_base_url = options.resolve_lighthouse_base_url() + except ValueError: + lighthouse_base_url = "" self.auth = AuthClient(self._http) self.sessions = SessionsClient(self._http) self.invoices = InvoicesClient(self._http) + self.lighthouse = LighthouseClient( + self._lighthouse_http, + lighthouse_base_url, + ) self.permissions = PermissionsClient(self._http) self.certificates = CertificatesClient(self._http) self.tokens = TokensClient(self._http) @@ -31,7 +42,10 @@ def __init__(self, options: KsefClientOptions, access_token: str | None = None) self.peppol = PeppolClient(self._http) def close(self) -> None: - self._http.close() + try: + self._http.close() + finally: + self._lighthouse_http.close() @property def http_client(self) -> BaseHttpClient: @@ -47,9 +61,19 @@ def __exit__(self, exc_type, exc, tb) -> None: class AsyncKsefClient: def __init__(self, options: KsefClientOptions, access_token: str | None = None) -> None: self._http = AsyncBaseHttpClient(options, access_token=access_token) + self._lighthouse_http = AsyncBaseHttpClient(options, access_token=None) + lighthouse_base_url = "" + try: + lighthouse_base_url = options.resolve_lighthouse_base_url() + except ValueError: + lighthouse_base_url = "" self.auth = AsyncAuthClient(self._http) self.sessions = AsyncSessionsClient(self._http) self.invoices = AsyncInvoicesClient(self._http) + self.lighthouse = AsyncLighthouseClient( + self._lighthouse_http, + lighthouse_base_url, + ) self.permissions = AsyncPermissionsClient(self._http) self.certificates = AsyncCertificatesClient(self._http) self.tokens = AsyncTokensClient(self._http) @@ -60,7 +84,10 @@ def __init__(self, options: KsefClientOptions, access_token: str | None = None) self.peppol = AsyncPeppolClient(self._http) async def aclose(self) -> None: - await self._http.aclose() + try: + await self._http.aclose() + finally: + await self._lighthouse_http.aclose() @property def http_client(self) -> AsyncBaseHttpClient: diff --git a/src/ksef_client/clients/__init__.py b/src/ksef_client/clients/__init__.py index 88fb52a..ac82945 100644 --- a/src/ksef_client/clients/__init__.py +++ b/src/ksef_client/clients/__init__.py @@ -1,6 +1,7 @@ from .auth import AsyncAuthClient, AuthClient from .certificates import AsyncCertificatesClient, CertificatesClient from .invoices import AsyncInvoicesClient, InvoicesClient +from .lighthouse import AsyncLighthouseClient, LighthouseClient from .limits import AsyncLimitsClient, LimitsClient from .peppol import AsyncPeppolClient, PeppolClient from .permissions import AsyncPermissionsClient, PermissionsClient @@ -17,6 +18,8 @@ "AsyncSessionsClient", "InvoicesClient", "AsyncInvoicesClient", + "LighthouseClient", + "AsyncLighthouseClient", "PermissionsClient", "AsyncPermissionsClient", "CertificatesClient", diff --git a/src/ksef_client/clients/lighthouse.py b/src/ksef_client/clients/lighthouse.py new file mode 100644 index 0000000..15b7fc2 --- /dev/null +++ b/src/ksef_client/clients/lighthouse.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +from typing import Any + +from ..models import LighthouseMessage, LighthouseStatusResponse +from .base import AsyncBaseApiClient, BaseApiClient + + +class LighthouseClient(BaseApiClient): + def __init__(self, http_client: Any, base_url: str) -> None: + super().__init__(http_client) + self._base_url = base_url.rstrip("/") + + def _require_base_url(self) -> str: + if self._base_url: + return self._base_url + raise ValueError("Unknown KSeF environment; set base_lighthouse_url explicitly.") + + def get_status(self) -> LighthouseStatusResponse: + payload = self._request_json("GET", f"{self._require_base_url()}/status") + if not isinstance(payload, dict): + payload = {} + return LighthouseStatusResponse.from_dict(payload) + + def get_messages(self) -> list[LighthouseMessage]: + payload = self._request_json("GET", f"{self._require_base_url()}/messages") + if not isinstance(payload, list): + return [] + return [LighthouseMessage.from_dict(item) for item in payload if isinstance(item, dict)] + + +class AsyncLighthouseClient(AsyncBaseApiClient): + def __init__(self, http_client: Any, base_url: str) -> None: + super().__init__(http_client) + self._base_url = base_url.rstrip("/") + + def _require_base_url(self) -> str: + if self._base_url: + return self._base_url + raise ValueError("Unknown KSeF environment; set base_lighthouse_url explicitly.") + + async def get_status(self) -> LighthouseStatusResponse: + payload = await self._request_json("GET", f"{self._require_base_url()}/status") + if not isinstance(payload, dict): + payload = {} + return LighthouseStatusResponse.from_dict(payload) + + async def get_messages(self) -> list[LighthouseMessage]: + payload = await self._request_json("GET", f"{self._require_base_url()}/messages") + if not isinstance(payload, list): + return [] + return [LighthouseMessage.from_dict(item) for item in payload if isinstance(item, dict)] diff --git a/src/ksef_client/config.py b/src/ksef_client/config.py index 16f29e4..6998123 100644 --- a/src/ksef_client/config.py +++ b/src/ksef_client/config.py @@ -17,6 +17,12 @@ class KsefQrEnvironment(str, Enum): PROD = "https://qr.ksef.mf.gov.pl" +class KsefLighthouseEnvironment(str, Enum): + TEST = "https://api-latarnia-test.ksef.mf.gov.pl" + PROD = "https://api-latarnia.ksef.mf.gov.pl" + PRD = PROD + + def _package_version() -> str: try: return metadata.version("ksef-client") @@ -35,6 +41,7 @@ def _default_user_agent() -> str: class KsefClientOptions: base_url: str base_qr_url: str | None = None + base_lighthouse_url: str | None = None timeout_seconds: float = 30.0 proxy: str | None = None custom_headers: dict[str, str] | None = None @@ -63,3 +70,19 @@ def resolve_qr_base_url(self) -> str: if base.startswith(KsefEnvironment.PROD.value): return KsefQrEnvironment.PROD.value raise ValueError("Unknown KSeF environment; set base_qr_url explicitly.") + + def resolve_lighthouse_base_url(self) -> str: + if self.base_lighthouse_url: + return self.base_lighthouse_url.rstrip("/") + base = self.base_url.rstrip("/") + if base.startswith(KsefLighthouseEnvironment.TEST.value): + return KsefLighthouseEnvironment.TEST.value + if base.startswith(KsefLighthouseEnvironment.PROD.value): + return KsefLighthouseEnvironment.PROD.value + if base.startswith(KsefEnvironment.TEST.value): + return KsefLighthouseEnvironment.TEST.value + if base.startswith(KsefEnvironment.DEMO.value): + return KsefLighthouseEnvironment.TEST.value + if base.startswith(KsefEnvironment.PROD.value): + return KsefLighthouseEnvironment.PROD.value + raise ValueError("Unknown KSeF environment; set base_lighthouse_url explicitly.") diff --git a/src/ksef_client/models.py b/src/ksef_client/models.py index 5f94885..414d455 100644 --- a/src/ksef_client/models.py +++ b/src/ksef_client/models.py @@ -1,7 +1,8 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any +from enum import Enum +from typing import Any, TypeVar @dataclass(frozen=True) @@ -108,6 +109,115 @@ def from_dict(data: dict[str, Any]) -> AuthenticationTokenRefreshResponse: ) +class LighthouseKsefStatus(str, Enum): + AVAILABLE = "AVAILABLE" + MAINTENANCE = "MAINTENANCE" + FAILURE = "FAILURE" + TOTAL_FAILURE = "TOTAL_FAILURE" + + +class LighthouseMessageCategory(str, Enum): + FAILURE = "FAILURE" + TOTAL_FAILURE = "TOTAL_FAILURE" + MAINTENANCE = "MAINTENANCE" + + +class LighthouseMessageType(str, Enum): + FAILURE_START = "FAILURE_START" + FAILURE_END = "FAILURE_END" + MAINTENANCE_ANNOUNCEMENT = "MAINTENANCE_ANNOUNCEMENT" + + +EnumT = TypeVar("EnumT", bound=Enum) + + +def _parse_enum(value: Any, enum_type: type[EnumT], default: EnumT) -> EnumT: + try: + return enum_type(value) + except Exception: + return default + + +@dataclass(frozen=True) +class LighthouseMessage: + id: str + event_id: int + category: LighthouseMessageCategory + type: LighthouseMessageType + title: str + text: str + start: str + end: str | None + version: int + published: str + + @staticmethod + def from_dict(data: dict[str, Any]) -> LighthouseMessage: + return LighthouseMessage( + id=str(data.get("id", "")), + event_id=int(data.get("eventId", 0)), + category=_parse_enum( + data.get("category"), + LighthouseMessageCategory, + LighthouseMessageCategory.FAILURE, + ), + type=_parse_enum( + data.get("type"), + LighthouseMessageType, + LighthouseMessageType.FAILURE_START, + ), + title=str(data.get("title", "")), + text=str(data.get("text", "")), + start=str(data.get("start", "")), + end=str(data["end"]) if data.get("end") is not None else None, + version=int(data.get("version", 0)), + published=str(data.get("published", "")), + ) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "eventId": self.event_id, + "category": self.category.value, + "type": self.type.value, + "title": self.title, + "text": self.text, + "start": self.start, + "end": self.end, + "version": self.version, + "published": self.published, + } + + +@dataclass(frozen=True) +class LighthouseStatusResponse: + status: LighthouseKsefStatus + messages: list[LighthouseMessage] | None = None + + @staticmethod + def from_dict(data: dict[str, Any]) -> LighthouseStatusResponse: + raw_messages = data.get("messages") + messages = ( + [LighthouseMessage.from_dict(item) for item in raw_messages] + if isinstance(raw_messages, list) + else None + ) + return LighthouseStatusResponse( + status=_parse_enum( + data.get("status"), + LighthouseKsefStatus, + LighthouseKsefStatus.AVAILABLE, + ), + messages=messages, + ) + + def to_dict(self) -> dict[str, Any]: + payload: dict[str, Any] = {"status": self.status.value} + if self.messages is not None: + payload["messages"] = [message.to_dict() for message in self.messages] + return payload + + @dataclass(frozen=True) class InvoicePackagePart: ordinal_number: int diff --git a/tests/cli/integration/test_lighthouse.py b/tests/cli/integration/test_lighthouse.py new file mode 100644 index 0000000..c772ab6 --- /dev/null +++ b/tests/cli/integration/test_lighthouse.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import json + +from ksef_client.cli.app import app +from ksef_client.cli.commands import lighthouse_cmd +from ksef_client.cli.errors import CliError +from ksef_client.cli.exit_codes import ExitCode + + +def _json_output(text: str) -> dict: + return json.loads(text.strip().splitlines()[-1]) + + +def test_lighthouse_help(runner) -> None: + result = runner.invoke(app, ["lighthouse", "--help"]) + assert result.exit_code == 0 + + +def test_lighthouse_status_success(runner, monkeypatch) -> None: + seen: dict[str, object] = {} + + def _fake_status(**kwargs): + seen.update(kwargs) + return {"status": "AVAILABLE", "messages": []} + + monkeypatch.setattr(lighthouse_cmd, "get_lighthouse_status", _fake_status) + result = runner.invoke( + app, + [ + "lighthouse", + "status", + "--base-url", + "https://api-latarnia-test.ksef.mf.gov.pl", + ], + ) + assert result.exit_code == 0 + assert seen["profile"] == "demo" + assert seen["lighthouse_base_url"] == "https://api-latarnia-test.ksef.mf.gov.pl" + assert "lighthouse.status" in result.stdout + + +def test_lighthouse_messages_json_success(runner, monkeypatch) -> None: + monkeypatch.setattr( + lighthouse_cmd, + "get_lighthouse_messages", + lambda **kwargs: {"count": 1, "items": [{"id": "m-1"}]}, + ) + result = runner.invoke(app, ["--json", "lighthouse", "messages"]) + assert result.exit_code == 0 + payload = _json_output(result.stdout) + assert payload["ok"] is True + assert payload["command"] == "lighthouse.messages" + assert payload["data"]["count"] == 1 + + +def test_lighthouse_status_validation_error_exit_code(runner, monkeypatch) -> None: + monkeypatch.setattr( + lighthouse_cmd, + "get_lighthouse_status", + lambda **kwargs: (_ for _ in ()).throw( + CliError("bad input", ExitCode.VALIDATION_ERROR, "fix args") + ), + ) + result = runner.invoke(app, ["lighthouse", "status"]) + assert result.exit_code == int(ExitCode.VALIDATION_ERROR) diff --git a/tests/cli/unit/test_core_coverage.py b/tests/cli/unit/test_core_coverage.py index a2eff75..871e867 100644 --- a/tests/cli/unit/test_core_coverage.py +++ b/tests/cli/unit/test_core_coverage.py @@ -97,6 +97,7 @@ def test_factory_create_client() -> None: client = create_client("https://api-demo.ksef.mf.gov.pl") try: assert isinstance(client, KsefClient) + assert client.lighthouse is not None finally: client.close() diff --git a/tests/cli/unit/test_sdk_adapters.py b/tests/cli/unit/test_sdk_adapters.py index a1a1f97..7bcc4d9 100644 --- a/tests/cli/unit/test_sdk_adapters.py +++ b/tests/cli/unit/test_sdk_adapters.py @@ -11,10 +11,19 @@ class _FakeClient: - def __init__(self, *, invoices=None, sessions=None, security=None, http_client=None) -> None: + def __init__( + self, + *, + invoices=None, + sessions=None, + security=None, + lighthouse=None, + http_client=None, + ) -> None: self.invoices = invoices self.sessions = sessions self.security = security + self.lighthouse = lighthouse self.http_client = http_client def __enter__(self) -> _FakeClient: @@ -124,6 +133,86 @@ def test_list_invoices_rejects_reverse_date_range(monkeypatch) -> None: assert exc.value.code == ExitCode.VALIDATION_ERROR +def test_get_lighthouse_status_success(monkeypatch) -> None: + seen: dict[str, object] = {} + + class _Lighthouse: + def get_status(self): + message = SimpleNamespace( + to_dict=lambda: { + "id": "m-1", + "eventId": 1, + "category": "MAINTENANCE", + "type": "MAINTENANCE_ANNOUNCEMENT", + "title": "Planned", + "text": "Window", + "start": "2026-03-15T01:00:00Z", + "end": "2026-03-15T06:00:00Z", + "version": 1, + "published": "2026-03-10T12:00:00Z", + } + ) + return SimpleNamespace(status=SimpleNamespace(value="MAINTENANCE"), messages=[message]) + + def _fake_create_client(base_url, access_token=None, base_lighthouse_url=None): + seen["base_url"] = base_url + seen["access_token"] = access_token + seen["base_lighthouse_url"] = base_lighthouse_url + return _FakeClient(lighthouse=_Lighthouse()) + + monkeypatch.setattr(adapters, "create_client", _fake_create_client) + + result = adapters.get_lighthouse_status( + profile="demo", + base_url="https://api-demo.ksef.mf.gov.pl", + lighthouse_base_url="https://api-latarnia-test.ksef.mf.gov.pl", + ) + + assert seen["base_url"] == "https://api-demo.ksef.mf.gov.pl" + assert seen["base_lighthouse_url"] == "https://api-latarnia-test.ksef.mf.gov.pl" + assert seen["access_token"] is None + assert result["status"] == "MAINTENANCE" + assert len(result["messages"]) == 1 + + +def test_get_lighthouse_messages_success(monkeypatch) -> None: + class _Lighthouse: + def get_messages(self): + return [ + SimpleNamespace( + to_dict=lambda: { + "id": "m-1", + "eventId": 1, + "category": "FAILURE", + "type": "FAILURE_START", + "title": "Failure", + "text": "Down", + "start": "2026-05-12T10:00:00Z", + "end": None, + "version": 1, + "published": "2026-05-12T10:01:00Z", + } + ) + ] + + monkeypatch.setattr( + adapters, + "create_client", + lambda base_url, access_token=None, base_lighthouse_url=None: _FakeClient( + lighthouse=_Lighthouse() + ), + ) + + result = adapters.get_lighthouse_messages( + profile="demo", + base_url="https://api-demo.ksef.mf.gov.pl", + lighthouse_base_url=None, + ) + + assert result["count"] == 1 + assert result["items"][0]["id"] == "m-1" + + def test_download_invoice_xml_success(monkeypatch, tmp_path) -> None: class _Invoices: def get_invoice(self, ksef_number, access_token): diff --git a/tests/test_client.py b/tests/test_client.py index f3076a3..f2899e9 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -9,22 +9,36 @@ class ClientTests(unittest.TestCase): def test_sync_client_context(self): options = KsefClientOptions(base_url="https://api-test.ksef.mf.gov.pl") client = KsefClient(options) - with patch.object(client._http, "close", Mock()) as close_mock: + with ( + patch.object(client._http, "close", Mock()) as close_mock, + patch.object(client._lighthouse_http, "close", Mock()) as lighthouse_close_mock, + ): with client as ctx: self.assertIs(ctx, client) self.assertIsNotNone(client.http_client) + self.assertIsNotNone(client.lighthouse) close_mock.assert_called_once() + lighthouse_close_mock.assert_called_once() class AsyncClientTests(unittest.IsolatedAsyncioTestCase): async def test_async_client_context(self): options = KsefClientOptions(base_url="https://api-test.ksef.mf.gov.pl") client = AsyncKsefClient(options) - with patch.object(client._http, "aclose", AsyncMock()) as aclose_mock: + with ( + patch.object(client._http, "aclose", AsyncMock()) as aclose_mock, + patch.object( + client._lighthouse_http, + "aclose", + AsyncMock(), + ) as lighthouse_aclose_mock, + ): async with client as ctx: self.assertIs(ctx, client) self.assertIsNotNone(client.http_client) + self.assertIsNotNone(client.lighthouse) aclose_mock.assert_called_once() + lighthouse_aclose_mock.assert_called_once() if __name__ == "__main__": diff --git a/tests/test_clients.py b/tests/test_clients.py index b75cfa1..df7ec06 100644 --- a/tests/test_clients.py +++ b/tests/test_clients.py @@ -11,6 +11,7 @@ InvoicesClient, _normalize_datetime_without_offset, ) +from ksef_client.clients.lighthouse import AsyncLighthouseClient, LighthouseClient from ksef_client.clients.limits import AsyncLimitsClient, LimitsClient from ksef_client.clients.peppol import AsyncPeppolClient, PeppolClient from ksef_client.clients.permissions import AsyncPermissionsClient, PermissionsClient @@ -288,6 +289,34 @@ def test_peppol_client(self): with patch.object(client, "_request_json", Mock(return_value={"ok": True})): client.list_providers(page_offset=0, page_size=10) + def test_lighthouse_client(self): + client = LighthouseClient(self.http, "https://api-latarnia-test.ksef.mf.gov.pl") + with patch.object( + client, + "_request_json", + Mock( + side_effect=[ + { + "status": "AVAILABLE", + "messages": [], + }, + [], + ] + ), + ) as request_json_mock: + status = client.get_status() + messages = client.get_messages() + self.assertEqual(status.status.value, "AVAILABLE") + self.assertEqual(messages, []) + self.assertEqual( + request_json_mock.call_args_list[0].args[1], + "https://api-latarnia-test.ksef.mf.gov.pl/status", + ) + self.assertEqual( + request_json_mock.call_args_list[1].args[1], + "https://api-latarnia-test.ksef.mf.gov.pl/messages", + ) + class AsyncClientsTests(unittest.IsolatedAsyncioTestCase): async def test_async_clients(self): @@ -557,6 +586,25 @@ async def test_async_clients(self): with patch.object(peppol, "_request_json", AsyncMock(return_value={"ok": True})): await peppol.list_providers(page_offset=0, page_size=10) + lighthouse = AsyncLighthouseClient(http, "https://api-latarnia-test.ksef.mf.gov.pl") + with patch.object( + lighthouse, + "_request_json", + AsyncMock(side_effect=[{"status": "AVAILABLE", "messages": []}, []]), + ) as request_json_mock: + status = await lighthouse.get_status() + messages = await lighthouse.get_messages() + self.assertEqual(status.status.value, "AVAILABLE") + self.assertEqual(messages, []) + self.assertEqual( + request_json_mock.call_args_list[0].args[1], + "https://api-latarnia-test.ksef.mf.gov.pl/status", + ) + self.assertEqual( + request_json_mock.call_args_list[1].args[1], + "https://api-latarnia-test.ksef.mf.gov.pl/messages", + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py index 064859e..6d4091b 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,7 +1,12 @@ import unittest from unittest.mock import patch -from ksef_client.config import KsefClientOptions, KsefEnvironment, _package_version +from ksef_client.config import ( + KsefClientOptions, + KsefEnvironment, + KsefLighthouseEnvironment, + _package_version, +) class ConfigTests(unittest.TestCase): @@ -39,6 +44,41 @@ def test_resolve_qr_base_url(self): with self.assertRaises(ValueError): options.resolve_qr_base_url() + def test_resolve_lighthouse_base_url(self): + options = KsefClientOptions( + base_url="https://api-test.ksef.mf.gov.pl", + base_lighthouse_url="https://example.com/lh/", + ) + self.assertEqual(options.resolve_lighthouse_base_url(), "https://example.com/lh") + + options = KsefClientOptions(base_url=KsefEnvironment.TEST.value) + self.assertEqual( + options.resolve_lighthouse_base_url(), + KsefLighthouseEnvironment.TEST.value, + ) + + options = KsefClientOptions(base_url=KsefEnvironment.DEMO.value) + self.assertEqual( + options.resolve_lighthouse_base_url(), + KsefLighthouseEnvironment.TEST.value, + ) + + options = KsefClientOptions(base_url=KsefEnvironment.PROD.value) + self.assertEqual( + options.resolve_lighthouse_base_url(), + KsefLighthouseEnvironment.PROD.value, + ) + + options = KsefClientOptions(base_url=KsefLighthouseEnvironment.PRD.value) + self.assertEqual( + options.resolve_lighthouse_base_url(), + KsefLighthouseEnvironment.PROD.value, + ) + + options = KsefClientOptions(base_url="https://example.com") + with self.assertRaises(ValueError): + options.resolve_lighthouse_base_url() + if __name__ == "__main__": unittest.main() diff --git a/tests/test_lighthouse_openapi_coverage.py b/tests/test_lighthouse_openapi_coverage.py new file mode 100644 index 0000000..1157c74 --- /dev/null +++ b/tests/test_lighthouse_openapi_coverage.py @@ -0,0 +1,102 @@ +import ast +import json +import re +import unittest +from pathlib import Path + + +def _normalize_path(path: str) -> str: + return re.sub(r"\{[^}]+\}", "{}", path) + + +def _const_str(node: ast.AST) -> str | None: + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + return None + + +def _render_path(node: ast.AST) -> str | None: + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + if isinstance(node, ast.JoinedStr): + parts: list[str] = [] + for value in node.values: + if isinstance(value, ast.Constant) and isinstance(value.value, str): + parts.append(value.value) + else: + parts.append("{}") + return "".join(parts) + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): + left = _render_path(node.left) + right = _render_path(node.right) + if left is None or right is None: + return None + return left + right + return None + + +def _extract_path_suffix(path: str) -> str | None: + if path.startswith("/"): + return path + match = re.search(r"(/[^/?#]+)$", path) + if not match: + return None + return match.group(1) + + +def _extract_python_lighthouse_operations() -> set[tuple[str, str]]: + project_root = Path(__file__).resolve().parents[1] + py_file = project_root / "src" / "ksef_client" / "clients" / "lighthouse.py" + ops: set[tuple[str, str]] = set() + + tree = ast.parse(py_file.read_text(encoding="utf-8")) + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Attribute): + continue + if node.func.attr != "_request_json": + continue + if len(node.args) < 2: + continue + + method = _const_str(node.args[0]) + path_expr = _render_path(node.args[1]) + if not method or not path_expr: + continue + suffix = _extract_path_suffix(path_expr) + if not suffix: + continue + ops.add((method.upper(), _normalize_path(suffix))) + + return ops + + +def _extract_openapi_lighthouse_operations(openapi_path: Path) -> set[tuple[str, str]]: + spec = json.loads(openapi_path.read_text(encoding="utf-8")) + ops: set[tuple[str, str]] = set() + for path, methods in spec["paths"].items(): + for method in methods: + method_upper = method.upper() + if method_upper in {"GET", "POST", "PUT", "PATCH", "DELETE"}: + ops.add((method_upper, _normalize_path(path))) + return ops + + +class LighthouseOpenApiCoverageTests(unittest.TestCase): + def test_python_lighthouse_client_covers_openapi_spec(self) -> None: + repo_root = Path(__file__).resolve().parents[2] + openapi_path = repo_root / "ksef-latarnia" / "open-api.json" + if not openapi_path.exists(): + self.skipTest("ksef-latarnia/open-api.json not found; coverage test requires monorepo") + + spec_ops = _extract_openapi_lighthouse_operations(openapi_path) + py_ops = _extract_python_lighthouse_operations() + + missing = sorted(spec_ops - py_ops) + extra = sorted(py_ops - spec_ops) + + self.assertFalse(missing, f"Missing Lighthouse OpenAPI operations: {missing}") + self.assertFalse(extra, f"Extra Lighthouse operations not in spec: {extra}") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_models.py b/tests/test_models.py index 55f4425..b9740e2 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -76,6 +76,58 @@ def test_part_upload_request(self): parsed = models.PartUploadRequest.from_dict(data) self.assertEqual(parsed.headers["x"], "y") + def test_lighthouse_message_and_status(self): + message_data = { + "id": "m-1", + "eventId": 10, + "category": "MAINTENANCE", + "type": "MAINTENANCE_ANNOUNCEMENT", + "title": "T", + "text": "X", + "start": "2026-03-15T01:00:00Z", + "end": "2026-03-15T06:00:00Z", + "version": 1, + "published": "2026-03-10T10:00:00Z", + } + message = models.LighthouseMessage.from_dict(message_data) + self.assertEqual(message.category, models.LighthouseMessageCategory.MAINTENANCE) + self.assertEqual(message.type, models.LighthouseMessageType.MAINTENANCE_ANNOUNCEMENT) + self.assertEqual(message.to_dict()["eventId"], 10) + + status = models.LighthouseStatusResponse.from_dict( + { + "status": "MAINTENANCE", + "messages": [message_data], + } + ) + self.assertEqual(status.status, models.LighthouseKsefStatus.MAINTENANCE) + self.assertIsNotNone(status.messages) + assert status.messages is not None + self.assertEqual(status.messages[0].id, "m-1") + self.assertEqual(status.to_dict()["status"], "MAINTENANCE") + + def test_lighthouse_enum_fallbacks(self): + message = models.LighthouseMessage.from_dict( + { + "id": "m-2", + "eventId": 11, + "category": "UNKNOWN", + "type": "UNKNOWN", + "title": "", + "text": "", + "start": "", + "end": None, + "version": 0, + "published": "", + } + ) + self.assertEqual(message.category, models.LighthouseMessageCategory.FAILURE) + self.assertEqual(message.type, models.LighthouseMessageType.FAILURE_START) + + status = models.LighthouseStatusResponse.from_dict({"status": "UNKNOWN"}) + self.assertEqual(status.status, models.LighthouseKsefStatus.AVAILABLE) + self.assertIsNone(status.messages) + if __name__ == "__main__": unittest.main() From bff593d940c68d647c68b0808ba05335ac430a88 Mon Sep 17 00:00:00 2001 From: smkc Date: Wed, 25 Feb 2026 22:55:45 +0100 Subject: [PATCH 2/3] fix(lighthouse): allow no-profile CLI fallback and harden openapi parity path parsing --- README.md | 1 + docs/api/lighthouse.md | 6 +++ docs/cli/README.md | 4 +- src/ksef_client/cli/auth/manager.py | 22 +++++++- .../cli/commands/lighthouse_cmd.py | 26 ++++++---- tests/cli/integration/test_lighthouse.py | 51 +++++++++++++++++++ tests/cli/unit/test_auth_manager.py | 34 +++++++++++++ tests/test_lighthouse_openapi_coverage.py | 41 +++++++++++---- 8 files changed, 164 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index 58a71d0..259d6d4 100644 --- a/README.md +++ b/README.md @@ -100,6 +100,7 @@ Najważniejsze grupy komend: - operacje: `invoice ...`, `send ...`, `upo ...`, `export ...` - diagnostyka: `health check` - latarnia: `lighthouse status`, `lighthouse messages` + - komendy Latarni sa publiczne (dzialaja bez logowania; bez profilu domyslnie uzywana jest latarnia test) Pełna specyfikacja CLI: [`docs/cli/README.md`](docs/cli/README.md) diff --git a/docs/api/lighthouse.md b/docs/api/lighthouse.md index 604cf7f..4a0bf65 100644 --- a/docs/api/lighthouse.md +++ b/docs/api/lighthouse.md @@ -25,3 +25,9 @@ Domyślne mapowanie dla `KsefClientOptions.base_url`: Nadpisanie mapowania: - `KsefClientOptions(base_lighthouse_url="https://...")` + +## CLI (`ksef lighthouse ...`) + +- Komendy Latarni działają bez tokenu i bez aktywnego profilu. +- Jeśli nie ustawisz profilu i nie podasz `--base-url`/`KSEF_LIGHTHOUSE_BASE_URL`, CLI używa domyślnie **latarni test**: + - `https://api-latarnia-test.ksef.mf.gov.pl` diff --git a/docs/cli/README.md b/docs/cli/README.md index f73fa6c..c1de9a6 100644 --- a/docs/cli/README.md +++ b/docs/cli/README.md @@ -123,7 +123,8 @@ Zachowanie profilu: - gdy nie podasz `--base-url`, CLI bierze kolejno: CLI option -> `KSEF_BASE_URL` -> `profile.base_url` -> DEMO. Brak aktywnego profilu: -- komendy `auth`, `health`, `lighthouse`, `invoice`, `send`, `upo`, `export` wymagaja aktywnego profilu, +- komendy `auth`, `health`, `invoice`, `send`, `upo`, `export` wymagaja aktywnego profilu, +- komendy `lighthouse ...` dzialaja bez profilu (publiczne API Latarni), - jesli profil nie jest ustawiony, CLI zwraca czytelny blad z podpowiedzia: - `ksef init --set-active` - `ksef profile use --name ` @@ -310,6 +311,7 @@ Options: Uwagi: - komenda korzysta z publicznego API Latarni (bez tokenu dostepowego), - domyslnie base URL Latarni mapowany jest z profilu KSeF (`TEST`/`DEMO` -> latarnia TEST, `PROD` -> latarnia PROD), +- gdy brak profilu i brak override (`--base-url`/`KSEF_LIGHTHOUSE_BASE_URL`), uzywana jest domyslnie **latarnia test**: `https://api-latarnia-test.ksef.mf.gov.pl`, - `--base-url` lub `KSEF_LIGHTHOUSE_BASE_URL` pozwala nadpisac endpoint Latarni. ## `ksef lighthouse messages` diff --git a/src/ksef_client/cli/auth/manager.py b/src/ksef_client/cli/auth/manager.py index eac4043..6813ee2 100644 --- a/src/ksef_client/cli/auth/manager.py +++ b/src/ksef_client/cli/auth/manager.py @@ -3,7 +3,7 @@ from pathlib import Path from typing import Any -from ksef_client.config import KsefEnvironment +from ksef_client.config import KsefClientOptions, KsefEnvironment, KsefLighthouseEnvironment from ksef_client.services.workflows import AuthCoordinator from ksef_client.services.xades import XadesKeyPair @@ -51,6 +51,26 @@ def resolve_base_url(base_url: str | None, *, profile: str | None = None) -> str return KsefEnvironment.DEMO.value +def resolve_lighthouse_base_url( + base_lighthouse_url: str | None, + *, + profile: str | None = None, +) -> str: + if base_lighthouse_url and base_lighthouse_url.strip(): + return base_lighthouse_url.strip() + + if profile: + profile_base_url, _, _ = _profile_context(profile) + if profile_base_url and profile_base_url.strip(): + try: + options = KsefClientOptions(base_url=profile_base_url.strip()) + return options.resolve_lighthouse_base_url() + except ValueError: + pass + + return KsefLighthouseEnvironment.TEST.value + + def login_with_token( *, profile: str, diff --git a/src/ksef_client/cli/commands/lighthouse_cmd.py b/src/ksef_client/cli/commands/lighthouse_cmd.py index b86ad4f..d4846fb 100644 --- a/src/ksef_client/cli/commands/lighthouse_cmd.py +++ b/src/ksef_client/cli/commands/lighthouse_cmd.py @@ -6,8 +6,8 @@ from ksef_client.exceptions import KsefApiError, KsefHttpError, KsefRateLimitError -from ..auth.manager import resolve_base_url -from ..context import profile_label, require_context, require_profile +from ..auth.manager import resolve_base_url, resolve_lighthouse_base_url +from ..context import profile_label, require_context from ..errors import CliError from ..exit_codes import ExitCode from ..output import get_renderer @@ -71,12 +71,15 @@ def lighthouse_status( cli_ctx = require_context(ctx) renderer = get_renderer(cli_ctx) profile = profile_label(cli_ctx) + selected_profile = cli_ctx.profile try: - profile = require_profile(cli_ctx) result = get_lighthouse_status( - profile=profile, - base_url=resolve_base_url(os.getenv("KSEF_BASE_URL"), profile=profile), - lighthouse_base_url=base_url or os.getenv("KSEF_LIGHTHOUSE_BASE_URL"), + profile=selected_profile or "", + base_url=resolve_base_url(os.getenv("KSEF_BASE_URL"), profile=selected_profile), + lighthouse_base_url=resolve_lighthouse_base_url( + base_url or os.getenv("KSEF_LIGHTHOUSE_BASE_URL"), + profile=selected_profile, + ), ) except Exception as exc: _render_error(ctx, "lighthouse.status", exc) @@ -93,12 +96,15 @@ def lighthouse_messages( cli_ctx = require_context(ctx) renderer = get_renderer(cli_ctx) profile = profile_label(cli_ctx) + selected_profile = cli_ctx.profile try: - profile = require_profile(cli_ctx) result = get_lighthouse_messages( - profile=profile, - base_url=resolve_base_url(os.getenv("KSEF_BASE_URL"), profile=profile), - lighthouse_base_url=base_url or os.getenv("KSEF_LIGHTHOUSE_BASE_URL"), + profile=selected_profile or "", + base_url=resolve_base_url(os.getenv("KSEF_BASE_URL"), profile=selected_profile), + lighthouse_base_url=resolve_lighthouse_base_url( + base_url or os.getenv("KSEF_LIGHTHOUSE_BASE_URL"), + profile=selected_profile, + ), ) except Exception as exc: _render_error(ctx, "lighthouse.messages", exc) diff --git a/tests/cli/integration/test_lighthouse.py b/tests/cli/integration/test_lighthouse.py index c772ab6..c5c0a46 100644 --- a/tests/cli/integration/test_lighthouse.py +++ b/tests/cli/integration/test_lighthouse.py @@ -1,9 +1,11 @@ from __future__ import annotations import json +from pathlib import Path from ksef_client.cli.app import app from ksef_client.cli.commands import lighthouse_cmd +from ksef_client.cli.config import paths from ksef_client.cli.errors import CliError from ksef_client.cli.exit_codes import ExitCode @@ -64,3 +66,52 @@ def test_lighthouse_status_validation_error_exit_code(runner, monkeypatch) -> No ) result = runner.invoke(app, ["lighthouse", "status"]) assert result.exit_code == int(ExitCode.VALIDATION_ERROR) + + +def test_lighthouse_status_works_without_active_profile(runner, monkeypatch, tmp_path) -> None: + _write_unconfigured_config(monkeypatch, tmp_path) + seen: dict[str, object] = {} + + def _fake_status(**kwargs): + seen.update(kwargs) + return {"status": "AVAILABLE", "messages": []} + + monkeypatch.setattr(lighthouse_cmd, "get_lighthouse_status", _fake_status) + result = runner.invoke(app, ["lighthouse", "status"]) + assert result.exit_code == 0 + assert seen["profile"] == "" + assert seen["lighthouse_base_url"] == "https://api-latarnia-test.ksef.mf.gov.pl" + assert "" in result.stdout + + +def test_lighthouse_status_without_profile_prefers_env_base_url( + runner, monkeypatch, tmp_path +) -> None: + _write_unconfigured_config(monkeypatch, tmp_path) + seen: dict[str, object] = {} + + def _fake_status(**kwargs): + seen.update(kwargs) + return {"status": "AVAILABLE", "messages": []} + + monkeypatch.setattr(lighthouse_cmd, "get_lighthouse_status", _fake_status) + monkeypatch.setenv("KSEF_LIGHTHOUSE_BASE_URL", "https://api-latarnia.ksef.mf.gov.pl") + result = runner.invoke(app, ["lighthouse", "status"]) + assert result.exit_code == 0 + assert seen["lighthouse_base_url"] == "https://api-latarnia.ksef.mf.gov.pl" + + +def _write_unconfigured_config(monkeypatch, tmp_path: Path) -> Path: + config_path = tmp_path / "config.json" + config_path.write_text( + """ +{ + "version": 1, + "active_profile": null, + "profiles": {} +} +""".strip(), + encoding="utf-8", + ) + monkeypatch.setattr(paths, "config_file", lambda: config_path) + return config_path diff --git a/tests/cli/unit/test_auth_manager.py b/tests/cli/unit/test_auth_manager.py index 7866697..374f4cb 100644 --- a/tests/cli/unit/test_auth_manager.py +++ b/tests/cli/unit/test_auth_manager.py @@ -9,6 +9,7 @@ from ksef_client.cli.config.schema import CliConfig, ProfileConfig from ksef_client.cli.errors import CliError from ksef_client.cli.exit_codes import ExitCode +from ksef_client.config import KsefLighthouseEnvironment class _FakeClient: @@ -213,6 +214,39 @@ def test_resolve_base_url_uses_profile_when_missing(monkeypatch) -> None: assert manager.resolve_base_url(None, profile="demo") == "https://profile.example" +def test_resolve_lighthouse_base_url_prefers_explicit_value() -> None: + assert manager.resolve_lighthouse_base_url(" https://api-latarnia-test.ksef.mf.gov.pl/ ") == ( + "https://api-latarnia-test.ksef.mf.gov.pl/" + ).strip() + + +def test_resolve_lighthouse_base_url_uses_profile_mapping(monkeypatch) -> None: + monkeypatch.setattr( + manager, + "load_config", + lambda: CliConfig( + active_profile="demo", + profiles={ + "demo": ProfileConfig( + name="demo", + env="DEMO", + base_url="https://api-demo.ksef.mf.gov.pl", + context_type="nip", + context_value="123", + ) + }, + ), + ) + assert manager.resolve_lighthouse_base_url(None, profile="demo") == ( + KsefLighthouseEnvironment.TEST.value + ) + + +def test_resolve_lighthouse_base_url_fallbacks_to_lighthouse_test(monkeypatch) -> None: + monkeypatch.setattr(manager, "load_config", lambda: CliConfig()) + assert manager.resolve_lighthouse_base_url(None) == KsefLighthouseEnvironment.TEST.value + + def test_refresh_access_token_missing_token_in_response(monkeypatch) -> None: class _FakeClientNoToken: def __init__(self) -> None: diff --git a/tests/test_lighthouse_openapi_coverage.py b/tests/test_lighthouse_openapi_coverage.py index 1157c74..a628e1f 100644 --- a/tests/test_lighthouse_openapi_coverage.py +++ b/tests/test_lighthouse_openapi_coverage.py @@ -3,6 +3,7 @@ import re import unittest from pathlib import Path +from urllib.parse import urlparse def _normalize_path(path: str) -> str: @@ -35,13 +36,19 @@ def _render_path(node: ast.AST) -> str | None: return None -def _extract_path_suffix(path: str) -> str | None: - if path.startswith("/"): - return path - match = re.search(r"(/[^/?#]+)$", path) - if not match: +def _normalize_request_path(path: str) -> str | None: + stripped = path.strip() + if not stripped: return None - return match.group(1) + if stripped.startswith("http://") or stripped.startswith("https://"): + parsed = urlparse(stripped) + return parsed.path or "/" + if stripped.startswith("/"): + return stripped.split("?", 1)[0].split("#", 1)[0] + slash_index = stripped.find("/") + if slash_index >= 0: + return stripped[slash_index:].split("?", 1)[0].split("#", 1)[0] + return None def _extract_python_lighthouse_operations() -> set[tuple[str, str]]: @@ -62,10 +69,10 @@ def _extract_python_lighthouse_operations() -> set[tuple[str, str]]: path_expr = _render_path(node.args[1]) if not method or not path_expr: continue - suffix = _extract_path_suffix(path_expr) - if not suffix: + normalized_path = _normalize_request_path(path_expr) + if not normalized_path: continue - ops.add((method.upper(), _normalize_path(suffix))) + ops.add((method.upper(), _normalize_path(normalized_path))) return ops @@ -82,6 +89,22 @@ def _extract_openapi_lighthouse_operations(openapi_path: Path) -> set[tuple[str, class LighthouseOpenApiCoverageTests(unittest.TestCase): + def test_normalize_request_path(self) -> None: + self.assertEqual( + _normalize_request_path("https://api-latarnia-test.ksef.mf.gov.pl/messages"), + "/messages", + ) + self.assertEqual( + _normalize_request_path("https://example.com/api/v1/lighthouse/status?x=1"), + "/api/v1/lighthouse/status", + ) + self.assertEqual( + _normalize_request_path("/status?expand=true"), + "/status", + ) + self.assertEqual(_normalize_request_path("{}/messages"), "/messages") + self.assertIsNone(_normalize_request_path("status")) + def test_python_lighthouse_client_covers_openapi_spec(self) -> None: repo_root = Path(__file__).resolve().parents[2] openapi_path = repo_root / "ksef-latarnia" / "open-api.json" From d72f2b3657b10380d959dc13bcaaf9ce47659fe3 Mon Sep 17 00:00:00 2001 From: smkc Date: Wed, 25 Feb 2026 23:19:20 +0100 Subject: [PATCH 3/3] fix(ci): restore 100% coverage and ignore lighthouse in main openapi check --- tests/cli/integration/test_lighthouse.py | 43 ++++++++++++++++++++++++ tests/cli/unit/test_auth_manager.py | 22 ++++++++++++ tests/test_client.py | 18 ++++++++++ tests/test_clients.py | 30 +++++++++++++++++ tests/test_config.py | 6 ++++ tools/check_coverage.py | 19 +++++++++-- 6 files changed, 136 insertions(+), 2 deletions(-) diff --git a/tests/cli/integration/test_lighthouse.py b/tests/cli/integration/test_lighthouse.py index c5c0a46..88f8d7b 100644 --- a/tests/cli/integration/test_lighthouse.py +++ b/tests/cli/integration/test_lighthouse.py @@ -8,6 +8,7 @@ from ksef_client.cli.config import paths from ksef_client.cli.errors import CliError from ksef_client.cli.exit_codes import ExitCode +from ksef_client.exceptions import KsefHttpError, KsefRateLimitError def _json_output(text: str) -> dict: @@ -68,6 +69,38 @@ def test_lighthouse_status_validation_error_exit_code(runner, monkeypatch) -> No assert result.exit_code == int(ExitCode.VALIDATION_ERROR) +def test_lighthouse_status_rate_limit_error_exit_code(runner, monkeypatch) -> None: + monkeypatch.setattr( + lighthouse_cmd, + "get_lighthouse_status", + lambda **kwargs: (_ for _ in ()).throw( + KsefRateLimitError(status_code=429, message="Too Many Requests") + ), + ) + result = runner.invoke(app, ["lighthouse", "status"]) + assert result.exit_code == int(ExitCode.RETRY_EXHAUSTED) + + +def test_lighthouse_status_api_error_exit_code(runner, monkeypatch) -> None: + monkeypatch.setattr( + lighthouse_cmd, + "get_lighthouse_status", + lambda **kwargs: (_ for _ in ()).throw(KsefHttpError(status_code=500, message="boom")), + ) + result = runner.invoke(app, ["lighthouse", "status"]) + assert result.exit_code == int(ExitCode.API_ERROR) + + +def test_lighthouse_status_unexpected_error_exit_code(runner, monkeypatch) -> None: + monkeypatch.setattr( + lighthouse_cmd, + "get_lighthouse_status", + lambda **kwargs: (_ for _ in ()).throw(RuntimeError("unexpected")), + ) + result = runner.invoke(app, ["lighthouse", "status"]) + assert result.exit_code == int(ExitCode.CONFIG_ERROR) + + def test_lighthouse_status_works_without_active_profile(runner, monkeypatch, tmp_path) -> None: _write_unconfigured_config(monkeypatch, tmp_path) seen: dict[str, object] = {} @@ -101,6 +134,16 @@ def _fake_status(**kwargs): assert seen["lighthouse_base_url"] == "https://api-latarnia.ksef.mf.gov.pl" +def test_lighthouse_messages_unexpected_error_exit_code(runner, monkeypatch) -> None: + monkeypatch.setattr( + lighthouse_cmd, + "get_lighthouse_messages", + lambda **kwargs: (_ for _ in ()).throw(RuntimeError("unexpected")), + ) + result = runner.invoke(app, ["lighthouse", "messages"]) + assert result.exit_code == int(ExitCode.CONFIG_ERROR) + + def _write_unconfigured_config(monkeypatch, tmp_path: Path) -> Path: config_path = tmp_path / "config.json" config_path.write_text( diff --git a/tests/cli/unit/test_auth_manager.py b/tests/cli/unit/test_auth_manager.py index 374f4cb..354f07e 100644 --- a/tests/cli/unit/test_auth_manager.py +++ b/tests/cli/unit/test_auth_manager.py @@ -247,6 +247,28 @@ def test_resolve_lighthouse_base_url_fallbacks_to_lighthouse_test(monkeypatch) - assert manager.resolve_lighthouse_base_url(None) == KsefLighthouseEnvironment.TEST.value +def test_resolve_lighthouse_base_url_invalid_profile_base_fallback(monkeypatch) -> None: + monkeypatch.setattr( + manager, + "load_config", + lambda: CliConfig( + active_profile="demo", + profiles={ + "demo": ProfileConfig( + name="demo", + env="DEMO", + base_url="https://unknown.example", + context_type="nip", + context_value="123", + ) + }, + ), + ) + assert manager.resolve_lighthouse_base_url(None, profile="demo") == ( + KsefLighthouseEnvironment.TEST.value + ) + + def test_refresh_access_token_missing_token_in_response(monkeypatch) -> None: class _FakeClientNoToken: def __init__(self) -> None: diff --git a/tests/test_client.py b/tests/test_client.py index f2899e9..f04e02d 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -20,6 +20,15 @@ def test_sync_client_context(self): close_mock.assert_called_once() lighthouse_close_mock.assert_called_once() + def test_sync_client_unknown_lighthouse_mapping_fallbacks_to_empty_base(self): + options = KsefClientOptions(base_url="https://unknown.example") + client = KsefClient(options) + with patch.object(client._http, "close", Mock()), patch.object( + client._lighthouse_http, "close", Mock() + ): + self.assertEqual(client.lighthouse._base_url, "") + client.close() + class AsyncClientTests(unittest.IsolatedAsyncioTestCase): async def test_async_client_context(self): @@ -40,6 +49,15 @@ async def test_async_client_context(self): aclose_mock.assert_called_once() lighthouse_aclose_mock.assert_called_once() + async def test_async_client_unknown_lighthouse_mapping_fallbacks_to_empty_base(self): + options = KsefClientOptions(base_url="https://unknown.example") + client = AsyncKsefClient(options) + with patch.object(client._http, "aclose", AsyncMock()), patch.object( + client._lighthouse_http, "aclose", AsyncMock() + ): + self.assertEqual(client.lighthouse._base_url, "") + await client.aclose() + if __name__ == "__main__": unittest.main() diff --git a/tests/test_clients.py b/tests/test_clients.py index df7ec06..54874ac 100644 --- a/tests/test_clients.py +++ b/tests/test_clients.py @@ -317,6 +317,18 @@ def test_lighthouse_client(self): "https://api-latarnia-test.ksef.mf.gov.pl/messages", ) + def test_lighthouse_client_handles_invalid_payload_and_missing_base_url(self): + client = LighthouseClient(self.http, "https://api-latarnia-test.ksef.mf.gov.pl") + with patch.object(client, "_request_json", Mock(side_effect=[None, {"unexpected": True}])): + status = client.get_status() + messages = client.get_messages() + self.assertEqual(status.status.value, "AVAILABLE") + self.assertEqual(messages, []) + + missing_base_client = LighthouseClient(self.http, "") + with self.assertRaises(ValueError): + missing_base_client.get_status() + class AsyncClientsTests(unittest.IsolatedAsyncioTestCase): async def test_async_clients(self): @@ -605,6 +617,24 @@ async def test_async_clients(self): "https://api-latarnia-test.ksef.mf.gov.pl/messages", ) + async def test_async_lighthouse_handles_invalid_payload_and_missing_base_url(self): + response = HttpResponse(200, httpx.Headers({}), b"{}") + http = DummyAsyncHttp(response) + lighthouse = AsyncLighthouseClient(http, "https://api-latarnia-test.ksef.mf.gov.pl") + with patch.object( + lighthouse, + "_request_json", + AsyncMock(side_effect=[None, {"unexpected": True}]), + ): + status = await lighthouse.get_status() + messages = await lighthouse.get_messages() + self.assertEqual(status.status.value, "AVAILABLE") + self.assertEqual(messages, []) + + missing_base = AsyncLighthouseClient(http, "") + with self.assertRaises(ValueError): + await missing_base.get_messages() + if __name__ == "__main__": unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py index 6d4091b..4c3679f 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -75,6 +75,12 @@ def test_resolve_lighthouse_base_url(self): KsefLighthouseEnvironment.PROD.value, ) + options = KsefClientOptions(base_url=KsefLighthouseEnvironment.TEST.value) + self.assertEqual( + options.resolve_lighthouse_base_url(), + KsefLighthouseEnvironment.TEST.value, + ) + options = KsefClientOptions(base_url="https://example.com") with self.assertRaises(ValueError): options.resolve_lighthouse_base_url() diff --git a/tools/check_coverage.py b/tools/check_coverage.py index 780404c..53dbf20 100644 --- a/tools/check_coverage.py +++ b/tools/check_coverage.py @@ -221,9 +221,15 @@ def _extract_expected_status(self, node: ast.Call) -> set[int]: return set() -def get_implemented_endpoints_deep(source_dir: Path) -> list[ImplementedEndpoint]: +def get_implemented_endpoints_deep( + source_dir: Path, + excluded_file_names: set[str] | None = None, +) -> list[ImplementedEndpoint]: endpoints = [] + excluded = excluded_file_names or set() for py_file in source_dir.rglob("*.py"): + if py_file.name in excluded: + continue try: visitor = AdvancedClientVisitor(str(py_file)) tree = ast.parse(py_file.read_text(encoding="utf-8")) @@ -249,13 +255,22 @@ def main(): parser.add_argument( "--src", required=True, type=Path, help="Path to source directory containing clients" ) + parser.add_argument( + "--exclude-files", + nargs="*", + default=["lighthouse.py"], + help="Client file names to exclude from this OpenAPI coverage check.", + ) args = parser.parse_args() # 1. Load OpenAPI Specs openapi_specs = get_openapi_specs(args.openapi) # 2. Analyze Code - implemented_eps = get_implemented_endpoints_deep(args.src) + implemented_eps = get_implemented_endpoints_deep( + args.src, + excluded_file_names=set(args.exclude_files), + ) # 3. Compare openapi_keys = set(openapi_specs.keys())