Merge with upstream

Author: jwlodek

LICENSE

@@ -0,0 +1,28 @@
+BSD 3-Clause License
+
+Copyright (c) 2025, Brookhaven National Laboratory
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice, this
+   list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright notice,
+   this list of conditions and the following disclaimer in the documentation
+   and/or other materials provided with the distribution.
+
+3. Neither the name of the copyright holder nor the names of its
+   contributors may be used to endorse or promote products derived from
+   this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

src/manage_iocs/commands.py

@@ -1,11 +1,15 @@
 import inspect
 import socket
 import sys
+import time as ttime
 from subprocess import PIPE, Popen
 
 from . import __version__, utils
 
-EXTRA_PAD_WIDTH = 3
+EXTRA_PAD_WIDTH = 5
+
+# Length added to string to account for ANSI color escape codes
+ANSI_COLOR_ESC_CODE_LEN = 9
 
 
 def version():
@@ -50,23 +54,38 @@ def attach(ioc: str):
 
 def report():
     """Show config(s) of an all IOCs on localhost"""
+    base_hostname = socket.gethostname()
+    if "." in base_hostname:
+        base_hostname = base_hostname.split(".")[0]
+
     iocs = [
         ioc_config
         for ioc_config in utils.find_iocs().values()
-        if ioc_config.host == "localhost" or ioc_config.host == socket.gethostname()
+        if ioc_config.host == "localhost"
+        or ioc_config.host == base_hostname
+        or ioc_config.host == socket.gethostname()
     ]
+
+    if len(iocs) == 0:
+        print("No IOCs found on configured to run on this host.")
+        print(f"Searched in: {utils.IOC_SEARCH_PATH}")
+        return 1
+
     max_base_len = max(len(str(ioc.path)) for ioc in iocs) + EXTRA_PAD_WIDTH
     max_ioc_name_len = max(len(ioc.name) for ioc in iocs) + EXTRA_PAD_WIDTH
     max_user_len = max(len(ioc.user) for ioc in iocs) + EXTRA_PAD_WIDTH
     max_port_len = max(len(str(ioc.procserv_port)) for ioc in iocs) + EXTRA_PAD_WIDTH
-    max_exec_len = max(len(ioc.exec_path) for ioc in iocs) + max_base_len
+    max_exec_len = max(len(ioc.exec_path) for ioc in iocs) + max_base_len - EXTRA_PAD_WIDTH
 
-    print(
+    header = (
         f"{'BASE'.ljust(max_base_len)}| {'IOC'.ljust(max_ioc_name_len)}| "
         f"{'USER'.ljust(max_user_len)}| {'PORT'.ljust(max_port_len)}| "
         f"{'EXEC'.ljust(max_exec_len)}"
     )
+    print(header)
+    print("-" * len(header))
     for ioc in iocs:
+        ttime.sleep(0.01)
         print(
             f"{str(ioc.path).ljust(max_base_len)}| {ioc.name.ljust(max_ioc_name_len)}| "
             f"{ioc.user.ljust(max_user_len)}| {str(ioc.procserv_port).ljust(max_port_len)}| "
@@ -197,8 +216,11 @@ def install(ioc: str):
 
     service_file = utils.SYSTEMD_SERVICE_PATH / f"softioc-{ioc}.service"
     ioc_config = utils.find_iocs()[ioc]
+    base_hostname = socket.gethostname()
+    if "." in base_hostname:
+        base_hostname = base_hostname.split(".")[0]
 
-    if socket.gethostname() != ioc_config.host and ioc_config.host != "localhost":
+    if ioc_config.host not in [base_hostname, "localhost", socket.gethostname()]:
         raise RuntimeError(
             f"Cannot install IOC '{ioc}' on this host; configured host is '{ioc_config.host}'!"
         )
@@ -245,21 +267,35 @@ def status():
     """Get the status of the given IOC."""
 
     ret = 0
-    statuses: dict[str, tuple[str, str]] = {}
+    statuses: dict[str, tuple[str, bool]] = {}
     installed_iocs = utils.find_installed_iocs().keys()
+    if len(installed_iocs) == 0:
+        print("No Installed IOCs found on this host.")
+        return 1
 
     for installed_ioc in installed_iocs:
-        status_ret, status = utils.get_ioc_statuses(installed_ioc)
-        ret += status_ret  # TODO: Log warning
-        statuses[installed_ioc] = status
+        try:
+            statuses[installed_ioc] = utils.get_ioc_status(installed_ioc)
+        except RuntimeError:
+            pass  # TODO: Handle this better?
 
     max_ioc_name_len = max(len(ioc_name) for ioc_name in statuses.keys()) + EXTRA_PAD_WIDTH
     max_status_len = max(len(status[0]) for status in statuses.values()) + EXTRA_PAD_WIDTH
-    max_enabled_len = max(len(status[1]) for status in statuses.values()) + EXTRA_PAD_WIDTH
+    max_enabled_len = len("Auto-Start")
 
     print(f"{'IOC'.ljust(max_ioc_name_len)}{'Status'.ljust(max_status_len)}Auto-Start")
+    ttime.sleep(0.01)
     print("-" * (max_ioc_name_len + max_status_len + max_enabled_len))
-    for ioc_name, (status_str, enabled_str) in statuses.items():
-        print(f"{ioc_name.ljust(max_ioc_name_len)}{status_str.ljust(max_status_len)}{enabled_str}")
+    for ioc_name, (state, is_enabled) in statuses.items():
+        ttime.sleep(0.01)
+        if state == "Running":
+            state_str = f"\033[92m{state}\033[0m"  # Green
+        elif state == "Stopped":
+            state_str = f"\033[91m{state}\033[0m"  # Red
+        else:
+            state_str = f"\033[93m{state}\033[0m"  # Yellow
+        print(
+            f"{ioc_name.ljust(max_ioc_name_len)}{state_str.ljust(max_status_len + ANSI_COLOR_ESC_CODE_LEN)}{'Enabled' if is_enabled else 'Disabled'}"  # noqa: E501
+        )
 
     return ret

src/manage_iocs/utils.py

@@ -5,12 +5,12 @@
 from pathlib import Path
 from subprocess import PIPE, Popen
 
-IOC_SEARCH_PATH = (
-    [Path("/epics/iocs"), Path("/opt/epics/iocs"), Path("/opt/iocs")]
-    + [Path(p) for p in os.environ["MANAGE_IOCS_SEARCH_PATH"].split(os.pathsep)]
-    if "MANAGE_IOCS_SEARCH_PATH" in os.environ
-    else []
-)
+IOC_SEARCH_PATH = [Path("/epics/iocs"), Path("/opt/epics/iocs"), Path("/opt/iocs")]
+if "MANAGE_IOCS_SEARCH_PATH" in os.environ:
+    IOC_SEARCH_PATH.extend(
+        [Path(p) for p in os.environ["MANAGE_IOCS_SEARCH_PATH"].split(os.pathsep)]
+    )
+
 SYSTEMD_SERVICE_PATH = Path("/etc/systemd/system")
 
 
@@ -90,24 +90,27 @@ def systemctl_passthrough(action: str, ioc: str) -> tuple[str, str, int]:
     """Helper to call systemctl with the given action and IOC name."""
     proc = Popen(["systemctl", action, f"softioc-{ioc}.service"], stdin=PIPE, stdout=PIPE)
     out, err = proc.communicate()
-    return out.decode().strip(), err.decode().strip(), proc.returncode
+    decoded_out = out.decode().strip() if out else ""
+    decoded_err = err.decode().strip() if err else ""
+    return decoded_out, decoded_err, proc.returncode
 
 
-def get_ioc_statuses(ioc_name: str) -> tuple[int, tuple[str, str]]:
+def get_ioc_status(ioc_name: str) -> tuple[str, bool]:
     """Get the active and enabled status of the given IOC."""
 
-    status, _, ret = systemctl_passthrough("is-active", ioc_name)
-    if status == "inactive":
-        status = "Stopped"
-    elif status == "active":
-        status = "Running"
-    else:
-        status = status.capitalize()
+    state, err, _ = systemctl_passthrough("is-active", ioc_name)
+
+    # Convert to more user-friendly terms
+    if state == "active":
+        state = "Running"
+    elif state == "inactive":
+        state = "Stopped"
 
-    enabled, _, ret_enable = systemctl_passthrough("is-enabled", ioc_name)
-    ret = ret + ret_enable
+    enabled, err, _ = systemctl_passthrough("is-enabled", ioc_name)
+    if enabled not in ("enabled", "disabled"):
+        raise RuntimeError(err)
 
-    return ret, (status, enabled.capitalize())
+    return state.capitalize(), enabled == "enabled"
 
 
 def requires_root(func: Callable):

tests/test_commands.py

@@ -5,7 +5,14 @@
 import manage_iocs
 import manage_iocs.commands as cmds
 import manage_iocs.utils
-from manage_iocs.utils import find_installed_iocs, get_ioc_statuses
+from manage_iocs.utils import find_installed_iocs, get_ioc_status
+
+
+def strip_ansi_codes(s: str) -> str:
+    import re
+
+    ansi_escape = re.compile(r"\x1B\[[0-?]*[ -/]*[@-~]")
+    return ansi_escape.sub("", s)
 
 
 def test_version(capsys):
@@ -44,22 +51,35 @@ def normalize_whitespace(s: str) -> str:
         assert normalize_whitespace(line) in normalize_whitespace(captured.out)
 
 
+def test_report_no_iocs(monkeypatch, capsys):
+    monkeypatch.setattr(
+        manage_iocs.utils,
+        "find_iocs",
+        lambda: {},
+    )
+
+    rc = cmds.report()
+    captured = capsys.readouterr()
+    assert "No IOCs found on configured to run on this host." in captured.out
+    assert rc == 1
+
+
 @pytest.mark.parametrize(
     "ioc_name, command, before_state, before_enabled, after_state, after_enabled, as_root",
     [
-        ("ioc1", cmds.stop, "Running", "Enabled", "Stopped", "Enabled", False),
-        ("ioc3", cmds.stop, "Running", "Disabled", "Stopped", "Disabled", False),
-        ("ioc4", cmds.stop, "Stopped", "Disabled", "Stopped", "Disabled", False),
-        ("ioc1", cmds.start, "Running", "Enabled", "Running", "Enabled", False),
-        ("ioc3", cmds.start, "Running", "Disabled", "Running", "Disabled", False),
-        ("ioc4", cmds.start, "Stopped", "Disabled", "Running", "Disabled", False),
-        ("ioc3", cmds.enable, "Running", "Disabled", "Running", "Enabled", True),
-        ("ioc4", cmds.enable, "Stopped", "Disabled", "Stopped", "Enabled", True),
-        ("ioc1", cmds.disable, "Running", "Enabled", "Running", "Disabled", True),
-        ("ioc3", cmds.disable, "Running", "Disabled", "Running", "Disabled", True),
-        ("ioc1", cmds.restart, "Running", "Enabled", "Running", "Enabled", False),
-        ("ioc4", cmds.restart, "Stopped", "Disabled", "Running", "Disabled", False),
-        ("ioc3", cmds.restart, "Running", "Disabled", "Running", "Disabled", False),
+        ("ioc1", cmds.stop, "Running", True, "Stopped", True, False),
+        ("ioc3", cmds.stop, "Running", False, "Stopped", False, False),
+        ("ioc4", cmds.stop, "Stopped", False, "Stopped", False, False),
+        ("ioc1", cmds.start, "Running", True, "Running", True, False),
+        ("ioc3", cmds.start, "Running", False, "Running", False, False),
+        ("ioc4", cmds.start, "Stopped", False, "Running", False, False),
+        ("ioc3", cmds.enable, "Running", False, "Running", True, True),
+        ("ioc4", cmds.enable, "Stopped", False, "Stopped", True, True),
+        ("ioc1", cmds.disable, "Running", True, "Running", False, True),
+        ("ioc3", cmds.disable, "Running", False, "Running", False, True),
+        ("ioc1", cmds.restart, "Running", True, "Running", True, False),
+        ("ioc4", cmds.restart, "Stopped", False, "Running", False, False),
+        ("ioc3", cmds.restart, "Running", False, "Running", False, False),
     ],
 )
 def test_state_change_commands(
@@ -76,14 +96,12 @@ def test_state_change_commands(
     if not as_root:
         monkeypatch.setattr(os, "geteuid", lambda: 1000)  # Mock as non-root user
 
-    _, status = get_ioc_statuses(ioc_name)
-    assert status == (before_state, before_enabled)
+    assert get_ioc_status(ioc_name) == (before_state, before_enabled)
 
     rc = command(ioc_name)
     assert rc == 0
 
-    _, status = get_ioc_statuses(ioc_name)
-    assert status == (after_state, after_enabled)
+    assert get_ioc_status(ioc_name) == (after_state, after_enabled)
 
 
 def test_install_new_ioc(sample_iocs, monkeypatch):
@@ -92,8 +110,7 @@ def test_install_new_ioc(sample_iocs, monkeypatch):
     rc = cmds.install("ioc2")
     assert rc == 0
 
-    _, status = get_ioc_statuses("ioc2")
-    assert status == ("Stopped", "Disabled")
+    assert get_ioc_status("ioc2") == ("Stopped", False)
 
     assert "ioc2" in find_installed_iocs()
 
@@ -135,8 +152,8 @@ def test_requires_root(sample_iocs, monkeypatch, command):
     [
         (cmds.startall, "Running", None),
         (cmds.stopall, "Stopped", None),
-        (cmds.enableall, None, "Enabled"),
-        (cmds.disableall, None, "Disabled"),
+        (cmds.enableall, None, True),
+        (cmds.disableall, None, False),
     ],
 )
 def test_state_change_all(sample_iocs, cmd, expected_state, expected_enabled):
@@ -150,9 +167,9 @@ def test_state_change_all(sample_iocs, cmd, expected_state, expected_enabled):
     # Check all are running
     for ioc in installed_iocs.values():
         if expected_state is not None:
-            assert get_ioc_statuses(ioc.name)[1][0] == expected_state
+            assert get_ioc_status(ioc.name)[0] == expected_state
         if expected_enabled is not None:
-            assert get_ioc_statuses(ioc.name)[1][1] == expected_enabled
+            assert get_ioc_status(ioc.name)[1] is expected_enabled
 
 
 @pytest.mark.parametrize("as_root", [True, False])
@@ -173,22 +190,38 @@ def test_status(sample_iocs, capsys, monkeypatch, as_root):
     rc = cmds.status()
     captured = capsys.readouterr()
     expected_output = """IOC Status Auto-Start
-----------------------------
+--------------------------
 ioc1 Running Enabled
 ioc3 Running Disabled
 ioc4 Stopped Disabled
 ioc5 Stopped Enabled
 """
 
-    def normalize_whitespace(s: str) -> str:
-        return "\n".join(" ".join(line.split()) for line in s.strip().splitlines())
+    def normalize_whitespace_and_ansi_codes(s: str) -> str:
+        whitespace_normalized = "\n".join(" ".join(line.split()) for line in s.strip().splitlines())
+        return strip_ansi_codes(whitespace_normalized)
 
     for line in expected_output.strip().splitlines():
-        assert normalize_whitespace(line) in normalize_whitespace(captured.out)
+        assert normalize_whitespace_and_ansi_codes(line) in normalize_whitespace_and_ansi_codes(
+            captured.out
+        )
 
     assert rc == 0
 
 
+def test_status_no_installed_iocs(sample_iocs, monkeypatch, capsys):
+    monkeypatch.setattr(
+        manage_iocs.utils,
+        "find_installed_iocs",
+        lambda: {},
+    )
+
+    rc = cmds.status()
+    captured = capsys.readouterr()
+    assert "No Installed IOCs found on this host." in captured.out
+    assert rc == 1
+
+
 @pytest.mark.parametrize(
     "cmd, expected_message",
     [