From c72a42087508e2aa03c0a183cf140dcc2b5aa093 Mon Sep 17 00:00:00 2001
From: Joe Shimkus <35382397+jshimkus-rh@users.noreply.github.com>
Date: Wed, 3 Jan 2024 12:04:34 -0500
Subject: [PATCH] feat: implement banner output (#625)

Adds banner-type output to ansible-rulebook in the manner previously
demonstrated.

There's an open question as to how we want to handle outputting events
in rule_set_runner.py.
The previous code disabled args.print_events if verbosity was 2+ as the
logging would output the event.
We shouldn't have two different forms of output for the same data and
I'd opt for only the args.print_event form in all cases for consistency
unless there's specific, great enough value in using the logger when
verbosity is 2+.
At present, this code forcibly enables args.print_events when verbosity
is 2+ so both forms are output.

---------

Signed-off-by: Joe Shimkus <jshimkus@redhat.com>
---
 CHANGELOG.md                           |   3 +
 ansible_rulebook/action/debug.py       |  26 ++--
 ansible_rulebook/action/print_event.py |  15 ++-
 ansible_rulebook/action/shutdown.py    |  17 +--
 ansible_rulebook/cli.py                |  12 +-
 ansible_rulebook/collection.py         |   6 +-
 ansible_rulebook/rule_set_runner.py    |  20 ++-
 ansible_rulebook/terminal.py           | 177 +++++++++++++++++++++++++
 ansible_rulebook/util.py               |  18 ++-
 tests/e2e/test_actions.py              |  34 +++--
 tests/e2e/test_operators.py            |  18 ++-
 tests/unit/action/test_print_event.py  |   5 +-
 tests/unit/test_terminal.py            | 117 ++++++++++++++++
 13 files changed, 400 insertions(+), 68 deletions(-)
 create mode 100644 ansible_rulebook/terminal.py
 create mode 100644 tests/unit/test_terminal.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index a316a167..3a819173 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,9 @@
 - Support for standalone boolean in conditions
 - Add basic auth to controller
 
+### Changed
+- Generic print as well as printing of events use new banner style
+
 ### Fixed
 
 ## [1.0.4] - 2023-10-30
diff --git a/ansible_rulebook/action/debug.py b/ansible_rulebook/action/debug.py
index ba2b0d4d..cd1d7cc7 100644
--- a/ansible_rulebook/action/debug.py
+++ b/ansible_rulebook/action/debug.py
@@ -15,12 +15,11 @@
 import logging
 import sys
 from dataclasses import asdict
-from pprint import pprint
 
 import dpath
 from drools import ruleset as lang
 
-from ansible_rulebook.util import get_horizontal_rule
+from ansible_rulebook import terminal
 
 from .control import Control
 from .helper import Helper
@@ -41,6 +40,7 @@ class Debug:
     def __init__(self, metadata: Metadata, control: Control, **action_args):
         self.helper = Helper(metadata, control, "debug")
         self.action_args = action_args
+        self.display = terminal.Display()
 
     async def __call__(self):
         if "msg" in self.action_args:
@@ -48,21 +48,18 @@ async def __call__(self):
             if not isinstance(messages, list):
                 messages = [messages]
             for msg in messages:
-                print(msg)
+                self.display.banner("debug", msg)
         elif "var" in self.action_args:
             key = self.action_args.get("var")
             try:
-                print(
-                    dpath.get(
-                        self.helper.control.variables, key, separator="."
-                    )
+                value = dpath.get(
+                    self.helper.control.variables, key, separator="."
                 )
+                self.display.banner("debug", f"{key}: {value}")
             except KeyError:
                 logger.error("Key %s not found in variable pool", key)
                 raise
         else:
-            print(get_horizontal_rule("="))
-            print("kwargs:")
             args = asdict(self.helper.metadata)
             project_data_file = self.helper.control.project_data_file
             args.update(
@@ -73,11 +70,12 @@ async def __call__(self):
                     "project_data_file": project_data_file,
                 }
             )
-            pprint(args)
-            print(get_horizontal_rule("="))
-            print("facts:")
-            pprint(lang.get_facts(self.helper.metadata.rule_set))
-            print(get_horizontal_rule("="))
+            self.display.banner("debug: kwargs", args, pretty=True)
+            self.display.banner(
+                "debug: facts",
+                lang.get_facts(self.helper.metadata.rule_set),
+                pretty=True,
+            )
 
         sys.stdout.flush()
         await self.helper.send_default_status()
diff --git a/ansible_rulebook/action/print_event.py b/ansible_rulebook/action/print_event.py
index 2de0c39a..95a145e2 100644
--- a/ansible_rulebook/action/print_event.py
+++ b/ansible_rulebook/action/print_event.py
@@ -13,8 +13,8 @@
 #  limitations under the License.
 
 import sys
-from pprint import pprint
-from typing import Callable
+
+from ansible_rulebook import terminal
 
 from .control import Control
 from .helper import Helper
@@ -30,16 +30,17 @@ class PrintEvent:
     def __init__(self, metadata: Metadata, control: Control, **action_args):
         self.helper = Helper(metadata, control, "print_event")
         self.action_args = action_args
+        self.display = terminal.Display()
 
     async def __call__(self):
-        print_fn: Callable = print
-        if self.action_args.get("pretty", False):
-            print_fn = pprint
-
         var_name = (
             "events" if "events" in self.helper.control.variables else "event"
         )
 
-        print_fn(self.helper.control.variables[var_name])
+        self.display.banner(
+            "event",
+            self.helper.control.variables[var_name],
+            pretty=self.action_args.get("pretty", False),
+        )
         sys.stdout.flush()
         await self.helper.send_default_status()
diff --git a/ansible_rulebook/action/shutdown.py b/ansible_rulebook/action/shutdown.py
index e33c3225..00024ceb 100644
--- a/ansible_rulebook/action/shutdown.py
+++ b/ansible_rulebook/action/shutdown.py
@@ -14,6 +14,7 @@
 
 from ansible_rulebook.exception import ShutdownException
 from ansible_rulebook.messages import Shutdown as ShutdownMessage
+from ansible_rulebook.terminal import Display
 from ansible_rulebook.util import run_at
 
 from .control import Control
@@ -43,16 +44,12 @@ async def __call__(self):
                 "kind": kind,
             }
         )
-        print(
-            "Ruleset: %s rule: %s has initiated shutdown of type: %s. "
-            "Delay: %.3f seconds, Message: %s"
-            % (
-                self.helper.metadata.rule_set,
-                self.helper.metadata.rule,
-                kind,
-                delay,
-                message,
-            )
+        Display.instance().banner(
+            "ruleset",
+            f"{self.helper.metadata.rule_set}"
+            f" rule: {self.helper.metadata.rule}"
+            f" has initiated shutdown of type: {kind}."
+            f" Delay: {delay:.3f} seconds, Message: {message}",
         )
         raise ShutdownException(
             ShutdownMessage(message=message, delay=delay, kind=kind)
diff --git a/ansible_rulebook/cli.py b/ansible_rulebook/cli.py
index e0efff34..e4470985 100644
--- a/ansible_rulebook/cli.py
+++ b/ansible_rulebook/cli.py
@@ -33,11 +33,14 @@
 
 import ansible_rulebook  # noqa: E402
 from ansible_rulebook import app  # noqa: E402
+from ansible_rulebook import terminal  # noqa: E402
 from ansible_rulebook.conf import settings  # noqa: E402
 from ansible_rulebook.job_template_runner import (  # noqa: E402
     job_template_runner,
 )
 
+display = terminal.Display()
+
 DEFAULT_VERBOSITY = 0
 
 logger = logging.getLogger(__name__)
@@ -227,11 +230,15 @@ def setup_logging(args: argparse.Namespace) -> None:
     if args.verbosity >= 2:
         level = logging.DEBUG
         stream = sys.stdout
-        args.print_events = False
     elif args.verbosity == 1:
         level = logging.INFO
         stream = sys.stdout
 
+    # As Display is a singleton if it was created elsewhere we may need to
+    # adjust the level.
+    if display.level > level:
+        display.level = level
+
     logging.basicConfig(stream=stream, level=level, format=LOG_FORMAT)
     logging.getLogger("drools.").setLevel(level)
 
@@ -244,6 +251,7 @@ def main(args: List[str] = None) -> int:
 
     args = parser.parse_args(args)
     validate_args(args)
+    setup_logging(args)
 
     if args.controller_url:
         job_template_runner.host = args.controller_url
@@ -271,8 +279,6 @@ def main(args: List[str] = None) -> int:
     if args.execution_strategy:
         settings.default_execution_strategy = args.execution_strategy
 
-    setup_logging(args)
-
     try:
         asyncio.run(app.run(args))
     except KeyboardInterrupt:
diff --git a/ansible_rulebook/collection.py b/ansible_rulebook/collection.py
index d1f7b044..9dc1dc02 100644
--- a/ansible_rulebook/collection.py
+++ b/ansible_rulebook/collection.py
@@ -20,6 +20,8 @@
 
 import yaml
 
+from ansible_rulebook import terminal
+
 ANSIBLE_GALAXY = shutil.which("ansible-galaxy")
 EDA_PATH_PREFIX = "extensions/eda"
 
@@ -130,7 +132,9 @@ def load_rulebook(collection, rulebook):
     if not location:
         return False
     with open(location) as f:
-        print(f"Loading rulebook from {location}")
+        terminal.Display.instance().banner(
+            "collection", f"Loading rulebook from {location}"
+        )
         return yaml.safe_load(f.read())
 
 
diff --git a/ansible_rulebook/rule_set_runner.py b/ansible_rulebook/rule_set_runner.py
index 8d6c534c..695d49b8 100644
--- a/ansible_rulebook/rule_set_runner.py
+++ b/ansible_rulebook/rule_set_runner.py
@@ -16,7 +16,7 @@
 import gc
 import logging
 import uuid
-from pprint import PrettyPrinter, pformat
+from pprint import pformat
 from types import MappingProxyType
 from typing import Dict, List, Optional, Union, cast
 
@@ -28,6 +28,7 @@
 )
 from drools.ruleset import session_stats
 
+from ansible_rulebook import terminal
 from ansible_rulebook.action.control import Control
 from ansible_rulebook.action.debug import Debug
 from ansible_rulebook.action.metadata import Metadata
@@ -102,6 +103,7 @@ def __init__(
         self.active_actions = set()
         self.broadcast_method = broadcast_method
         self.event_counter = 0
+        self.display = terminal.Display()
 
     async def run_ruleset(self):
         tasks = []
@@ -203,12 +205,20 @@ async def _drain_source_queue(self):
         try:
             while True:
                 data = await self.ruleset_queue_plan.source_queue.get()
+                # Default to output events at debug level.
+                level = logging.DEBUG
+
+                # If print_events is specified adjust the level to the
+                # display's current level to guarantee output.
                 if self.parsed_args and self.parsed_args.print_events:
-                    PrettyPrinter(indent=4).pprint(data)
+                    level = self.display.level
+
+                self.display.banner("received event", level=level)
+                self.display.output(f"Ruleset: {self.name}", level=level)
+                self.display.output("Event:", level=level)
+                self.display.output(data, pretty=True, level=level)
+                self.display.banner(level=level)
 
-                logger.debug(
-                    "Ruleset: %s, received event: %s ", self.name, str(data)
-                )
                 if isinstance(data, Shutdown):
                     self.shutdown = data
                     return await self._handle_shutdown()
diff --git a/ansible_rulebook/terminal.py b/ansible_rulebook/terminal.py
new file mode 100644
index 00000000..a15f0c7e
--- /dev/null
+++ b/ansible_rulebook/terminal.py
@@ -0,0 +1,177 @@
+#  Copyright 2023 Red Hat, Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+import datetime
+import logging
+import os
+import pprint
+import re
+import threading
+import typing
+
+logger = logging.getLogger(__name__)
+
+
+class _Singleton(type):
+    def __init__(cls, name, bases, dct, **kwargs) -> None:
+        super().__init__(name, bases, dct, **kwargs)
+        cls.__singleton = None
+        cls.__lock = threading.RLock()
+
+    def __call__(cls, *args, **kwargs) -> type:
+        if not cls.__singleton:
+            with cls.__lock:
+                if not cls.__singleton:
+                    cls.__singleton = super().__call__(*args, **kwargs)
+        return cls.__singleton
+
+
+class Singleton(metaclass=_Singleton):
+    pass
+
+
+class DisplayBannerError(Exception):
+    pass
+
+
+class DisplayBannerIncompleteError(DisplayBannerError):
+    pass
+
+
+class Display(Singleton):
+    @classmethod
+    def instance(cls, level: typing.Optional[int] = None):
+        instance = cls()
+        # Display is a singleton; adjust the level if specified
+        if level is not None:
+            instance.level = level
+        return instance
+
+    @classmethod
+    def get_banners(cls, banner: str, content: str) -> typing.List[str]:
+        """
+        Searches for all specified banners in the provided content (assumed to
+        contain embedded line separators) and returns a list of such banners
+        each of which is a string with embedded line separators.
+
+        If no banners are found returns an empty list.
+        If an incomplete banner is found raises DisplayBannerIncompleteError.
+
+        A "banner" output has the following form:
+
+            ** <date> <time> [<banner>] ***************************************
+            <content>
+            *******************************************************************
+
+        The number of asterixes varies as it is limited by the terminal window
+        (if there is one) width or 80 columnns otherwise.
+        """
+        start = re.compile(
+            r"^\*\* \d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{6} "
+            r"\[" + banner + r"\] \*{0,}$"
+        )
+        end = re.compile(r"^\*{1,}$")
+
+        banners = []
+        current_banner = []
+        for line in content.strip().splitlines():
+            if current_banner:
+                current_banner.append(line)
+                if re.search(end, line):
+                    banners.append(os.linesep.join(current_banner))
+                    current_banner = []
+                continue
+
+            if re.search(start, line):
+                current_banner.append(line)
+
+        if current_banner:
+            raise DisplayBannerIncompleteError(
+                f"banner '{banner}' incomplete: {current_banner}"
+            )
+
+        return banners
+
+    def banner(
+        self,
+        banner: str = "",
+        content: typing.Optional[typing.Any] = None,
+        *,
+        pretty: bool = False,
+        level: int = logging.INFO,
+        **kwargs,
+    ) -> None:
+        banner = banner.strip()
+        if banner:
+            banner = f"[{banner}]"
+        formatted_banner = self._format_banner(banner)
+        if banner:
+            formatted_banner = f"\n{formatted_banner}"
+        self.output(
+            formatted_banner,
+            level=level,
+            file=kwargs.get("file", None),
+            flush=kwargs.get("flush", False),
+        )
+        if content is not None:
+            self.output(content, pretty=pretty, level=level, **kwargs)
+            self.banner(level=level, **kwargs)
+
+    def output(
+        self,
+        content: typing.Any,
+        *,
+        pretty: bool = False,
+        level: int = logging.INFO,
+        **kwargs,
+    ) -> None:
+        if level >= self.level:
+            if pretty:
+                content = pprint.pformat(content)
+            print(content, **kwargs)
+
+    def __init__(self, level: int = logging.INFO) -> None:
+        super().__init__()
+        self.level = level
+
+    def _format_banner(self, banner: str) -> str:
+        if len(banner) > 0:
+            banner = self._prefix_time(banner)
+        banner = self._rule_embed(banner)
+        return banner
+
+    def _prefix_time(self, content: str) -> str:
+        return f"{datetime.datetime.now().isoformat(' ')} {content}"
+
+    def _rule(self, character: str = "*", /, text_length: int = 0) -> str:
+        try:
+            rule_length = int(os.get_terminal_size()[0])
+        except OSError:
+            rule_length = 80
+
+        rule_length = max(0, rule_length - text_length)
+        return character * rule_length
+
+    def _rule_embed(self, content: str) -> str:
+        prefix = "** "
+        suffix = " "
+
+        if len(content) == 0:
+            prefix = ""
+            suffix = ""
+
+        length = len(prefix) + len(content) + len(suffix)
+        rule = self._rule(text_length=length)
+        content = f"{prefix}{content}{suffix}{rule}"
+        return content
diff --git a/ansible_rulebook/util.py b/ansible_rulebook/util.py
index 6160f8e0..ae9c83d3 100644
--- a/ansible_rulebook/util.py
+++ b/ansible_rulebook/util.py
@@ -32,6 +32,7 @@
 from packaging import version
 from packaging.version import InvalidVersion
 
+from ansible_rulebook import terminal
 from ansible_rulebook.conf import settings
 from ansible_rulebook.exception import (
     InvalidFilterNameException,
@@ -40,14 +41,8 @@
 
 logger = logging.getLogger(__name__)
 
-EDA_BUILTIN_FILTER_PREFIX = "eda.builtin."
-
 
-def get_horizontal_rule(character):
-    try:
-        return character * int(os.get_terminal_size()[0])
-    except OSError:
-        return character * 80
+EDA_BUILTIN_FILTER_PREFIX = "eda.builtin."
 
 
 def render_string(value: str, context: Dict) -> str:
@@ -182,7 +177,8 @@ def check_jvm():
     """
     java_home = get_java_home()
     if not java_home:
-        print(
+        terminal.Display.instance().banner(
+            "util",
             "Java executable or JAVA_HOME environment variable not found."
             "Please install a valid JVM.",
             file=sys.stderr,
@@ -197,14 +193,16 @@ def check_jvm():
         if clean_version:
             java_version = clean_version.groups()[0]
         if version.parse(java_version) < version.parse("17"):
-            print(
+            terminal.Display.instance().banner(
+                "util",
                 "The minimum supported Java version is 17. "
                 f"Found version: {java_version}",
                 file=sys.stderr,
             )
             sys.exit(1)
     except InvalidVersion as exinfo:
-        print(
+        terminal.Display.instance().banner(
+            "util: exception",
             exinfo,
             file=sys.stderr,
         )
diff --git a/tests/e2e/test_actions.py b/tests/e2e/test_actions.py
index 76a84ff4..a1f734a3 100644
--- a/tests/e2e/test_actions.py
+++ b/tests/e2e/test_actions.py
@@ -8,6 +8,8 @@
 import pytest
 from pytest_check import check
 
+from ansible_rulebook import terminal
+
 from . import utils
 from .settings import SETTINGS
 
@@ -155,25 +157,37 @@ def test_actions_sanity(update_environment):
             not in result.stdout
         ), "retract_fact action failed"
 
+    with check:
+        assert (
+            "'action': 'multiple_actions'" in result.stdout
+        ), "multiple_action action failed"
+
     multiple_actions_expected_output = (
-        "Ruleset: Test actions sanity rule: Test multiple actions in "
+        "Test actions sanity rule: Test multiple actions in "
         "sequential order has initiated shutdown of type: graceful. "
-        "Delay: 0.000 seconds, Message: Sequential action #2: shutdown\n"
-        "Sequential action #3: debug"
+        "Delay: 0.000 seconds, Message: Sequential action #2: shutdown"
     )
 
     with check:
-        assert (
-            multiple_actions_expected_output in result.stdout
-        ), "multiple sequential actions failed"
+        banners = terminal.Display.get_banners("ruleset", result.stdout)
+        banners = [
+            banner
+            for banner in banners
+            if multiple_actions_expected_output in banner
+        ]
+        assert banners, "multiple sequential actions ruleset failed"
 
     with check:
-        assert (
-            "'action': 'multiple_actions'" in result.stdout
-        ), "multiple_action action failed"
+        banners = terminal.Display.get_banners("debug", result.stdout)
+        banners = [
+            banner
+            for banner in banners
+            if "Sequential action #3: debug" in banner
+        ]
+        assert banners, "multiple sequential actions debug failed"
 
     assert (
-        len(result.stdout.splitlines()) == 105
+        len(result.stdout.splitlines()) == 121
     ), "unexpected output from the rulebook"
 
 
diff --git a/tests/e2e/test_operators.py b/tests/e2e/test_operators.py
index 8d0bcb15..74de9a53 100644
--- a/tests/e2e/test_operators.py
+++ b/tests/e2e/test_operators.py
@@ -7,6 +7,8 @@
 import pytest
 from pytest_check import check
 
+from ansible_rulebook import terminal
+
 from . import utils
 from .settings import SETTINGS
 
@@ -227,12 +229,16 @@ def test_relational_operators(update_environment):
         ), "testcase #22 failed"
 
     with check:
-        assert (
-            "Ruleset: Test relational operators rule: Finish"
-            " - test shutdown msg has initiated shutdown" in result.stdout
-        ), "Shutdown message failed"
+        banners = terminal.Display.get_banners("ruleset", result.stdout)
+        banners = [
+            banner
+            for banner in banners
+            if "Test relational operators rule: Finish"
+            " - test shutdown msg has initiated shutdown" in banner
+        ]
+        assert banners, "Shutdown message failed"
 
-    assert len(result.stdout.splitlines()) == 25, "Unexpected output"
+    assert len(result.stdout.splitlines()) == 100, "Unexpected output"
 
 
 @pytest.mark.e2e
@@ -359,7 +365,7 @@ def test_membership_operators(update_environment):
             "Output for Testcase #12" in result.stdout
         ), "Testcase #12 failed"
 
-    assert len(result.stdout.splitlines()) == 19, "Unexpected output"
+    assert len(result.stdout.splitlines()) == 76, "Unexpected output"
 
 
 @pytest.mark.e2e
diff --git a/tests/unit/action/test_print_event.py b/tests/unit/action/test_print_event.py
index 5d455da2..143d98a9 100644
--- a/tests/unit/action/test_print_event.py
+++ b/tests/unit/action/test_print_event.py
@@ -130,7 +130,8 @@ async def test_pretty_print_event(capsys):
             unpretty_action = event
 
     unpretty = capsys.readouterr()
-    assert len(unpretty.out.strip().splitlines()) == 1
+    # Three lines as a result of displaying in a banner.
+    assert len(unpretty.out.strip().splitlines()) == 3
 
     # The pretty-printed event.
     action_args = dict(pretty=True)
@@ -144,7 +145,7 @@ async def test_pretty_print_event(capsys):
             pretty_action = event
 
     pretty = capsys.readouterr()
-    assert len(pretty.out.strip().splitlines()) > 1
+    assert len(pretty.out.strip().splitlines()) > 3
 
     # Assert there was no difference in content.
     assert pretty_action == unpretty_action
diff --git a/tests/unit/test_terminal.py b/tests/unit/test_terminal.py
new file mode 100644
index 00000000..5d10cb7e
--- /dev/null
+++ b/tests/unit/test_terminal.py
@@ -0,0 +1,117 @@
+import copy
+import logging
+import os
+
+import pytest
+
+from ansible_rulebook import terminal
+
+
+@pytest.fixture
+def display():
+    # Use a copy of the Display singleton so tests that
+    # change the display level don't mess up other tests
+    # that expect the defaults.
+    return copy.copy(terminal.Display())
+
+
+def test_display_get_banners(display):
+    banner = "banner"
+    incomplete = os.linesep.join(
+        [
+            f"** 2023-12-12 12:34:56.123456 [{banner}] ******",
+            "line",
+            "line",
+        ]
+    )
+    complete = os.linesep.join(
+        [
+            incomplete,
+            "***********************************************",
+        ]
+    )
+
+    # Incomplete banner output.
+    with pytest.raises(terminal.DisplayBannerIncompleteError):
+        display.get_banners(banner, incomplete)
+
+    # Complete banner output.
+    result = display.get_banners(banner, complete)
+    assert len(result) == 1, "get complete banner failed"
+    assert result[0] == complete, "incorrect banner returned"
+
+    # Multiple banners.
+    multiple = os.linesep.join(
+        [
+            "line",
+            complete,
+            "line",
+            complete,
+            "line",
+        ]
+    )
+    result = display.get_banners(banner, multiple)
+    assert len(result) == 2, "get multiple banners failed"
+    assert result[0] == complete, "incorrect first banner returned"
+    assert result[1] == complete, "incorrect second banner returned"
+
+    # Specific banner from multiple.
+    specific = os.linesep.join(
+        [
+            "** 2023-12-12 12:34:56.123456 [specific] ******",
+            "line",
+            "line",
+            "***********************************************",
+        ]
+    )
+    result = display.get_banners(
+        "specific",
+        os.linesep.join(
+            [
+                "line",
+                complete,
+                specific,
+                complete,
+                "line",
+            ]
+        ),
+    )
+    assert len(result) == 1, "get specific banner failed"
+    assert result[0] == specific, "incorrect specific banner returned"
+
+
+def test_display_banner(display, capsys):
+    lines = ["line-one", "line-two", "line-three"]
+
+    # One-shot banner.
+    display.banner("banner", lines)
+    captured = capsys.readouterr()
+    result = display.get_banners("banner", captured.out)
+    assert result, "one-shot banner not found"
+
+    # Manually constructed banner.
+    display.banner("banner")
+    display.output(lines)
+    display.banner()
+    captured = capsys.readouterr()
+    result = display.get_banners("banner", captured.out)
+    assert result, "manual banner not found"
+
+    # Output at higher level.
+    display.banner("banner", lines, level=logging.WARNING)
+    captured = capsys.readouterr()
+    result = display.get_banners("banner", captured.out)
+    assert result, "banner not found"
+
+    # Output at changed level.
+    display.level = logging.WARNING
+    display.banner("banner", lines, level=logging.WARNING)
+    captured = capsys.readouterr()
+    result = display.get_banners("banner", captured.out)
+    assert result, "banner not found"
+
+    # No output for lower level.
+    display.banner("banner", lines)
+    captured = capsys.readouterr()
+    result = display.get_banners("banner", captured.out)
+    assert not result, "banner found"