Use Rich for log messages (#851)

* Replace warnings.warn with logger.warning

* Use rich for logging

* Use Rich for top-level errors

* Update log level colors

* Use caplog in tests instead of capsys.

When using Rich, capsys results in wrapped output, which would be
very hard to make assertions on.

* Test color output

* Support --no-color

* Fix --no-color test

* Use dictConfig

* Fix unused import

* Avoid keyring backend warning

* Add coverage for HTTP error logging

* Use reconfigure instead of module attribute

* Add changelog entry

changelog/818.feature.rst

@@ -0,0 +1 @@
+Use Rich to add color to ``upload`` logging output.

mypy.ini

@@ -18,10 +18,6 @@ warn_return_any = True
 no_implicit_reexport = True
 strict_equality = True
 
-[mypy-colorama]
-; https://github.com/tartley/colorama/issues/206
-ignore_missing_imports = True
-
 [mypy-pkginfo]
 ; https://bugs.launchpad.net/pkginfo/+bug/1876591
 ignore_missing_imports = True

setup.cfg

@@ -45,7 +45,7 @@ install_requires=
     importlib_metadata >= 3.6
     keyring >= 15.1
     rfc3986 >= 1.4.0
-    colorama >= 0.4.3
+    rich
 include_package_data = True
 
 [options.entry_points]

tests/test_auth.py

@@ -149,21 +149,17 @@ def get_credential(system, username):
 
 
 def test_get_username_runtime_error_suppressed(
-    entered_username, keyring_no_backends_get_credential, recwarn, config
+    entered_username, keyring_no_backends_get_credential, caplog, config
 ):
     assert auth.Resolver(config, auth.CredentialInput()).username == "entered user"
-    assert len(recwarn) == 1
-    warning = recwarn.pop(UserWarning)
-    assert "fail!" in str(warning)
+    assert caplog.messages == ["fail!"]
 
 
 def test_get_password_runtime_error_suppressed(
-    entered_password, keyring_no_backends, recwarn, config
+    entered_password, keyring_no_backends, caplog, config
 ):
     assert auth.Resolver(config, auth.CredentialInput("user")).password == "entered pw"
-    assert len(recwarn) == 1
-    warning = recwarn.pop(UserWarning)
-    assert "fail!" in str(warning)
+    assert caplog.messages == ["fail!"]
 
 
 def test_get_username_return_none(entered_username, monkeypatch, config):
@@ -223,8 +219,11 @@ def test_warns_for_empty_password(
     config,
     caplog,
 ):
+    # Avoiding additional warning "No recommended backend was available"
+    monkeypatch.setattr(auth.keyring, "get_password", lambda system, user: None)
+
     monkeypatch.setattr(getpass, "getpass", lambda prompt: password)
 
     assert auth.Resolver(config, auth.CredentialInput()).password == password
 
-    assert caplog.messages[0].startswith(f"  {warning}")
+    assert caplog.messages[0].startswith(warning)

tests/test_main.py

@@ -12,18 +12,73 @@
 
 import sys
 
-import colorama
+import pretend
+import requests
 
 from twine import __main__ as dunder_main
+from twine.commands import upload
 
 
-def test_exception_handling(monkeypatch):
+def test_exception_handling(monkeypatch, capsys):
     monkeypatch.setattr(sys, "argv", ["twine", "upload", "missing.whl"])
-    message = "InvalidDistribution: Cannot find file (or expand pattern): 'missing.whl'"
-    assert dunder_main.main() == colorama.Fore.RED + message + colorama.Style.RESET_ALL
 
+    error = dunder_main.main()
+    assert error
 
-def test_no_color_exception(monkeypatch):
+    captured = capsys.readouterr()
+
+    # Hard-coding control characters for red text; couldn't find a succint alternative.
+    # Removing trailing whitespace on wrapped lines; trying to test it was ugly.
+    level = "\x1b[31mERROR   \x1b[0m"
+    assert [line.rstrip() for line in captured.out.splitlines()] == [
+        f"{level} InvalidDistribution: Cannot find file (or expand pattern):",
+        "         'missing.whl'",
+    ]
+
+
+def test_http_exception_handling(monkeypatch, capsys):
+    monkeypatch.setattr(sys, "argv", ["twine", "upload", "test.whl"])
+    monkeypatch.setattr(
+        upload,
+        "upload",
+        pretend.raiser(
+            requests.HTTPError(
+                response=pretend.stub(
+                    url="https://example.org",
+                    status_code=400,
+                    reason="Error reason",
+                )
+            )
+        ),
+    )
+
+    error = dunder_main.main()
+    assert error
+
+    captured = capsys.readouterr()
+
+    # Hard-coding control characters for red text; couldn't find a succint alternative.
+    # Removing trailing whitespace on wrapped lines; trying to test it was ugly.
+    level = "\x1b[31mERROR   \x1b[0m"
+    assert [line.rstrip() for line in captured.out.splitlines()] == [
+        f"{level} HTTPError: 400 Bad Request from https://example.org",
+        "         Error reason",
+    ]
+
+
+def test_no_color_exception(monkeypatch, capsys):
     monkeypatch.setattr(sys, "argv", ["twine", "--no-color", "upload", "missing.whl"])
-    message = "InvalidDistribution: Cannot find file (or expand pattern): 'missing.whl'"
-    assert dunder_main.main() == message
+
+    error = dunder_main.main()
+    assert error
+
+    captured = capsys.readouterr()
+
+    # Removing trailing whitespace on wrapped lines; trying to test it was ugly.
+    assert [line.rstrip() for line in captured.out.splitlines()] == [
+        "ERROR    InvalidDistribution: Cannot find file (or expand pattern):",
+        "         'missing.whl'",
+    ]
+
+
+# TODO: Test verbose output formatting

tests/test_settings.py

@@ -66,16 +66,14 @@ def test_setup_logging(verbose, log_level):
     "verbose",
     [True, False],
 )
-def test_print_config_path_if_verbose(config_file, capsys, make_settings, verbose):
+def test_print_config_path_if_verbose(config_file, caplog, make_settings, verbose):
     """Print the location of the .pypirc config used by the user."""
     make_settings(verbose=verbose)
 
-    captured = capsys.readouterr()
-
     if verbose:
-        assert captured.out == f"Using configuration from {config_file}\n"
+        assert caplog.messages == [f"Using configuration from {config_file}"]
     else:
-        assert captured.out == ""
+        assert caplog.messages == []
 
 
 def test_identity_requires_sign():

tests/test_upload.py

@@ -59,7 +59,7 @@ def upload_settings(make_settings, stub_repository):
     return upload_settings
 
 
-def test_make_package_pre_signed_dist(upload_settings, capsys):
+def test_make_package_pre_signed_dist(upload_settings, caplog):
     """Create a PackageFile and print path, size, and user-provided signature."""
     filename = helpers.WHEEL_FIXTURE
     expected_size = "15.4 KB"
@@ -74,12 +74,13 @@ def test_make_package_pre_signed_dist(upload_settings, capsys):
     assert package.filename == filename
     assert package.gpg_signature is not None
 
-    captured = capsys.readouterr()
-    assert captured.out.count(f"{filename} ({expected_size})") == 1
-    assert captured.out.count(f"Signed with {signed_filename}") == 1
+    assert caplog.messages == [
+        f"{filename} ({expected_size})",
+        f"Signed with {signed_filename}",
+    ]
 
 
-def test_make_package_unsigned_dist(upload_settings, monkeypatch, capsys):
+def test_make_package_unsigned_dist(upload_settings, monkeypatch, caplog):
     """Create a PackageFile and print path, size, and Twine-generated signature."""
     filename = helpers.NEW_WHEEL_FIXTURE
     expected_size = "21.9 KB"
@@ -98,9 +99,10 @@ def stub_sign(package, *_):
     assert package.filename == filename
     assert package.gpg_signature is not None
 
-    captured = capsys.readouterr()
-    assert captured.out.count(f"{filename} ({expected_size})") == 1
-    assert captured.out.count(f"Signed with {package.signed_filename}") == 1
+    assert caplog.messages == [
+        f"{filename} ({expected_size})",
+        f"Signed with {package.signed_filename}",
+    ]
 
 
 def test_successs_prints_release_urls(upload_settings, stub_repository, capsys):
@@ -123,27 +125,26 @@ def test_successs_prints_release_urls(upload_settings, stub_repository, capsys):
     assert captured.out.count(NEW_RELEASE_URL) == 1
 
 
-def test_print_packages_if_verbose(upload_settings, capsys):
+def test_print_packages_if_verbose(upload_settings, caplog):
     """Print the path and file size of each distribution attempting to be uploaded."""
     dists_to_upload = {
         helpers.WHEEL_FIXTURE: "15.4 KB",
+        helpers.NEW_WHEEL_FIXTURE: "21.9 KB",
         helpers.SDIST_FIXTURE: "20.8 KB",
         helpers.NEW_SDIST_FIXTURE: "26.1 KB",
-        helpers.NEW_WHEEL_FIXTURE: "21.9 KB",
     }
 
     upload_settings.verbose = True
 
     result = upload.upload(upload_settings, dists_to_upload.keys())
     assert result is None
 
-    captured = capsys.readouterr()
-
-    for filename, size in dists_to_upload.items():
-        assert captured.out.count(f"{filename} ({size})") == 1
+    assert [m for m in caplog.messages if m.endswith("KB)")] == [
+        f"{filename} ({size})" for filename, size in dists_to_upload.items()
+    ]
 
 
-def test_print_response_if_verbose(upload_settings, stub_response, capsys):
+def test_print_response_if_verbose(upload_settings, stub_response, caplog):
     """Print details about the response from the repostiry."""
     upload_settings.verbose = True
 
@@ -153,13 +154,12 @@ def test_print_response_if_verbose(upload_settings, stub_response, capsys):
     )
     assert result is None
 
-    captured = capsys.readouterr()
     response_log = (
         f"Response from {stub_response.url}:\n"
         f"{stub_response.status_code} {stub_response.reason}"
     )
 
-    assert captured.out.count(response_log) == 2
+    assert caplog.messages.count(response_log) == 2
 
 
 def test_success_with_pre_signed_distribution(upload_settings, stub_repository):
@@ -205,7 +205,9 @@ def test_success_when_gpg_is_run(upload_settings, stub_repository, monkeypatch):
 
 
 @pytest.mark.parametrize("verbose", [False, True])
-def test_exception_for_http_status(verbose, upload_settings, stub_response, capsys):
+def test_exception_for_http_status(
+    verbose, upload_settings, stub_response, capsys, caplog
+):
     upload_settings.verbose = verbose
 
     stub_response.is_redirect = False
@@ -221,11 +223,15 @@ def test_exception_for_http_status(verbose, upload_settings, stub_response, caps
     assert RELEASE_URL not in captured.out
 
     if verbose:
-        assert stub_response.text in captured.out
-        assert "--verbose" not in captured.out
+        assert caplog.messages == [
+            f"{helpers.WHEEL_FIXTURE} (15.4 KB)",
+            f"Response from {stub_response.url}:\n403 {stub_response.reason}",
+            stub_response.text,
+        ]
     else:
-        assert stub_response.text not in captured.out
-        assert "--verbose" in captured.out
+        assert caplog.messages == [
+            "Error during upload. Retry with the --verbose option for more details."
+        ]
 
 
 def test_get_config_old_format(make_settings, config_file):

tests/test_utils.py

@@ -265,7 +265,7 @@ def test_check_status_code_for_deprecated_pypi_url(repo_url):
     [True, False],
 )
 def test_check_status_code_for_missing_status_code(
-    capsys, repo_url, verbose, make_settings
+    caplog, repo_url, verbose, make_settings, config_file
 ):
     """Print HTTP errors based on verbosity level."""
     response = pretend.stub(
@@ -280,9 +280,10 @@ def test_check_status_code_for_missing_status_code(
     with pytest.raises(requests.HTTPError):
         utils.check_status_code(response, verbose)
 
-    captured = capsys.readouterr()
-
-    assert captured.out.count("--verbose option") == 0 if verbose else 1
+        message = (
+            "Error during upload. Retry with the --verbose option for more details."
+        )
+        assert caplog.messages.count(message) == 0 if verbose else 1
 
 
 @pytest.mark.parametrize(

twine/__main__.py

@@ -13,40 +13,38 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import http
+import logging
 import sys
 from typing import Any
 
-import colorama
 import requests
 
 from twine import cli
 from twine import exceptions
 
+logger = logging.getLogger(__name__)
+
 
 def main() -> Any:
+    # Ensure that all errors are logged, even before argparse
+    cli.configure_output()
+
     try:
-        result = cli.dispatch(sys.argv[1:])
+        error = cli.dispatch(sys.argv[1:])
     except requests.HTTPError as exc:
+        error = True
         status_code = exc.response.status_code
         status_phrase = http.HTTPStatus(status_code).phrase
-        result = (
+        logger.error(
             f"{exc.__class__.__name__}: {status_code} {status_phrase} "
             f"from {exc.response.url}\n"
             f"{exc.response.reason}"
         )
     except exceptions.TwineException as exc:
-        result = f"{exc.__class__.__name__}: {exc.args[0]}"
-
-    return _format_error(result) if isinstance(result, str) else result
-
-
-def _format_error(message: str) -> str:
-    pre_style, post_style = "", ""
-    if not cli.args.no_color:
-        colorama.init()
-        pre_style, post_style = colorama.Fore.RED, colorama.Style.RESET_ALL
+        error = True
+        logger.error(f"{exc.__class__.__name__}: {exc.args[0]}")
 
-    return f"{pre_style}{message}{post_style}"
+    return error
 
 
 if __name__ == "__main__":

twine/auth.py

@@ -1,7 +1,6 @@
 import functools
 import getpass
 import logging
-import warnings
 from typing import Callable, Optional, Type, cast
 
 import keyring
@@ -64,7 +63,7 @@ def get_username_from_keyring(self) -> Optional[str]:
             # To support keyring prior to 15.2
             pass
         except Exception as exc:
-            warnings.warn(str(exc))
+            logger.warning(str(exc))
         return None
 
     def get_password_from_keyring(self) -> Optional[str]:
@@ -74,7 +73,7 @@ def get_password_from_keyring(self) -> Optional[str]:
             logger.info("Querying keyring for password")
             return cast(str, keyring.get_password(system, username))
         except Exception as exc:
-            warnings.warn(str(exc))
+            logger.warning(str(exc))
         return None
 
     def username_from_keyring_or_prompt(self) -> str: