Merge pull request #1333 from dineshbaburam91/vmhost_final_changes

Added vmhost paramater in dev.facts

lib/jnpr/junos/facts/__init__.py

@@ -48,6 +48,7 @@
 import jnpr.junos.facts.personality
 import jnpr.junos.facts.swver
 import jnpr.junos.facts.is_linux
+import jnpr.junos.facts.vmhost
 
 
 def _build_fact_callbacks_and_doc_strings():

lib/jnpr/junos/facts/vmhost.py

@@ -0,0 +1,40 @@
+from jnpr.junos.exception import RpcError
+import re
+from lxml import etree
+
+
+def provides_facts():
+    """
+    Returns a dictionary keyed on the facts provided by this module. The value
+    of each key is the doc string describing the fact.
+    """
+    return {
+        "vmhost": "A boolean indicating if the device is vmhost.",
+    }
+
+
+def get_facts(device):
+    """
+    Gathers facts from the sysctl command.
+    """
+    SYSCTL_VMHOST_MODE = "sysctl -n hw.re.vmhost_mode"
+    vmhost = None
+
+    if device.facts["_is_linux"]:
+        vmhost = False
+    else:
+        try:
+            rsp = device.rpc.request_shell_execute(command=SYSCTL_VMHOST_MODE)
+            if rsp.tag == "rpc-error":
+                raise RpcError()
+            result = re.sub("<[^<]+>", "", etree.tostring(rsp).decode())
+            if result.strip() == "1":
+                vmhost = True
+            else:
+                vmhost = False
+        except RpcError:
+            pass
+
+    return {
+        "vmhost": vmhost,
+    }

requirements.txt

@@ -1,6 +1,6 @@
 lxml>=3.2.4
-# ncclient version 0.6.10 has issues with PyEZ(junos-eznc) and needs to be avoided
-ncclient>=0.6.15
+# ncclient version 0.6.10 and 0.6.16 has issues with PyEZ(junos-eznc) and needs to be avoided
+ncclient==0.6.15
 scp>=0.7.0
 jinja2>=2.7.1
 PyYAML>=5.1

tests/unit/facts/rpc-reply/vmhost_error_request-shell-execute.xml

@@ -0,0 +1,11 @@
+<rpc-reply xmlns:junos="http://xml.juniper.net/junos/16.1R1/junos">
+<rpc-error>
+<error-type>protocol</error-type>
+<error-tag>operation-failed</error-tag>
+<error-severity>error</error-severity>
+<error-message>permission denied</error-message>
+<error-info>
+<bad-element>system</bad-element>
+</error-info>
+</rpc-error>
+</rpc-reply>

tests/unit/facts/rpc-reply/vmhost_false_request-shell-execute.xml

@@ -0,0 +1,5 @@
+<rpc-reply xmlns:junos="http://xml.juniper.net/junos/16.1R1/junos">
+<output>
+0
+</output>
+</rpc-reply>

tests/unit/facts/rpc-reply/vmhost_true_request-shell-execute.xml

@@ -0,0 +1,5 @@
+<rpc-reply xmlns:junos="http://xml.juniper.net/junos/16.1R1/junos">
+<output>
+1
+</output>
+</rpc-reply>

tests/unit/facts/test_vmhost.py

@@ -0,0 +1,76 @@
+__author__ = "Stacy Smith"
+__credits__ = "Jeremy Schulman, Nitin Kumar"
+
+try:
+    import unittest2 as unittest
+except:
+    import unittest
+import nose2
+from unittest.mock import patch, MagicMock
+import os
+
+from jnpr.junos import Device
+
+from ncclient.manager import Manager, make_device_handler
+from ncclient.transport import SSHSession
+
+
+class TestVmHost(unittest.TestCase):
+    @patch("ncclient.manager.connect")
+    def setUp(self, mock_connect):
+        mock_connect.side_effect = self._mock_manager_setup
+        self.dev = Device(
+            host="1.1.1.1", user="rick", password="password123", gather_facts=False
+        )
+        self.dev.open()
+
+    @patch("jnpr.junos.Device.execute")
+    def test_vmhost_true(self, mock_execute):
+        mock_execute.side_effect = self._mock_manager_vmhost_true
+        self.assertEqual(self.dev.facts["vmhost"], True)
+
+    @patch("jnpr.junos.Device.execute")
+    def test_vmhost_false(self, mock_execute):
+        mock_execute.side_effect = self._mock_manager_vmhost_false
+        self.assertEqual(self.dev.facts["vmhost"], False)
+
+    @patch("jnpr.junos.Device.execute")
+    def test_vmhost_linux(self, mock_execute):
+        mock_execute.side_effect = self._mock_manager_vmhost_false
+        self.dev.facts._cache["_is_linux"] = True
+        self.assertEqual(self.dev.facts["vmhost"], False)
+
+    @patch("jnpr.junos.Device.execute")
+    def test_vmhost_error(self, mock_execute):
+        mock_execute.side_effect = self._mock_manager_vmhost_error
+        self.assertEqual(self.dev.facts["vmhost"], None)
+
+    def _read_file(self, fname):
+        from ncclient.xml_ import NCElement
+
+        fpath = os.path.join(os.path.dirname(__file__), "rpc-reply", fname)
+        foo = open(fpath).read()
+
+        rpc_reply = NCElement(
+            foo, self.dev._conn._device_handler.transform_reply()
+        )._NCElement__doc[0]
+        return rpc_reply
+
+    def _mock_manager_setup(self, *args, **kwargs):
+        if kwargs:
+            device_params = kwargs["device_params"]
+            device_handler = make_device_handler(device_params)
+            session = SSHSession(device_handler)
+            return Manager(session, device_handler)
+
+    def _mock_manager_vmhost_true(self, *args, **kwargs):
+        if args:
+            return self._read_file("vmhost_true_" + args[0].tag + ".xml")
+
+    def _mock_manager_vmhost_false(self, *args, **kwargs):
+        if args:
+            return self._read_file("vmhost_false_" + args[0].tag + ".xml")
+
+    def _mock_manager_vmhost_error(self, *args, **kwargs):
+        if args:
+            return self._read_file("vmhost_error_" + args[0].tag + ".xml")