Skip to content
Snippets Groups Projects
Commit 2adee7e1 authored by Stéphane Diemer's avatar Stéphane Diemer
Browse files

Fixed wowza heap size retrieving, use single quotes in wowza test | refs #33327

parent 48383f5a
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
tests/scripts/test_wowza.py
+ 57
− 50
View file @ 2adee7e1
#!/usr/bin/env python3
"""
'''
Criticality: Normal
Checks that the streaming server (Wowza) is running correctly.
"""
'''
from pathlib import Path
import re
......@@ -20,15 +20,15 @@ sys.path.append(str(Path(__file__).parents[1].resolve()))
# pylint: disable=wrong-import-position
from utilities import logging as lg # noqa: E402
LATEST_VERSION = "4.8.5"
WOWZA_TUNE_FILE = "/usr/local/WowzaStreamingEngine/conf/Tune.xml"
LATEST_VERSION = '4.8.5'
WOWZA_TUNE_FILE = '/usr/local/WowzaStreamingEngine/conf/Tune.xml'
WOWZA_RECOMMENDED_HEAP_SIZE = 2000
def main():
"""Run all checks and exits with corresponding exit code."""
'''Run all checks and exits with corresponding exit code.'''
print("Checking Wowza settings:")
print('Checking Wowza settings:')
warnings = 0
errors = 0
......@@ -73,142 +73,149 @@ def main():
def check_installed() -> bool:
"""Check that Wowza is installed.
'''Check that Wowza is installed.
:return: Exit return codes
:rtype: bool
"""
'''
cmd = "dpkg --get-selections 'wowza*'"
cmd = 'dpkg --get-selections "wowza*"'
out = subprocess.getoutput(cmd)
state = out.split()[-1]
if state != "install":
lg.info("not installed, skip test")
if state != 'install':
lg.info('not installed, skip test')
return False
return True
def check_version() -> tuple:
"""Check the Wowza version installed.
'''Check the Wowza version installed.
:return: Exit return codes
:rtype: bool
"""
'''
warnings = 0
errors = 0
cmd = "dpkg --get-selections 'wowza*'"
cmd = 'dpkg --get-selections "wowza*"'
out = subprocess.getoutput(cmd)
version = None
for line in out.split("\n"):
if line.split()[-1] == "install":
for line in out.split('\n'):
if line.split()[-1] == 'install':
if version:
lg.error("many Wowza versions installed, keep only latest")
lg.error('many Wowza versions installed, keep only latest')
errors += 1
version = ".".join(re.findall(r"\d", line))
version = '.'.join(re.findall(r'\d', line))
if not version:
lg.error("cannot find Wowza version")
lg.error('cannot find Wowza version')
errors += 1
if parse_version(version) < parse_version(LATEST_VERSION):
lg.info(
"using outdated version: {0} < {1} (recommended)"
'using outdated version: {0} < {1} (recommended)'
.format(version, LATEST_VERSION)
)
elif parse_version(version) > parse_version(LATEST_VERSION):
lg.success(
"using newer version than the recommended: {0} > {1} (recommended)"
'using newer version than the recommended: {0} > {1} (recommended)'
.format(version, LATEST_VERSION)
)
else:
lg.success("using the recommended version: {0}".format(version))
lg.success('using the recommended version: {0}'.format(version))
return warnings, errors
def check_heap_size() -> tuple:
"""Check the heap size configured.
'''Check the heap size configured.
:return: Exit return codes
:rtype: bool
"""
'''
warnings = 0
errors = 0
# Configured wowza heap size extraction
tune_xml = etree.parse(WOWZA_TUNE_FILE)
heap_size = tune_xml.find('Tune/HeapSize').text[0:-1]
if int(heap_size) < WOWZA_RECOMMENDED_HEAP_SIZE:
lg.error(
"not using recommended heap size: {0}M < {1}M (recommended)"
.format(int(heap_size), WOWZA_RECOMMENDED_HEAP_SIZE)
try:
tune_xml = etree.parse(WOWZA_TUNE_FILE)
value = tune_xml.find('Tune/HeapSize').text[0:-1]
heap_size = int(value)
except Exception as e:
lg.info(
'failed to get heap size value: {0}'
.format(e)
)
errors += 1
else:
lg.success(
"using recommended heap size or above: {0}M"
.format(int(heap_size))
)
if heap_size < WOWZA_RECOMMENDED_HEAP_SIZE:
lg.error(
'not using recommended heap size: {0}M < {1}M (recommended)'
.format(heap_size, WOWZA_RECOMMENDED_HEAP_SIZE)
)
errors += 1
else:
lg.success(
'using recommended heap size or above: {0}M'
.format(heap_size)
)
return warnings, errors
def check_running() -> tuple:
"""Check that Wowza is running.
'''Check that Wowza is running.
:return: Exit return codes
:rtype: bool
"""
'''
warnings = 0
errors = 0
cmd = "systemctl status WowzaStreamingEngine"
cmd = 'systemctl status WowzaStreamingEngine'
out = subprocess.getoutput(cmd)
if "Active: active (running)" not in out:
lg.error("service not running")
if 'Active: active (running)' not in out:
lg.error('service not running')
errors += 1
else:
lg.success("service running")
lg.success('service running')
return warnings, errors
def check_listening() -> tuple:
"""Check that Wowza is listening on configured port.
'''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
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[1] for c in net_connections(kind="inet") if c.status == "LISTEN"
c.laddr[1] 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:
lg.error("not listening on port {}".format(port))
lg.error('not listening on port {}'.format(port))
errors += 1
else:
lg.success("listening on port {}".format(port))
lg.success('listening on port {}'.format(port))
return warnings, errors
if __name__ == "__main__":
if __name__ == '__main__':
main()
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment