Update I/O handling for better CI output (#2)

Rich is a really great library for use with CLI tools, but internally it's structured around a terminal. Specifically, the issue is that it requires a width. In the absence of one, it picks 80, which is normally fine, but it makes wide log outputs look really bad and take up more space than they should.

This PR tries to determine if pretty terminal output makes sense, and if it doesn't, it foregoes the use of Rich altogether.

src/mads/build/logging.py

@@ -2,21 +2,28 @@
 Configure logging for the build and define helper functions.
 """
 
+import sys
 import time
 import logging
 from typing import List, Union
 from pathlib import Path
 from datetime import datetime, timedelta
-from rich.logging import RichHandler
-from rich.console import Console
 
+from mads.environ import InOut
 
 LSS_START = "┌"
 LSS_END = "└"
 LOG_FILE = Path("build-log.txt")
 DATA_SUFFIXES = ["B", "KB", "MB", "GB", "TB", "PB"]
 
-console = Console(stderr=True)
+io = InOut()
+console = None
+
+if io.is_terminal:
+    from rich.logging import RichHandler
+    from rich.console import Console
+
+    console = Console(stderr=True)
 
 
 class BuildLogger(logging.LoggerAdapter):
@@ -152,12 +159,17 @@ def new_logger() -> BuildLogger:
     # Add a null handler to prevent it from being populated automatically
     root.addHandler(logging.NullHandler())
 
-    fmt = logging.Formatter(fmt="%(message)s", datefmt="%H:%M:%S")
-
     # Set up the handlers we want to use
-    handlers: list[logging.Handler] = [
-        RichHandler(console=console, show_level=False, show_path=False),
-    ]
+    handlers: list[logging.Handler] = []
+
+    fmtstr = "%(asctime)s %(message)s"
+    if io.is_terminal:
+        fmtstr = "%(message)s"
+        handlers.append(RichHandler(console=console, show_level=False, show_path=False))
+    else:
+        handlers.append(logging.StreamHandler(sys.stdout))
+
+    fmt = logging.Formatter(fmt=fmtstr, datefmt="%H:%M:%S")
 
     # Configure our logger with those handlers
     mlog.setLevel(logging.DEBUG)

src/mads/cli/commands/environ.py

@@ -18,5 +18,6 @@ def run(args):
         p(environ.Docker())
         p(environ.Runner.current())
         p(environ.Resources())
+        p(environ.InOut())
 
     parser.set_defaults(func=run)

src/mads/environ/__init__.py

@@ -1,3 +1,4 @@
+from .in_out import InOut
 from .git import Git
 from .resources import Resources
 from .runners.runner import Runner
@@ -8,4 +9,5 @@
     "Resources",
     "Runner",
     "Docker",
+    "InOut",
 ]

src/mads/environ/in_out.py

@@ -0,0 +1,50 @@
+import sys
+from pydantic import computed_field
+from pydantic_settings import BaseSettings
+from .runners import Runner
+
+
+def runner_settings() -> dict:
+    """Override the detected settings with runner information"""
+
+    runner = Runner.current
+    return runner.io_settings
+
+
+class InOut(BaseSettings):
+    """The input/output we're running with. IO is a reserved name in some contexts."""
+
+    term: str | None = None
+    force_terminal: bool | None = None
+
+    @computed_field
+    def is_terminal(self) -> bool:
+        if self.force_terminal is not None:
+            return self.force_terminal
+        return self.is_atty
+
+    @computed_field
+    def is_atty(self) -> bool:
+        return sys.stdout.isatty()
+
+    @computed_field
+    def is_interactive(self) -> bool:
+        return self.is_atty and not self.is_dumb
+
+    @property
+    def is_dumb(self) -> bool:
+        return self.term is not None and self.term.lower() in ("dumb", "unknown")
+
+    def __rich_repr__(self):
+        yield "term", self.term
+        yield "is_terminal", self.is_terminal
+        yield "force_terminal", self.force_terminal, None
+        yield "is_atty", self.is_atty, True
+        yield "is_interactive", self.is_interactive
+
+    @classmethod
+    def settings_customise_sources(cls, *args, env_settings, **kwargs) -> tuple:
+        return (
+            runner_settings,
+            env_settings,
+        )

src/mads/environ/runners/github_actions.py

@@ -1,6 +1,7 @@
 """Information about the GitHub Actions environment."""
 
 import os
+from typing import ClassVar
 from pydantic import Field
 from .runner import Runner
 from pydantic_settings import SettingsConfigDict
@@ -24,6 +25,8 @@ class GitHubActions(Runner):
     repository_owner: str = ""
     run_attempt: int = 1
 
+    io_settings: ClassVar[dict] = {"force_terminal": False}
+
     @classmethod
     def detect(cls) -> bool:
         """Detect if we're running in GitHub Actions"""

src/mads/environ/runners/local.py

@@ -1,7 +1,6 @@
 import time
 from pydantic import Field
 
-
 from mads.lib.path import current_repo
 from .runner import Runner
 

src/mads/environ/runners/runner.py

@@ -6,13 +6,12 @@
 """
 
 from typing import ClassVar, Type
-from abc import ABC, abstractmethod
 from pydantic import field_validator
 from pydantic_settings import BaseSettings
 from mads.lib.classproperty import classproperty
 
 
-class Runner(BaseSettings, ABC):
+class Runner(BaseSettings):
     """
     A Runner should tell us details about why this operation is running by pulling
     information from the environment.
@@ -41,7 +40,9 @@ class Runner(BaseSettings, ABC):
     def url(self) -> str | None:
         pass
 
-    _runners: ClassVar = []
+    _runners: ClassVar[list] = []
+
+    io_settings: ClassVar[dict] = {}
 
     # Save a list of all subclasses so we can detect which one we're running in
     def __init_subclass__(cls, /, **kwargs):
@@ -57,9 +58,9 @@ def current(cls) -> Type["Runner"]:
         raise RuntimeError("Unable to detect runner")
 
     @classmethod
-    @abstractmethod
     def detect(cls) -> bool:
         """Detect if we're running in this class of runner"""
+        return False
 
     @field_validator("repo", "head_ref", "base_ref", mode="before")
     def must_be_name_only(cls, v: str) -> str | None:

test/mads/test_cli.py

@@ -1,17 +1,34 @@
 """Test the CLI interface"""
 
 from mads import cli
+from mads.environ import Git
 
 
 def test_determine_tag(capsys):
     """Test that the help message works"""
 
     cli.main(["docker", "tag"])
-    assert capsys.readouterr().out == "dev\n"
+
+    git = Git()
+    expected = {
+        "main": "latest",
+        "master": "latest",
+        "beta": "beta",
+        None: "dev",
+    }.get(git.branch, "dev")
+    assert capsys.readouterr().out == f"{expected}\n"
 
 
 def test_determine_tag_default(capsys):
     """Test that the help message works"""
 
     cli.main(["docker", "tag", "--default", "beta"])
-    assert capsys.readouterr().out == "beta\n"
+
+    git = Git()
+    expected = {
+        "main": "latest",
+        "master": "latest",
+        "beta": "beta",
+        None: "beta",
+    }.get(git.branch, "beta")
+    assert capsys.readouterr().out == f"{expected}\n"