python/jc
1
0
Fork 0
mirror of https://github.com/kellyjonbrazil/jc.git synced 2025-06-17 00:07:37 +02:00
Code Issues Releases Activity

use -qq to ignore streaming exceptions

Browse Source
This commit is contained in:
Kelly Brazil
2021-09-23 11:48:39 -07:00
parent 9c6c6c4330
commit 98a7686db4
18 changed files with 160 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -191,7 +191,7 @@ The JSON output can be compact (default) or pretty formatted with the `-p` optio
- `-h` help. Use `jc -h --parser_name` for parser documentation
- `-m` monochrome JSON output
- `-p` pretty format the JSON output
- `-q` quiet mode. Suppresses parser warning messages
- `-q` quiet mode. Suppresses parser warning messages (use `-qq` to ignore streaming parser errors)
- `-r` raw output. Provides a more literal JSON output, typically with string values and no additional semantic processing
- `-u` unbuffer output
- `-v` version information

5
docs/parsers/ls_s.md
View File

@ -66,7 +66,7 @@ Provides parser metadata (version, author, etc.)
## parse
```python
parse(data, raw=False, quiet=False)
parse(data, raw=False, quiet=False, ignore_exceptions=False)
```
Main text parsing generator function. Returns an iterator object.
@ -75,7 +75,8 @@ Parameters:
data: (iterable) line-based text data to parse (e.g. sys.stdin or str.splitlines())
raw: (boolean) output preprocessed JSON if True
quiet: (boolean) suppress warning messages and ignore parsing exceptions if True
quiet: (boolean) suppress warning messages if True
ignore_exceptions: (boolean) ignore parsing exceptions if True
Yields:

5
docs/parsers/ping_s.md
View File

@ -73,7 +73,7 @@ Provides parser metadata (version, author, etc.)
## parse
```python
parse(data, raw=False, quiet=False)
parse(data, raw=False, quiet=False, ignore_exceptions=False)
```
Main text parsing generator function. Returns an iterator object.
@ -82,7 +82,8 @@ Parameters:
data: (iterable) line-based text data to parse (e.g. sys.stdin or str.splitlines())
raw: (boolean) output preprocessed JSON if True
quiet: (boolean) suppress warning messages and ignore parsing exceptions if True
quiet: (boolean) suppress warning messages if True
ignore_exceptions: (boolean) ignore parsing exceptions if True
Yields:

5
docs/parsers/vmstat_s.md
View File

@ -80,7 +80,7 @@ Provides parser metadata (version, author, etc.)
## parse
```python
parse(data, raw=False, quiet=False)
parse(data, raw=False, quiet=False, ignore_exceptions=False)
```
Main text parsing generator function. Returns an iterator object.
@ -89,7 +89,8 @@ Parameters:
data: (iterable) line-based text data to parse (e.g. sys.stdin or str.splitlines())
raw: (boolean) output preprocessed JSON if True
quiet: (boolean) suppress warning messages and ignore parsing exceptions if True
quiet: (boolean) suppress warning messages if True
ignore_exceptions: (boolean) ignore parsing exceptions if True
Yields:

6
docs/utils.md
View File

@ -119,15 +119,15 @@ Returns:
## stream_success
```python
stream_success(output_line, quiet)
stream_success(output_line, ignore_exceptions)
```
add _meta object to output line if -q (quiet) option is used
## stream_error
```python
stream_error(e, quiet, line)
stream_error(e, ignore_exceptions, line)
```
reraise the stream exception with annotation or print an error _meta field if quiet=True
reraise the stream exception with annotation or print an error _meta field if ignore_exceptions=True
## timestamp
```python

7
jc/cli.py
View File

@ -335,7 +335,7 @@ def helptext():
-h help (-h --parser_name for parser documentation)
-m monochrome output
-p pretty print output
-q quiet - suppress parser warnings
-q quiet - suppress parser warnings (-qq to ignore streaming errors)
-r raw JSON output
-u unbuffer output
-v version info
@ -528,6 +528,7 @@ def main():
help_me = 'h' in options
pretty = 'p' in options
quiet = 'q' in options
ignore_exceptions = True if options.count('q') > 1 else False
raw = 'r' in options
unbuffer = 'u' in options
version_info = 'v' in options
@ -623,7 +624,7 @@ def main():
# streaming
if getattr(parser.info, 'streaming', None):
result = parser.parse(sys.stdin, raw=raw, quiet=quiet)
result = parser.parse(sys.stdin, raw=raw, quiet=quiet, ignore_exceptions=ignore_exceptions)
for line in result:
print(json_out(line,
pretty=pretty,
@ -672,7 +673,7 @@ def main():
else:
streaming_msg = ''
if getattr(parser.info, 'streaming', None):
streaming_msg = ' Use the -q option to ignore streaming parser errors.\n'
streaming_msg = ' Use the -qq option to ignore streaming parser errors.\n'
jc.utils.error_message(
f'{parser_name} parser could not parse the input data. Did you use the correct parser?\n'

9
jc/parsers/foo_s.py
View File

@ -75,7 +75,7 @@ def _process(proc_data):
return proc_data
def parse(data, raw=False, quiet=False):
def parse(data, raw=False, quiet=False, ignore_exceptions=False):
"""
Main text parsing generator function. Returns an iterator object.
@ -83,7 +83,8 @@ def parse(data, raw=False, quiet=False):
data: (iterable) line-based text data to parse (e.g. sys.stdin or str.splitlines())
raw: (boolean) output preprocessed JSON if True
quiet: (boolean) suppress warning messages and ignore parsing exceptions if True
quiet: (boolean) suppress warning messages if True
ignore_exceptions: (boolean) ignore parsing exceptions if True
Yields:
@ -102,7 +103,7 @@ def parse(data, raw=False, quiet=False):
# parse the input here
#
yield stream_success(output_line, quiet) if raw else stream_success(_process(output_line), quiet)
yield stream_success(output_line, ignore_exceptions) if raw else stream_success(_process(output_line), ignore_exceptions)
except Exception as e:
yield stream_error(e, quiet, line)
yield stream_error(e, ignore_exceptions, line)

9
jc/parsers/ls_s.py
View File

@ -102,7 +102,7 @@ def _process(proc_data):
return proc_data
def parse(data, raw=False, quiet=False):
def parse(data, raw=False, quiet=False, ignore_exceptions=False):
"""
Main text parsing generator function. Returns an iterator object.
@ -110,7 +110,8 @@ def parse(data, raw=False, quiet=False):
data: (iterable) line-based text data to parse (e.g. sys.stdin or str.splitlines())
raw: (boolean) output preprocessed JSON if True
quiet: (boolean) suppress warning messages and ignore parsing exceptions if True
quiet: (boolean) suppress warning messages if True
ignore_exceptions: (boolean) ignore parsing exceptions if True
Yields:
@ -171,7 +172,7 @@ def parse(data, raw=False, quiet=False):
output_line['size'] = parsed_line[4]
output_line['date'] = ' '.join(parsed_line[5:8])
yield stream_success(output_line, quiet) if raw else stream_success(_process(output_line), quiet)
yield stream_success(output_line, ignore_exceptions) if raw else stream_success(_process(output_line), ignore_exceptions)
except Exception as e:
yield stream_error(e, quiet, line)
yield stream_error(e, ignore_exceptions, line)

9
jc/parsers/ping_s.py
View File

@ -450,7 +450,7 @@ def _linux_parse(line, s):
return output_line
def parse(data, raw=False, quiet=False):
def parse(data, raw=False, quiet=False, ignore_exceptions=False):
"""
Main text parsing generator function. Returns an iterator object.
@ -458,7 +458,8 @@ def parse(data, raw=False, quiet=False):
data: (iterable) line-based text data to parse (e.g. sys.stdin or str.splitlines())
raw: (boolean) output preprocessed JSON if True
quiet: (boolean) suppress warning messages and ignore parsing exceptions if True
quiet: (boolean) suppress warning messages if True
ignore_exceptions: (boolean) ignore parsing exceptions if True
Yields:
@ -520,9 +521,9 @@ def parse(data, raw=False, quiet=False):
# yield the output line if it has data
if output_line:
yield stream_success(output_line, quiet) if raw else stream_success(_process(output_line), quiet)
yield stream_success(output_line, ignore_exceptions) if raw else stream_success(_process(output_line), ignore_exceptions)
else:
continue
except Exception as e:
yield stream_error(e, quiet, line)
yield stream_error(e, ignore_exceptions, line)

9
jc/parsers/vmstat_s.py
View File

@ -117,7 +117,7 @@ def _process(proc_data):
return proc_data
def parse(data, raw=False, quiet=False):
def parse(data, raw=False, quiet=False, ignore_exceptions=False):
"""
Main text parsing generator function. Returns an iterator object.
@ -125,7 +125,8 @@ def parse(data, raw=False, quiet=False):
data: (iterable) line-based text data to parse (e.g. sys.stdin or str.splitlines())
raw: (boolean) output preprocessed JSON if True
quiet: (boolean) suppress warning messages and ignore parsing exceptions if True
quiet: (boolean) suppress warning messages if True
ignore_exceptions: (boolean) ignore parsing exceptions if True
Yields:
@ -225,7 +226,7 @@ def parse(data, raw=False, quiet=False):
'timezone': tz or None
}
yield stream_success(output_line, quiet) if raw else stream_success(_process(output_line), quiet)
yield stream_success(output_line, ignore_exceptions) if raw else stream_success(_process(output_line), ignore_exceptions)
except Exception as e:
yield stream_error(e, quiet, line)
yield stream_error(e, ignore_exceptions, line)

12
jc/utils.py
View File

@ -173,18 +173,18 @@ def convert_to_bool(value):
return False
def stream_success(output_line, quiet):
def stream_success(output_line, ignore_exceptions):
"""add _meta object to output line if -q (quiet) option is used"""
if quiet:
if ignore_exceptions:
output_line.update({'_meta': {'success': True}})
return output_line
def stream_error(e, quiet, line):
"""reraise the stream exception with annotation or print an error _meta field if quiet=True"""
if not quiet:
e.args = (str(e) + '... Use the quiet option (-q) to ignore streaming parser errors.',)
def stream_error(e, ignore_exceptions, line):
"""reraise the stream exception with annotation or print an error _meta field if ignore_exceptions=True"""
if not ignore_exceptions:
e.args = (str(e) + '... Use the ignore_exceptions option (-qq) to ignore streaming parser errors.',)
raise e
else:
return {

4
man/jc.1
View File

@ -421,7 +421,7 @@ about jc (JSON output)
.TP
.B
\fB-d\fP
debug - show traceback (\fB-dd\fP for verbose traceback)
debug - show traceback (use \fB-dd\fP for verbose traceback)
.TP
.B
\fB-h\fP
@ -437,7 +437,7 @@ pretty print output
.TP
.B
\fB-q\fP
quiet - suppress warnings
quiet - suppress warnings (use \fB-qq\fP to ignore streaming parser errors)
.TP
.B
\fB-r\fP

4
templates/manpage_template
View File

@ -36,7 +36,7 @@ about jc (JSON output)
.TP
.B
\fB-d\fP
debug - show traceback (\fB-dd\fP for verbose traceback)
debug - show traceback (use \fB-dd\fP for verbose traceback)
.TP
.B
\fB-h\fP
@ -52,7 +52,7 @@ pretty print output
.TP
.B
\fB-q\fP
quiet - suppress warnings
quiet - suppress warnings (use \fB-qq\fP to ignore streaming parser errors)
.TP
.B
\fB-r\fP

2
templates/readme_template
View File

@ -114,7 +114,7 @@ The JSON output can be compact (default) or pretty formatted with the `-p` optio
- `-h` help. Use `jc -h --parser_name` for parser documentation
- `-m` monochrome JSON output
- `-p` pretty format the JSON output
- `-q` quiet mode. Suppresses parser warning messages
- `-q` quiet mode. Suppresses parser warning messages (use `-qq` to ignore streaming parser errors)
- `-r` raw output. Provides a more literal JSON output, typically with string values and no additional semantic processing
- `-u` unbuffer output
- `-v` version information

0
tests/fixtures/centos-7.7/ping-ip-O-streaming-quiet.json → tests/fixtures/centos-7.7/ping-ip-O-streaming-ignore-exceptions.json vendored
View File

24
tests/test_ls_s.py
View File

@ -109,7 +109,7 @@ class MyTests(unittest.TestCase):
"""
Test plain 'ls /' on Centos 7.7 (raises ParseError)
"""
g = jc.parsers.ls_s.parse(self.centos_7_7_ls.splitlines())
g = jc.parsers.ls_s.parse(self.centos_7_7_ls.splitlines(), quiet=True)
with self.assertRaises(ParseError):
list(g)
@ -117,67 +117,67 @@ class MyTests(unittest.TestCase):
"""
Test 'ls -al /' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ls_s.parse(self.centos_7_7_ls_al.splitlines())), self.centos_7_7_ls_al_streaming_json)
self.assertEqual(list(jc.parsers.ls_s.parse(self.centos_7_7_ls_al.splitlines(), quiet=True)), self.centos_7_7_ls_al_streaming_json)
def test_ls_al_ubuntu_18_4(self):
"""
Test 'ls -al /' on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ls_s.parse(self.ubuntu_18_4_ls_al.splitlines())), self.ubuntu_18_4_ls_al_streaming_json)
self.assertEqual(list(jc.parsers.ls_s.parse(self.ubuntu_18_4_ls_al.splitlines(), quiet=True)), self.ubuntu_18_4_ls_al_streaming_json)
def test_ls_al_osx_10_14_6(self):
"""
Test 'ls -al /' on OSX 10.14.6
"""
self.assertEqual(list(jc.parsers.ls_s.parse(self.osx_10_14_6_ls_al.splitlines())), self.osx_10_14_6_ls_al_streaming_json)
self.assertEqual(list(jc.parsers.ls_s.parse(self.osx_10_14_6_ls_al.splitlines(), quiet=True)), self.osx_10_14_6_ls_al_streaming_json)
def test_ls_alh_centos_7_7(self):
"""
Test 'ls -alh /' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ls_s.parse(self.centos_7_7_ls_alh.splitlines())), self.centos_7_7_ls_alh_streaming_json)
self.assertEqual(list(jc.parsers.ls_s.parse(self.centos_7_7_ls_alh.splitlines(), quiet=True)), self.centos_7_7_ls_alh_streaming_json)
def test_ls_alh_ubuntu_18_4(self):
"""
Test 'ls -alh /' on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ls_s.parse(self.ubuntu_18_4_ls_alh.splitlines())), self.ubuntu_18_4_ls_alh_streaming_json)
self.assertEqual(list(jc.parsers.ls_s.parse(self.ubuntu_18_4_ls_alh.splitlines(), quiet=True)), self.ubuntu_18_4_ls_alh_streaming_json)
def test_ls_alh_osx_10_14_6(self):
"""
Test 'ls -alh /' on OSX 10.14.6
"""
self.assertEqual(list(jc.parsers.ls_s.parse(self.osx_10_14_6_ls_alh.splitlines())), self.osx_10_14_6_ls_alh_streaming_json)
self.assertEqual(list(jc.parsers.ls_s.parse(self.osx_10_14_6_ls_alh.splitlines(), quiet=True)), self.osx_10_14_6_ls_alh_streaming_json)
def test_ls_alR_centos_7_7(self):
"""
Test 'ls -alR /usr' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ls_s.parse(self.centos_7_7_ls_alR.splitlines())), self.centos_7_7_ls_alR_streaming_json)
self.assertEqual(list(jc.parsers.ls_s.parse(self.centos_7_7_ls_alR.splitlines(), quiet=True)), self.centos_7_7_ls_alR_streaming_json)
def test_ls_alR_ubuntu_18_4(self):
"""
Test 'ls -alR /usr' on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ls_s.parse(self.ubuntu_18_4_ls_alR.splitlines())), self.ubuntu_18_4_ls_alR_streaming_json)
self.assertEqual(list(jc.parsers.ls_s.parse(self.ubuntu_18_4_ls_alR.splitlines(), quiet=True)), self.ubuntu_18_4_ls_alR_streaming_json)
def test_ls_alR_osx_10_14_6(self):
"""
Test 'ls -alR /usr' on OSX 10.14.6
"""
self.assertEqual(list(jc.parsers.ls_s.parse(self.osx_10_14_6_ls_alR.splitlines())), self.osx_10_14_6_ls_alR_streaming_json)
self.assertEqual(list(jc.parsers.ls_s.parse(self.osx_10_14_6_ls_alR.splitlines(), quiet=True)), self.osx_10_14_6_ls_alR_streaming_json)
def test_ls_lR_empty_folder_osx_10_14_6(self):
"""
Test 'ls -lR' for empty directories on OSX 10.14.6
"""
self.assertEqual(list(jc.parsers.ls_s.parse(self.osx_10_14_6_ls_lR_empty_folder.splitlines())), self.osx_10_14_6_ls_lR_empty_folder_streaming_json)
self.assertEqual(list(jc.parsers.ls_s.parse(self.osx_10_14_6_ls_lR_empty_folder.splitlines(), quiet=True)), self.osx_10_14_6_ls_lR_empty_folder_streaming_json)
def test_ls_l_iso_ubuntu_18_4(self):
"""
Test 'ls -l --time-style=full-iso' for files with convertible dates on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ls_s.parse(self.ubuntu_18_4_ls_l_iso.splitlines())), self.ubuntu_18_4_ls_l_iso_streaming_json)
self.assertEqual(list(jc.parsers.ls_s.parse(self.ubuntu_18_4_ls_l_iso.splitlines(), quiet=True)), self.ubuntu_18_4_ls_l_iso_streaming_json)
if __name__ == '__main__':

140
tests/test_ping_s.py
View File

@ -214,8 +214,8 @@ class MyTests(unittest.TestCase):
with open(os.path.join(THIS_DIR, os.pardir, 'tests/fixtures/centos-7.7/ping-ip-O-streaming.json'), 'r', encoding='utf-8') as f:
self.centos_7_7_ping_ip_O_streaming_json = json.loads(f.read())
with open(os.path.join(THIS_DIR, os.pardir, 'tests/fixtures/centos-7.7/ping-ip-O-streaming-quiet.json'), 'r', encoding='utf-8') as f:
self.centos_7_7_ping_ip_O_streaming_quiet_json = json.loads(f.read())
with open(os.path.join(THIS_DIR, os.pardir, 'tests/fixtures/centos-7.7/ping-ip-O-streaming-ignore-exceptions.json'), 'r', encoding='utf-8') as f:
self.centos_7_7_ping_ip_O_streaming_ignore_exceptions_json = json.loads(f.read())
with open(os.path.join(THIS_DIR, os.pardir, 'tests/fixtures/centos-7.7/ping-ip-O-D-streaming.json'), 'r', encoding='utf-8') as f:
self.centos_7_7_ping_ip_O_D_streaming_json = json.loads(f.read())
@ -399,61 +399,61 @@ class MyTests(unittest.TestCase):
"""
self.assertEqual(list(jc.parsers.ping_s.parse('')), [])
def test_ping_quiet_success(self):
def test_ping_ignore_exceptions_success(self):
"""
Test 'ping' with quiet option
Test 'ping' with -qq (ignore_exceptions) option
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_ip_O.splitlines(), quiet=True)), self.centos_7_7_ping_ip_O_streaming_quiet_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_ip_O.splitlines(), quiet=True, ignore_exceptions=True)), self.centos_7_7_ping_ip_O_streaming_ignore_exceptions_json)
def test_ping_quiet_error(self):
def test_ping_ignore_exceptions_error(self):
"""
Test 'ping' with quiet option and error
Test 'ping' with -qq (ignore_exceptions) option option and error
"""
data_in = 'not ping'
expected = json.loads('[{"_meta":{"success":false,"error":"ParseError: Could not detect ping OS","line":"not ping"}}]')
self.assertEqual(list(jc.parsers.ping_s.parse(data_in.splitlines(), quiet=True)), expected)
self.assertEqual(list(jc.parsers.ping_s.parse(data_in.splitlines(), quiet=True, ignore_exceptions=True)), expected)
def test_ping_ip_O_centos_7_7(self):
"""
Test 'ping <ip> -O' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_ip_O.splitlines())), self.centos_7_7_ping_ip_O_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_ip_O.splitlines(), quiet=True)), self.centos_7_7_ping_ip_O_streaming_json)
def test_ping_ip_O_D_centos_7_7(self):
"""
Test 'ping <ip> -O -D' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_ip_O_D.splitlines())), self.centos_7_7_ping_ip_O_D_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_ip_O_D.splitlines(), quiet=True)), self.centos_7_7_ping_ip_O_D_streaming_json)
def test_ping_hostname_O_centos_7_7(self):
"""
Test 'ping <hostname> -O' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_hostname_O.splitlines())), self.centos_7_7_ping_hostname_O_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_hostname_O.splitlines(), quiet=True)), self.centos_7_7_ping_hostname_O_streaming_json)
def test_ping_hostname_O_p_centos_7_7(self):
"""
Test 'ping <hostname> -O -p' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_hostname_O_p.splitlines())), self.centos_7_7_ping_hostname_O_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_hostname_O_p.splitlines(), quiet=True)), self.centos_7_7_ping_hostname_O_p_streaming_json)
def test_ping_hostname_O_D_p_s_centos_7_7(self):
"""
Test 'ping <hostname> -O -D -p -s' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_hostname_O_D_p_s.splitlines())), self.centos_7_7_ping_hostname_O_D_p_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_hostname_O_D_p_s.splitlines(), quiet=True)), self.centos_7_7_ping_hostname_O_D_p_s_streaming_json)
def test_ping6_ip_O_p_centos_7_7(self):
"""
Test 'ping6 <ip> -O -p' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping6_ip_O_p.splitlines())), self.centos_7_7_ping6_ip_O_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping6_ip_O_p.splitlines(), quiet=True)), self.centos_7_7_ping6_ip_O_p_streaming_json)
def test_ping6_ip_O_p_unparsable_centos_7_7(self):
"""
Test 'ping6 <ip> -O -p' with unparsable lines on Centos 7.7 (raises IndexError)
"""
g = jc.parsers.ping_s.parse(self.centos_7_7_ping6_ip_O_p_unparsable.splitlines())
g = jc.parsers.ping_s.parse(self.centos_7_7_ping6_ip_O_p_unparsable.splitlines(), quiet=True)
with self.assertRaises(IndexError):
list(g)
@ -461,37 +461,37 @@ class MyTests(unittest.TestCase):
"""
Test 'ping6 <ip> -O -D -p' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping6_ip_O_D_p.splitlines())), self.centos_7_7_ping6_ip_O_D_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping6_ip_O_D_p.splitlines(), quiet=True)), self.centos_7_7_ping6_ip_O_D_p_streaming_json)
def test_ping6_hostname_O_p_centos_7_7(self):
"""
Test 'ping6 <hostname> -O -p' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping6_hostname_O_p.splitlines())), self.centos_7_7_ping6_hostname_O_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping6_hostname_O_p.splitlines(), quiet=True)), self.centos_7_7_ping6_hostname_O_p_streaming_json)
def test_ping6_hostname_O_D_p_s_centos_7_7(self):
"""
Test 'ping6 <hostname> -O -D -p -s' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping6_hostname_O_D_p_s.splitlines())), self.centos_7_7_ping6_hostname_O_D_p_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping6_hostname_O_D_p_s.splitlines(), quiet=True)), self.centos_7_7_ping6_hostname_O_D_p_s_streaming_json)
def test_ping_ip_dup_centos_7_7(self):
"""
Test 'ping <ip>' to broadcast IP to get duplicate replies on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_ip_dup.splitlines())), self.centos_7_7_ping_ip_dup_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping_ip_dup.splitlines(), quiet=True)), self.centos_7_7_ping_ip_dup_streaming_json)
def test_ping6_ip_dup_centos_7_7(self):
"""
Test 'ping6 <ip>' to broadcast IP to get duplicate replies on Centos 7.7
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping6_ip_dup.splitlines())), self.centos_7_7_ping6_ip_dup_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.centos_7_7_ping6_ip_dup.splitlines(), quiet=True)), self.centos_7_7_ping6_ip_dup_streaming_json)
def test_ping_ip_O_unparsedlines_centos_7_7(self):
"""
Test 'ping <ip> -O' on Centos 7.7 with unparsable lines and error messages
"""
g = jc.parsers.ping_s.parse(self.centos_7_7_ping_ip_O_unparsedlines.splitlines())
g = jc.parsers.ping_s.parse(self.centos_7_7_ping_ip_O_unparsedlines.splitlines(), quiet=True)
with self.assertRaises(IndexError):
list(g)
@ -499,229 +499,229 @@ class MyTests(unittest.TestCase):
"""
Test 'ping <ip> -O' on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping_ip_O.splitlines())), self.ubuntu_18_4_ping_ip_O_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping_ip_O.splitlines(), quiet=True)), self.ubuntu_18_4_ping_ip_O_streaming_json)
def test_ping_ip_O_D_ubuntu_18_4(self):
"""
Test 'ping <ip> -O -D' on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping_ip_O_D.splitlines())), self.ubuntu_18_4_ping_ip_O_D_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping_ip_O_D.splitlines(), quiet=True)), self.ubuntu_18_4_ping_ip_O_D_streaming_json)
def test_ping_hostname_O_ubuntu_18_4(self):
"""
Test 'ping <hostname> -O' on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping_hostname_O.splitlines())), self.ubuntu_18_4_ping_hostname_O_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping_hostname_O.splitlines(), quiet=True)), self.ubuntu_18_4_ping_hostname_O_streaming_json)
def test_ping_hostname_O_p_ubuntu_18_4(self):
"""
Test 'ping <hostname> -O -p' on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping_hostname_O_p.splitlines())), self.ubuntu_18_4_ping_hostname_O_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping_hostname_O_p.splitlines(), quiet=True)), self.ubuntu_18_4_ping_hostname_O_p_streaming_json)
def test_ping_hostname_O_D_p_s_ubuntu_18_4(self):
"""
Test 'ping <hostname> -O -D -p -s' on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping_hostname_O_D_p_s.splitlines())), self.ubuntu_18_4_ping_hostname_O_D_p_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping_hostname_O_D_p_s.splitlines(), quiet=True)), self.ubuntu_18_4_ping_hostname_O_D_p_s_streaming_json)
def test_ping6_ip_O_p_ubuntu_18_4(self):
"""
Test 'ping6 <ip> -O -p' on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping6_ip_O_p.splitlines())), self.ubuntu_18_4_ping6_ip_O_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping6_ip_O_p.splitlines(), quiet=True)), self.ubuntu_18_4_ping6_ip_O_p_streaming_json)
def test_ping6_ip_O_D_p_ubuntu_18_4(self):
"""
Test 'ping6 <ip> -O -D -p' on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping6_ip_O_D_p.splitlines())), self.ubuntu_18_4_ping6_ip_O_D_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping6_ip_O_D_p.splitlines(), quiet=True)), self.ubuntu_18_4_ping6_ip_O_D_p_streaming_json)
def test_ping6_hostname_O_p_ubuntu_18_4(self):
"""
Test 'ping6 <hostname> -O -p' on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping6_hostname_O_p.splitlines())), self.ubuntu_18_4_ping6_hostname_O_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping6_hostname_O_p.splitlines(), quiet=True)), self.ubuntu_18_4_ping6_hostname_O_p_streaming_json)
def test_ping6_hostname_O_D_p_s_ubuntu_18_4(self):
"""
Test 'ping6 <hostname> -O -D -p -s' on Ubuntu 18.4
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping6_hostname_O_D_p_s.splitlines())), self.ubuntu_18_4_ping6_hostname_O_D_p_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.ubuntu_18_4_ping6_hostname_O_D_p_s.splitlines(), quiet=True)), self.ubuntu_18_4_ping6_hostname_O_D_p_s_streaming_json)
def test_ping_ip_O_fedora32(self):
"""
Test 'ping <ip> -O' on fedora32
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping_ip_O.splitlines())), self.fedora32_ping_ip_O_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping_ip_O.splitlines(), quiet=True)), self.fedora32_ping_ip_O_streaming_json)
def test_ping_ip_O_D_fedora32(self):
"""
Test 'ping <ip> -O -D' on fedora32
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping_ip_O_D.splitlines())), self.fedora32_ping_ip_O_D_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping_ip_O_D.splitlines(), quiet=True)), self.fedora32_ping_ip_O_D_streaming_json)
def test_ping_hostname_O_fedora32(self):
"""
Test 'ping <hostname> -O' on fedora32
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping_hostname_O.splitlines())), self.fedora32_ping_hostname_O_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping_hostname_O.splitlines(), quiet=True)), self.fedora32_ping_hostname_O_streaming_json)
def test_ping_hostname_O_p_fedora32(self):
"""
Test 'ping <hostname> -O -p' on fedora32
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping_hostname_O_p.splitlines())), self.fedora32_ping_hostname_O_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping_hostname_O_p.splitlines(), quiet=True)), self.fedora32_ping_hostname_O_p_streaming_json)
def test_ping_hostname_O_D_p_s_fedora32(self):
"""
Test 'ping <hostname> -O -D -p -s' on fedora32
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping_hostname_O_D_p_s.splitlines())), self.fedora32_ping_hostname_O_D_p_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping_hostname_O_D_p_s.splitlines(), quiet=True)), self.fedora32_ping_hostname_O_D_p_s_streaming_json)
def test_ping6_ip_O_p_fedora32(self):
"""
Test 'ping6 <ip> -O -p' on fedora32
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping6_ip_O_p.splitlines())), self.fedora32_ping6_ip_O_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping6_ip_O_p.splitlines(), quiet=True)), self.fedora32_ping6_ip_O_p_streaming_json)
def test_ping6_ip_O_D_p_fedora32(self):
"""
Test 'ping6 <ip> -O -D -p' on fedora32
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping6_ip_O_D_p.splitlines())), self.fedora32_ping6_ip_O_D_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping6_ip_O_D_p.splitlines(), quiet=True)), self.fedora32_ping6_ip_O_D_p_streaming_json)
def test_ping6_hostname_O_p_fedora32(self):
"""
Test 'ping6 <hostname> -O -p' on fedora32
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping6_hostname_O_p.splitlines())), self.fedora32_ping6_hostname_O_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping6_hostname_O_p.splitlines(), quiet=True)), self.fedora32_ping6_hostname_O_p_streaming_json)
def test_ping6_hostname_O_D_p_s_fedora32(self):
"""
Test 'ping6 <hostname> -O -D -p -s' on fedora32
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping6_hostname_O_D_p_s.splitlines())), self.fedora32_ping6_hostname_O_D_p_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.fedora32_ping6_hostname_O_D_p_s.splitlines(), quiet=True)), self.fedora32_ping6_hostname_O_D_p_s_streaming_json)
def test_ping_hostname_p_freebsd12(self):
"""
Test 'ping <hostname> -p' on freebsd12
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping_hostname_p.splitlines())), self.freebsd12_ping_hostname_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping_hostname_p.splitlines(), quiet=True)), self.freebsd12_ping_hostname_p_streaming_json)
def test_ping_hostname_s_freebsd12(self):
"""
Test 'ping <hostname> -s' on freebsd12
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping_hostname_s.splitlines())), self.freebsd12_ping_hostname_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping_hostname_s.splitlines(), quiet=True)), self.freebsd12_ping_hostname_s_streaming_json)
def test_ping_ping_hostname_freebsd12(self):
"""
Test 'ping <hostname>' on freebsd12
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping_hostname.splitlines())), self.freebsd12_ping_hostname_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping_hostname.splitlines(), quiet=True)), self.freebsd12_ping_hostname_streaming_json)
def test_ping_ip_p_freebsd12(self):
"""
Test 'ping <ip> -p' on freebsd12
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping_ip_p.splitlines())), self.freebsd12_ping_ip_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping_ip_p.splitlines(), quiet=True)), self.freebsd12_ping_ip_p_streaming_json)
def test_ping_ip_s_freebsd12(self):
"""
Test 'ping <ip> -s' on freebsd12
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping_ip_s.splitlines())), self.freebsd12_ping_ip_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping_ip_s.splitlines(), quiet=True)), self.freebsd12_ping_ip_s_streaming_json)
def test_ping_ip_freebsd12(self):
"""
Test 'ping6 <ip>' on freebsd127
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping_ip.splitlines())), self.freebsd12_ping_ip_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping_ip.splitlines(), quiet=True)), self.freebsd12_ping_ip_streaming_json)
def test_ping6_hostname_p_freebsd12(self):
"""
Test 'ping6 <hostname> -p' on freebsd12
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping6_hostname_p.splitlines())), self.freebsd12_ping6_hostname_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping6_hostname_p.splitlines(), quiet=True)), self.freebsd12_ping6_hostname_p_streaming_json)
def test_ping6_hostname_s_freebsd12(self):
"""
Test 'ping6 <hostname> -s' on freebsd12
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping6_hostname_s.splitlines())), self.freebsd12_ping6_hostname_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping6_hostname_s.splitlines(), quiet=True)), self.freebsd12_ping6_hostname_s_streaming_json)
def test_ping6_hostname_freebsd12(self):
"""
Test 'ping6 <hostname>' on freebsd12
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping6_hostname.splitlines())), self.freebsd12_ping6_hostname_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping6_hostname.splitlines(), quiet=True)), self.freebsd12_ping6_hostname_streaming_json)
def test_ping6_ip_p_freebsd12(self):
"""
Test 'ping6 <ip> -p' on freebsd12
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping6_ip_p.splitlines())), self.freebsd12_ping6_ip_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping6_ip_p.splitlines(), quiet=True)), self.freebsd12_ping6_ip_p_streaming_json)
def test_ping6_ip_s_freebsd12(self):
"""
Test 'ping6 <ip> -s' on freebsd12
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping6_ip_s.splitlines())), self.freebsd12_ping6_ip_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping6_ip_s.splitlines(), quiet=True)), self.freebsd12_ping6_ip_s_streaming_json)
def test_ping6_ip_freebsd12(self):
"""
Test 'ping6 <ip>' on freebsd12
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping6_ip.splitlines())), self.freebsd12_ping6_ip_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.freebsd12_ping6_ip.splitlines(), quiet=True)), self.freebsd12_ping6_ip_streaming_json)
def test_ping_hostname_p_osx_10_14_6(self):
"""
Test 'ping <hostname> -p' on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_hostname_p.splitlines())), self.osx_10_14_6_ping_hostname_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_hostname_p.splitlines(), quiet=True)), self.osx_10_14_6_ping_hostname_p_streaming_json)
def test_ping_hostname_s_osx_10_14_6(self):
"""
Test 'ping <hostname> -s' on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_hostname_s.splitlines())), self.osx_10_14_6_ping_hostname_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_hostname_s.splitlines(), quiet=True)), self.osx_10_14_6_ping_hostname_s_streaming_json)
def test_ping_hostname_osx_10_14_6(self):
"""
Test 'ping <hostname>' on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_hostname.splitlines())), self.osx_10_14_6_ping_hostname_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_hostname.splitlines(), quiet=True)), self.osx_10_14_6_ping_hostname_streaming_json)
def test_ping_ip_p_osx_10_14_6(self):
"""
Test 'ping <ip> -p' on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_ip_p.splitlines())), self.osx_10_14_6_ping_ip_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_ip_p.splitlines(), quiet=True)), self.osx_10_14_6_ping_ip_p_streaming_json)
def test_ping_ip_s_osx_10_14_6(self):
"""
Test 'ping <ip> -s' on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_ip_s.splitlines())), self.osx_10_14_6_ping_ip_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_ip_s.splitlines(), quiet=True)), self.osx_10_14_6_ping_ip_s_streaming_json)
def test_ping_ip_osx_10_14_6(self):
"""
Test 'ping <ip>' on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_ip.splitlines())), self.osx_10_14_6_ping_ip_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_ip.splitlines(), quiet=True)), self.osx_10_14_6_ping_ip_streaming_json)
def test_ping_ip_unreachable_osx_10_14_6(self):
"""
Test 'ping <ip>' with host unreachable error on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_ip_unreachable.splitlines())), self.osx_10_14_6_ping_ip_unreachable_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_ip_unreachable.splitlines(), quiet=True)), self.osx_10_14_6_ping_ip_unreachable_streaming_json)
def test_ping_ip_unknown_errors_osx_10_14_6(self):
"""
Test 'ping <ip>' with unknown/unparsable errors on osx 10.14.6
"""
g = jc.parsers.ping_s.parse(self.osx_10_14_6_ping_ip_unknown_errors.splitlines())
g = jc.parsers.ping_s.parse(self.osx_10_14_6_ping_ip_unknown_errors.splitlines(), quiet=True)
with self.assertRaises(IndexError):
list(g)
@ -729,43 +729,43 @@ class MyTests(unittest.TestCase):
"""
Test 'ping6 <hostname> -p' on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_hostname_p.splitlines())), self.osx_10_14_6_ping6_hostname_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_hostname_p.splitlines(), quiet=True)), self.osx_10_14_6_ping6_hostname_p_streaming_json)
def test_ping6_hostname_s_osx_10_14_6(self):
"""
Test 'ping6 <hostname> -s' on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_hostname_s.splitlines())), self.osx_10_14_6_ping6_hostname_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_hostname_s.splitlines(), quiet=True)), self.osx_10_14_6_ping6_hostname_s_streaming_json)
def test_ping6_hostname_osx_10_14_6(self):
"""
Test 'ping6 <hostname>' on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_hostname.splitlines())), self.osx_10_14_6_ping6_hostname_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_hostname.splitlines(), quiet=True)), self.osx_10_14_6_ping6_hostname_streaming_json)
def test_ping6_ip_p_osx_10_14_6(self):
"""
Test 'ping6 <ip> -p' on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_ip_p.splitlines())), self.osx_10_14_6_ping6_ip_p_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_ip_p.splitlines(), quiet=True)), self.osx_10_14_6_ping6_ip_p_streaming_json)
def test_ping6_ip_s_osx_10_14_6(self):
"""
Test 'ping6 <ip> -s' on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_ip_s.splitlines())), self.osx_10_14_6_ping6_ip_s_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_ip_s.splitlines(), quiet=True)), self.osx_10_14_6_ping6_ip_s_streaming_json)
def test_ping6_ip_osx_10_14_6(self):
"""
Test 'ping6 <ip>' on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_ip.splitlines())), self.osx_10_14_6_ping6_ip_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_ip.splitlines(), quiet=True)), self.osx_10_14_6_ping6_ip_streaming_json)
def test_ping6_ip_unparsable_osx_10_14_6(self):
"""
Test 'ping6 <ip>' with unparsable lines on osx 10.14.6
"""
g = jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_ip_unparsable.splitlines())
g = jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_ip_unparsable.splitlines(), quiet=True)
with self.assertRaises(IndexError):
list(g)
@ -773,25 +773,25 @@ class MyTests(unittest.TestCase):
"""
Test 'ping <ip>' to broadcast IP to get duplicate replies on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_ip_dup.splitlines())), self.osx_10_14_6_ping_ip_dup_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping_ip_dup.splitlines(), quiet=True)), self.osx_10_14_6_ping_ip_dup_streaming_json)
def test_ping6_ip_dup_osx_10_14_6(self):
"""
Test 'ping6 <ip>' to broadcast IP to get duplicate replies on osx 10.14.6
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_ip_dup.splitlines())), self.osx_10_14_6_ping6_ip_dup_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.osx_10_14_6_ping6_ip_dup.splitlines(), quiet=True)), self.osx_10_14_6_ping6_ip_dup_streaming_json)
def test_ping_ip_O_pi(self):
"""
Test 'ping6 <ip> -O' on raspberry pi
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.pi_ping_ip_O.splitlines())), self.pi_ping_ip_O_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.pi_ping_ip_O.splitlines(), quiet=True)), self.pi_ping_ip_O_streaming_json)
def test_ping_ip_O_D_pi(self):
"""
Test 'ping6 <ip> -O -D' on raspberry pi
"""
self.assertEqual(list(jc.parsers.ping_s.parse(self.pi_ping_ip_O_D.splitlines())), self.pi_ping_ip_O_D_streaming_json)
self.assertEqual(list(jc.parsers.ping_s.parse(self.pi_ping_ip_O_D.splitlines(), quiet=True)), self.pi_ping_ip_O_D_streaming_json)
if __name__ == '__main__':

16
tests/test_vmstat_s.py
View File

@ -74,7 +74,7 @@ class MyTests(unittest.TestCase):
def test_vmstat_unparsable(self):
data = 'unparsable data'
g = jc.parsers.vmstat_s.parse(data.splitlines())
g = jc.parsers.vmstat_s.parse(data.splitlines(), quiet=True)
with self.assertRaises(KeyError):
list(g)
@ -82,43 +82,43 @@ class MyTests(unittest.TestCase):
"""
Test 'vmstat' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat.splitlines())), self.centos_7_7_vmstat_streaming_json)
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat.splitlines(), quiet=True)), self.centos_7_7_vmstat_streaming_json)
def test_vmstat_a_centos_7_7(self):
"""
Test 'vmstat -a' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat_a.splitlines())), self.centos_7_7_vmstat_a_streaming_json)
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat_a.splitlines(), quiet=True)), self.centos_7_7_vmstat_a_streaming_json)
def test_vmstat_w_centos_7_7(self):
"""
Test 'vmstat -w' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat_w.splitlines())), self.centos_7_7_vmstat_w_streaming_json)
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat_w.splitlines(), quiet=True)), self.centos_7_7_vmstat_w_streaming_json)
def test_vmstat_at_5_10_centos_7_7(self):
"""
Test 'vmstat -at 5 10' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat_at_5_10.splitlines())), self.centos_7_7_vmstat_at_5_10_streaming_json)
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat_at_5_10.splitlines(), quiet=True)), self.centos_7_7_vmstat_at_5_10_streaming_json)
def test_vmstat_awt_centos_7_7(self):
"""
Test 'vmstat -awt' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat_awt.splitlines())), self.centos_7_7_vmstat_awt_streaming_json)
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat_awt.splitlines(), quiet=True)), self.centos_7_7_vmstat_awt_streaming_json)
def test_vmstat_d_centos_7_7(self):
"""
Test 'vmstat -d' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat_d.splitlines())), self.centos_7_7_vmstat_d_streaming_json)
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat_d.splitlines(), quiet=True)), self.centos_7_7_vmstat_d_streaming_json)
def test_vmstat_dt_centos_7_7(self):
"""
Test 'vmstat -dt' on Centos 7.7
"""
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat_dt.splitlines())), self.centos_7_7_vmstat_dt_streaming_json)
self.assertEqual(list(jc.parsers.vmstat_s.parse(self.centos_7_7_vmstat_dt.splitlines(), quiet=True)), self.centos_7_7_vmstat_dt_streaming_json)
if __name__ == '__main__':
unittest.main()