Chaffelson / nipyapi / #2

Committed 25 Mar 2025 10:32AM UTC coverage: 69.867% (+69.9%) from 0.0%

Build # #2

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

V022release (#384)

* Update history for release

* Bump version: 0.21.0 → 0.22.0

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

1048 of 1500 relevant lines covered (69.87%)

0.7 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.78

/nipyapi/system.py

# -*- coding: utf-8 -*-

"""
For system and cluster level functions interacting with the underlying NiFi
Services
"""

from __future__ import absolute_import
import nipyapi


__all__ = [
    "get_system_diagnostics", "get_cluster", "get_node",
    "get_nifi_version_info"
]


def get_system_diagnostics():
    """
    Returns NiFi Sytems diagnostics page

    Returns (json):
    """
    with nipyapi.utils.rest_exceptions():
        return nipyapi.nifi.SystemDiagnosticsApi().get_system_diagnostics()


def get_cluster():
    """
    EXPERIMENTAL
    Returns the contents of the NiFi cluster

    Returns (json):
    """
    with nipyapi.utils.rest_exceptions():
        return nipyapi.nifi.ControllerApi().get_cluster()


def get_node(nid):
    """
    Returns the cluster node information

    Args:
        nid (str): The UUID of the Node to target

    Returns:

    """
    with nipyapi.utils.rest_exceptions():
        return nipyapi.nifi.ControllerApi().get_node(nid)


def get_nifi_version_info():
    """
    Returns the version information of the connected NiFi instance

    Returns (VersionInfoDTO):
    """
    diags = get_system_diagnostics()
    return diags.system_diagnostics.aggregate_snapshot.version_info


def get_registry_version_info():
    """
    Returns the version information of the connected NiFi Registry instance

    Returns (VersionInfoDTO):
    """
    details = nipyapi.registry.AboutApi().get_version()
    return details.registry_about_version

1	# -- coding: utf-8 --
2
3	"""
4	For system and cluster level functions interacting with the underlying NiFi
5	Services
6	"""
7
8	from __future__ import absolute_import	1✔
9	import nipyapi	1✔
10
11
12	__all__ = [	1✔
13	"get_system_diagnostics", "get_cluster", "get_node",
14	"get_nifi_version_info"
15	]
16
17
18	def get_system_diagnostics():	1✔
19	"""
20	Returns NiFi Sytems diagnostics page
21
22	Returns (json):
23	"""
24	with nipyapi.utils.rest_exceptions():	1✔
25	return nipyapi.nifi.SystemDiagnosticsApi().get_system_diagnostics()	1✔
26
27
28	def get_cluster():	1✔
29	"""
30	EXPERIMENTAL
31	Returns the contents of the NiFi cluster
32
33	Returns (json):
34	"""
35	with nipyapi.utils.rest_exceptions():	×
36	return nipyapi.nifi.ControllerApi().get_cluster()	×
37
38
39	def get_node(nid):	1✔
40	"""
41	Returns the cluster node information
42
43	Args:
44	nid (str): The UUID of the Node to target
45
46	Returns:
47
48	"""
49	with nipyapi.utils.rest_exceptions():	×
50	return nipyapi.nifi.ControllerApi().get_node(nid)	×
51
52
53	def get_nifi_version_info():	1✔
54	"""
55	Returns the version information of the connected NiFi instance
56
57	Returns (VersionInfoDTO):
58	"""
59	diags = get_system_diagnostics()	1✔
60	return diags.system_diagnostics.aggregate_snapshot.version_info	1✔
61
62
63	def get_registry_version_info():	1✔
64	"""
65	Returns the version information of the connected NiFi Registry instance
66
67	Returns (VersionInfoDTO):
68	"""
69	details = nipyapi.registry.AboutApi().get_version()	1✔
70	return details.registry_about_version	1✔

Chaffelson / nipyapi / #2

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous