mirror of
https://github.com/kellyjonbrazil/jc.git
synced 2025-06-17 00:07:37 +02:00
fix for no data
This commit is contained in:
@ -15,6 +15,8 @@ jc changelog
|
|||||||
- Update ntpq parser to fix error on parsing empty data
|
- Update ntpq parser to fix error on parsing empty data
|
||||||
- Update ps parser to fix error on parsing empty data
|
- Update ps parser to fix error on parsing empty data
|
||||||
- Update route parser to fix error on parsing empty data
|
- Update route parser to fix error on parsing empty data
|
||||||
|
- Update systemctl parser to fix error on parsing empty data
|
||||||
|
- Update systemctl_lj parser to fix error on parsing empty data
|
||||||
- Add tests to all parsers for no data condition
|
- Add tests to all parsers for no data condition
|
||||||
- Update ss parser to fix integer fields
|
- Update ss parser to fix integer fields
|
||||||
|
|
||||||
|
@ -40,7 +40,7 @@ import jc.utils
|
|||||||
|
|
||||||
|
|
||||||
class info():
|
class info():
|
||||||
version = '1.1'
|
version = '1.2'
|
||||||
description = 'systemctl command parser'
|
description = 'systemctl command parser'
|
||||||
author = 'Kelly Brazil'
|
author = 'Kelly Brazil'
|
||||||
author_email = 'kellyjonbrazil@gmail.com'
|
author_email = 'kellyjonbrazil@gmail.com'
|
||||||
@ -99,24 +99,27 @@ def parse(data, raw=False, quiet=False):
|
|||||||
linedata = data.splitlines()
|
linedata = data.splitlines()
|
||||||
# Clear any blank lines
|
# Clear any blank lines
|
||||||
linedata = list(filter(None, linedata))
|
linedata = list(filter(None, linedata))
|
||||||
# clean up non-ascii characters, if any
|
|
||||||
cleandata = []
|
|
||||||
for entry in linedata:
|
|
||||||
cleandata.append(entry.encode('ascii', errors='ignore').decode())
|
|
||||||
|
|
||||||
header_text = cleandata[0]
|
|
||||||
header_list = header_text.lower().split()
|
|
||||||
|
|
||||||
raw_output = []
|
raw_output = []
|
||||||
|
|
||||||
for entry in cleandata[1:]:
|
if linedata:
|
||||||
if 'LOAD = ' in entry:
|
# clean up non-ascii characters, if any
|
||||||
break
|
cleandata = []
|
||||||
|
for entry in linedata:
|
||||||
|
cleandata.append(entry.encode('ascii', errors='ignore').decode())
|
||||||
|
|
||||||
else:
|
header_text = cleandata[0]
|
||||||
entry_list = entry.rstrip().split(maxsplit=4)
|
header_list = header_text.lower().split()
|
||||||
output_line = dict(zip(header_list, entry_list))
|
|
||||||
raw_output.append(output_line)
|
raw_output = []
|
||||||
|
|
||||||
|
for entry in cleandata[1:]:
|
||||||
|
if 'LOAD = ' in entry:
|
||||||
|
break
|
||||||
|
|
||||||
|
else:
|
||||||
|
entry_list = entry.rstrip().split(maxsplit=4)
|
||||||
|
output_line = dict(zip(header_list, entry_list))
|
||||||
|
raw_output.append(output_line)
|
||||||
|
|
||||||
if raw:
|
if raw:
|
||||||
return raw_output
|
return raw_output
|
||||||
|
@ -59,7 +59,7 @@ import jc.utils
|
|||||||
|
|
||||||
|
|
||||||
class info():
|
class info():
|
||||||
version = '1.1'
|
version = '1.2'
|
||||||
description = 'systemctl list-jobs command parser'
|
description = 'systemctl list-jobs command parser'
|
||||||
author = 'Kelly Brazil'
|
author = 'Kelly Brazil'
|
||||||
author_email = 'kellyjonbrazil@gmail.com'
|
author_email = 'kellyjonbrazil@gmail.com'
|
||||||
@ -126,24 +126,27 @@ def parse(data, raw=False, quiet=False):
|
|||||||
# Clear any blank lines
|
# Clear any blank lines
|
||||||
linedata = list(filter(None, linedata))
|
linedata = list(filter(None, linedata))
|
||||||
# clean up non-ascii characters, if any
|
# clean up non-ascii characters, if any
|
||||||
cleandata = []
|
|
||||||
for entry in linedata:
|
|
||||||
cleandata.append(entry.encode('ascii', errors='ignore').decode())
|
|
||||||
|
|
||||||
header_text = cleandata[0]
|
|
||||||
header_text = header_text.lower()
|
|
||||||
header_list = header_text.split()
|
|
||||||
|
|
||||||
raw_output = []
|
raw_output = []
|
||||||
|
|
||||||
for entry in cleandata[1:]:
|
if linedata:
|
||||||
if 'No jobs running.' in entry or 'jobs listed.' in entry:
|
cleandata = []
|
||||||
break
|
for entry in linedata:
|
||||||
|
cleandata.append(entry.encode('ascii', errors='ignore').decode())
|
||||||
|
|
||||||
else:
|
header_text = cleandata[0]
|
||||||
entry_list = entry.split(maxsplit=4)
|
header_text = header_text.lower()
|
||||||
output_line = dict(zip(header_list, entry_list))
|
header_list = header_text.split()
|
||||||
raw_output.append(output_line)
|
|
||||||
|
raw_output = []
|
||||||
|
|
||||||
|
for entry in cleandata[1:]:
|
||||||
|
if 'No jobs running.' in entry or 'jobs listed.' in entry:
|
||||||
|
break
|
||||||
|
|
||||||
|
else:
|
||||||
|
entry_list = entry.split(maxsplit=4)
|
||||||
|
output_line = dict(zip(header_list, entry_list))
|
||||||
|
raw_output.append(output_line)
|
||||||
|
|
||||||
if raw:
|
if raw:
|
||||||
return raw_output
|
return raw_output
|
||||||
|
@ -35,6 +35,12 @@ class MyTests(unittest.TestCase):
|
|||||||
with open(os.path.join(THIS_DIR, os.pardir, 'tests/fixtures/freebsd12/stat.json'), 'r', encoding='utf-8') as f:
|
with open(os.path.join(THIS_DIR, os.pardir, 'tests/fixtures/freebsd12/stat.json'), 'r', encoding='utf-8') as f:
|
||||||
self.freebsd12_stat_json = json.loads(f.read())
|
self.freebsd12_stat_json = json.loads(f.read())
|
||||||
|
|
||||||
|
def test_stat_nodata(self):
|
||||||
|
"""
|
||||||
|
Test 'stat' with no data
|
||||||
|
"""
|
||||||
|
self.assertEqual(jc.parsers.stat.parse('', quiet=True), [])
|
||||||
|
|
||||||
def test_stat_centos_7_7(self):
|
def test_stat_centos_7_7(self):
|
||||||
"""
|
"""
|
||||||
Test 'stat /bin/*' on Centos 7.7
|
Test 'stat /bin/*' on Centos 7.7
|
||||||
|
@ -23,6 +23,12 @@ class MyTests(unittest.TestCase):
|
|||||||
with open(os.path.join(THIS_DIR, os.pardir, 'tests/fixtures/ubuntu-18.04/systemctl.json'), 'r', encoding='utf-8') as f:
|
with open(os.path.join(THIS_DIR, os.pardir, 'tests/fixtures/ubuntu-18.04/systemctl.json'), 'r', encoding='utf-8') as f:
|
||||||
self.ubuntu_18_4_systemctl_json = json.loads(f.read())
|
self.ubuntu_18_4_systemctl_json = json.loads(f.read())
|
||||||
|
|
||||||
|
def test_systemctl_nodata(self):
|
||||||
|
"""
|
||||||
|
Test 'systemctl' with no data
|
||||||
|
"""
|
||||||
|
self.assertEqual(jc.parsers.systemctl.parse('', quiet=True), [])
|
||||||
|
|
||||||
def test_systemctl_centos_7_7(self):
|
def test_systemctl_centos_7_7(self):
|
||||||
"""
|
"""
|
||||||
Test 'systemctl -a' on Centos 7.7
|
Test 'systemctl -a' on Centos 7.7
|
||||||
|
@ -17,6 +17,12 @@ class MyTests(unittest.TestCase):
|
|||||||
with open(os.path.join(THIS_DIR, os.pardir, 'tests/fixtures/ubuntu-18.04/systemctl-lj.json'), 'r', encoding='utf-8') as f:
|
with open(os.path.join(THIS_DIR, os.pardir, 'tests/fixtures/ubuntu-18.04/systemctl-lj.json'), 'r', encoding='utf-8') as f:
|
||||||
self.ubuntu_18_4_systemctl_lj_json = json.loads(f.read())
|
self.ubuntu_18_4_systemctl_lj_json = json.loads(f.read())
|
||||||
|
|
||||||
|
def test_systemctl_lj_nodata(self):
|
||||||
|
"""
|
||||||
|
Test 'systemctl -a list-jobs' with no data
|
||||||
|
"""
|
||||||
|
self.assertEqual(jc.parsers.systemctl_lj.parse('', quiet=True), [])
|
||||||
|
|
||||||
def test_systemctl_lj_ubuntu_18_4(self):
|
def test_systemctl_lj_ubuntu_18_4(self):
|
||||||
"""
|
"""
|
||||||
Test 'systemctl -a list-jobs' on Ubuntu 18.4
|
Test 'systemctl -a list-jobs' on Ubuntu 18.4
|
||||||
|
Reference in New Issue
Block a user