Use python logging facilities in openapi

Author: mdellweg

CHANGES/pulp-glue/+debug_logging.feature

@@ -0,0 +1 @@
+Replaced `debug_callback` with python logging facilities.

pulp-glue/pulp_glue/common/openapi.py

@@ -2,6 +2,7 @@
 # GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
 
 import json
+import logging
 import os
 import typing as t
 import warnings
@@ -29,6 +30,7 @@
 
 translation = get_translation(__package__)
 _ = translation.gettext
+_logger = logging.getLogger("pulp_glue.openapi")
 
 UploadType = t.Union[bytes, t.IO[bytes]]
 
@@ -176,13 +178,22 @@ def __init__(
         safe_calls_only: t.Optional[bool] = None,
     ):
         if validate_certs is not None:
-            warnings.warn("validate_certs is deprecated; use verify_ssl instead.")
+            warnings.warn(
+                "validate_certs is deprecated; use verify_ssl instead.", DeprecationWarning
+            )
             verify_ssl = validate_certs
         if safe_calls_only is not None:
-            warnings.warn("safe_calls_only is deprecated; use dry_run instead.")
+            warnings.warn("safe_calls_only is deprecated; use dry_run instead.", DeprecationWarning)
             dry_run = safe_calls_only
+        if debug_callback is not None:
+            warnings.warn(
+                "debug_callback is deprecated; use logging with level DEBUG instead.",
+                DeprecationWarning,
+            )
 
-        self._debug_callback: t.Callable[[int, str], t.Any] = debug_callback or (lambda i, x: None)
+        self._debug_callback: t.Callable[[int, str], t.Any] = debug_callback or (
+            lambda lvl, msg: _logger.log(logging.DEBUG + 4 - lvl, msg)
+        )
         self._base_url: str = base_url
         self._doc_path: str = doc_path
         self._dry_run: bool = dry_run
@@ -635,9 +646,10 @@ def call(
 
         if query_params:
             qs = urlencode(query_params)
-            self._debug_callback(1, f"{operation_id} : {method} {url}?{qs}")
+            log_msg = f"{operation_id} : {method} {url}?{qs}"
         else:
-            self._debug_callback(1, f"{operation_id} : {method} {url}")
+            log_msg = f"{operation_id} : {method} {url}"
+        self._debug_callback(1, log_msg)
         self._debug_callback(
             2, "\n".join([f"  {key}=={value}" for key, value in query_params.items()])
         )

pulp-glue/tests/test_openapi_logging.py

@@ -0,0 +1,67 @@
+import json
+import logging
+import typing as t
+
+import pytest
+
+from pulp_glue.common.openapi import OpenAPI
+
+TEST_SCHEMA = json.dumps(
+    {
+        "openapi": "3.0.3",
+        "paths": {"test/": {"get": {"operationId": "test_id", "responses": {200: {}}}}},
+        "components": {"schemas": {}},
+    }
+).encode()
+
+
+class MockRequest:
+    headers: t.Dict[str, str] = {}
+    body: t.Dict[str, t.Any] = {}
+
+
+class MockResponse:
+    status_code = 200
+    headers: t.Dict[str, str] = {}
+    text = "{}"
+    content: t.Dict[str, t.Any] = {}
+
+    def raise_for_status(self) -> None:
+        pass
+
+
+class MockSession:
+    def prepare_request(self, *args: t.Any, **kwargs: t.Any) -> MockRequest:
+        return MockRequest()
+
+    def send(self, request: MockRequest) -> MockResponse:
+        return MockResponse()
+
+
+@pytest.fixture
+def openapi(monkeypatch: pytest.MonkeyPatch) -> OpenAPI:
+    monkeypatch.setattr(OpenAPI, "load_api", lambda self, refresh_cache: TEST_SCHEMA)
+    openapi = OpenAPI("base_url", "doc_path")
+    openapi._parse_api(TEST_SCHEMA)
+    monkeypatch.setattr(openapi, "_session", MockSession())
+    return openapi
+
+
+def test_openapi_logs_nothing_from_info(openapi: OpenAPI, caplog: pytest.LogCaptureFixture) -> None:
+    caplog.set_level(logging.INFO)
+    openapi.call("test_id")
+    assert caplog.record_tuples == []
+
+
+def test_openapi_logs_operation_info_to_debug(
+    openapi: OpenAPI, caplog: pytest.LogCaptureFixture
+) -> None:
+    caplog.set_level(logging.DEBUG)
+    openapi.call("test_id")
+    assert caplog.record_tuples == [
+        ("pulp_glue.openapi", logging.DEBUG + 3, "test_id : get test/"),
+        ("pulp_glue.openapi", logging.DEBUG + 2, ""),
+        ("pulp_glue.openapi", logging.DEBUG + 1, "{}"),
+        ("pulp_glue.openapi", logging.DEBUG + 3, "Response: 200"),
+        ("pulp_glue.openapi", logging.DEBUG + 1, "{}"),
+    ]

pulp_cli/__init__.py

@@ -1,3 +1,4 @@
+import logging
 import os
 import sys
 import typing as t
@@ -208,9 +209,8 @@ def main(
     timeout: int,
     cid: str,
 ) -> None:
-    def _debug_callback(level: int, x: str) -> None:
-        if verbose >= level:
-            click.secho(x, err=True, bold=True)
+    if verbose:
+        logging.basicConfig(level=logging.DEBUG + 4 - verbose, format="%(message)s")
 
     api_kwargs = dict(
         base_url=base_url,
@@ -220,7 +220,6 @@ def _debug_callback(level: int, x: str) -> None:
         verify_ssl=verify_ssl,
         refresh_cache=refresh_api,
         dry_run=dry_run,
-        debug_callback=_debug_callback,
         user_agent=f"Pulp-CLI/{__version__}",
         cid=cid,
     )