fix(ci): restore 100% coverage and ignore lighthouse in main openapi check

Author: smkc

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(

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:

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()

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()

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()

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())