python/jc
1
0
Fork 0
mirror of https://github.com/kellyjonbrazil/jc.git synced 2025-08-08 22:36:48 +02:00
Code Issues Releases Activity

raise if line data is unrecognized

Browse Source
This commit is contained in:
Kelly Brazil
2021-09-26 13:04:27 -07:00
parent ee075db598
commit 5044388ab2
2 changed files with 6 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
jc/parsers/vmstat_s.py
View File

@@ -74,6 +74,7 @@ Examples:
""" """
import jc.utils import jc.utils
from jc.utils import stream_success, stream_error from jc.utils import stream_success, stream_error
from jc.exceptions import ParseError
class info(): class info():
@@ -232,7 +233,10 @@ def parse(data, raw=False, quiet=False, ignore_exceptions=False):
'timezone': tz or None 'timezone': tz or None
} }
yield stream_success(output_line, ignore_exceptions) if raw else stream_success(_process(output_line), ignore_exceptions) if output_line:
yield stream_success(output_line, ignore_exceptions) if raw else stream_success(_process(output_line), ignore_exceptions)
else:
raise ParseError('Not vmstat data')
except Exception as e: except Exception as e:
yield stream_error(e, ignore_exceptions, line) yield stream_error(e, ignore_exceptions, line)

3
tests/test_vmstat_s.py
View File

@@ -71,7 +71,6 @@ class MyTests(unittest.TestCase):
with open(os.path.join(THIS_DIR, os.pardir, 'tests/fixtures/ubuntu-18.04/vmstat-1-long-streaming.json'), 'r', encoding='utf-8') as f: with open(os.path.join(THIS_DIR, os.pardir, 'tests/fixtures/ubuntu-18.04/vmstat-1-long-streaming.json'), 'r', encoding='utf-8') as f:
self.ubuntu_18_04_vmstat_1_long_streaming_json = json.loads(f.read()) self.ubuntu_18_04_vmstat_1_long_streaming_json = json.loads(f.read())
def test_vmstat_nodata(self): def test_vmstat_nodata(self):
""" """
Test 'vmstat' with no data Test 'vmstat' with no data
@@ -81,7 +80,7 @@ class MyTests(unittest.TestCase):
def test_vmstat_unparsable(self): def test_vmstat_unparsable(self):
data = 'unparsable data' data = 'unparsable data'
g = jc.parsers.vmstat_s.parse(data.splitlines(), quiet=True) g = jc.parsers.vmstat_s.parse(data.splitlines(), quiet=True)
with self.assertRaises(KeyError): with self.assertRaises(ParseError):
list(g) list(g)
def test_vmstat_centos_7_7(self): def test_vmstat_centos_7_7(self):