check that wowza listen on configured port | refs #28832

pkgs_envsetup.py

@@ -2,7 +2,13 @@
 
 from subprocess import call, DEVNULL
 
-PACKAGES = ["python3-openssl", "python3-psutil", "python3-requests", "python3-spf"]
+PACKAGES = [
+    "python3-defusedxml",
+    "python3-openssl",
+    "python3-psutil",
+    "python3-requests",
+    "python3-spf",
+]
 
 
 def main():

tests/test_wowza.py

@@ -10,6 +10,9 @@ import re
 import subprocess  # nosec: B404
 import sys
 
+from defusedxml.ElementTree import parse
+from psutil import net_connections
+
 sys.path.append(str(Path(__file__).parents[1].resolve()))
 
 # pylint: disable=wrong-import-position
@@ -50,6 +53,13 @@ def main():
     elif check_warn:
         warnings += 1
 
+    # check that wowza is listening
+    check_warn, check_err = check_listening()
+    if check_err:
+        errors += 1
+    elif check_warn:
+        warnings += 1
+
     if errors:
         exit(1)
     elif warnings:
@@ -152,5 +162,34 @@ def check_running() -> tuple:
     return warnings, errors
 
 
+def check_listening() -> tuple:
+    """Check that Wowza is listening on configured port.
+
+    :return: Exit return codes
+    :rtype: bool
+    """
+
+    warnings = 0
+    errors = 0
+
+    # get port number in Wowza config
+    conf = parse("/usr/local/WowzaStreamingEngine/conf/VHost.xml").getroot()
+    port = conf.findall("VHost/HostPortList/HostPort/Port")[0].text
+
+    # get listening ports
+    listening = set(
+        c.laddr.port for c in net_connections(kind="inet") if c.status == "LISTEN"
+    )
+
+    # check that system is listening on this port
+    if int(port) not in listening:
+        u.error("not listening on port {}".format(port))
+        errors += 1
+    else:
+        u.success("listening on port {}".format(port))
+
+    return warnings, errors
+
+
 if __name__ == "__main__":
     main()