six.b() is not needed for literals since Python 2.6/3.3

lib/jnpr/junos/jxml.py

@@ -1,7 +1,6 @@
 from ncclient import manager
 from ncclient.xml_ import NCElement
 from lxml import etree
-import six
 
 """
   These are Junos XML 'helper' definitions use for generic XML processing
@@ -226,8 +225,7 @@ def cscript_conf(reply):
 
 
 # xslt to remove prefix like junos:ns
-strip_namespaces_prefix = six.b(
-    """<?xml version="1.0" encoding="UTF-8" ?>
+strip_namespaces_prefix = b"""<?xml version="1.0" encoding="UTF-8" ?>
 <xsl:stylesheet version="1.0" xmlns:xsl="http://www.w3.org/1999/XSL/Transform">
   <xsl:output method="xml" indent="no" omit-xml-declaration="no" />
 
@@ -249,4 +247,3 @@ def cscript_conf(reply):
         </xsl:attribute>
     </xsl:template>
 </xsl:stylesheet>"""
-)

lib/jnpr/junos/transport/tty_netconf.py

@@ -16,18 +16,18 @@
 
 
 class PY6:
-    NEW_LINE = six.b("\n")
-    EMPTY_STR = six.b("")
-    NETCONF_EOM = six.b("]]>]]>")
-    STARTS_WITH = six.b("<!--")
+    NEW_LINE = b"\n"
+    EMPTY_STR = b""
+    NETCONF_EOM = b"]]>]]>"
+    STARTS_WITH = b"<!--"
 
 
 __all__ = ["xmlmode_netconf"]
 
-_NETCONF_EOM = six.b("]]>]]>")
-_xmlns = re.compile(six.b("xmlns=[^>]+"))
+_NETCONF_EOM = b"]]>]]>"
+_xmlns = re.compile(b"xmlns=[^>]+")
 _xmlns_strip = lambda text: _xmlns.sub(PY6.EMPTY_STR, text)
-_junosns = re.compile(six.b("junos:"))
+_junosns = re.compile(b"junos:")
 _junosns_strip = lambda text: _junosns.sub(PY6.EMPTY_STR, text)
 
 logger = logging.getLogger("jnpr.junos.tty_netconf")

lib/jnpr/junos/transport/tty_serial.py

@@ -10,7 +10,7 @@
 # Terminal connection over SERIAL CONSOLE
 # -------------------------------------------------------------------------
 
-_PROMPT = re.compile(six.b("|").join([six.b(i) for i in Terminal._RE_PAT]))
+_PROMPT = re.compile(b"|".join([six.b(i) for i in Terminal._RE_PAT]))
 
 
 class Serial(Terminal):
@@ -75,7 +75,7 @@ def read_prompt(self):
         regular-expression group. If a timeout occurs, then return
         the tuple(None,None).
         """
-        rxb = six.b("")
+        rxb = b""
         mark_start = datetime.now()
         mark_end = mark_start + timedelta(seconds=self.EXPECT_TIMEOUT)
 

lib/jnpr/junos/transport/tty_ssh.py

@@ -15,14 +15,14 @@
 # -------------------------------------------------------------------------
 # Terminal connection over SSH CONSOLE
 # -------------------------------------------------------------------------
-_PROMPT = re.compile(six.b("|").join([six.b(i) for i in Terminal._RE_PAT]))
+_PROMPT = re.compile(b"|".join([six.b(i) for i in Terminal._RE_PAT]))
 
 
 class PY6:
-    NEW_LINE = six.b("\n")
-    EMPTY_STR = six.b("")
-    NETCONF_EOM = six.b("]]>]]>")
-    IN_USE = six.b("in use")
+    NEW_LINE = b"\n"
+    EMPTY_STR = b""
+    NETCONF_EOM = b"]]>]]>"
+    IN_USE = b"in use"
 
 
 class SSH(Terminal):
@@ -158,7 +158,7 @@ def rawwrite(self, content):
 
     def read(self):
         """read a single line"""
-        rxb = six.b("")
+        rxb = b""
         while True:
             data = self._ssh.recv(self.RECVSZ)
             if data is None or len(data) <= 0:
@@ -180,7 +180,7 @@ def read_prompt(self):
         regular-expression group. If a timeout occurs, then return
         the tuple(None,None).
         """
-        rxb = six.b("")
+        rxb = b""
         timeout = time() + self.READ_PROMPT_DELAY
 
         while time() < timeout:
@@ -199,7 +199,7 @@ def read_prompt(self):
         return rxb, found.lastgroup
 
     def _read_until(self, match, timeout=None):
-        rxb = six.b("")
+        rxb = b""
         timeout = time() + self.READ_PROMPT_DELAY
 
         while time() < timeout:

lib/jnpr/junos/transport/tty_telnet.py

@@ -14,10 +14,10 @@
 
 
 class PY6:
-    NEW_LINE = six.b("\n")
-    EMPTY_STR = six.b("")
-    NETCONF_EOM = six.b("]]>]]>")
-    IN_USE = six.b("in use")
+    NEW_LINE = b"\n"
+    EMPTY_STR = b""
+    NETCONF_EOM = b"]]>]]>"
+    IN_USE = b("in use"
 
 
 class Telnet(Terminal):

lib/jnpr/junos/utils/start_shell.py

@@ -3,7 +3,6 @@
 import datetime
 from jnpr.junos.utils.ssh_client import open_ssh_client
 import subprocess
-import six
 from threading import Thread
 import time
 

tests/unit/test_console.py

@@ -9,7 +9,6 @@
 import sys
 import os
 from lxml import etree
-import six
 import socket
 
 from jnpr.junos.console import Console
@@ -40,16 +39,16 @@ def tearDownClass(cls):
     def setUp(self, mock_write, mock_expect, mock_open):
         tty_netconf.open = MagicMock()
         mock_expect.side_effect = [
-            (1, re.search("(?P<login>ogin:\s*$)", "login: "), six.b("\r\r\n ogin:")),
+            (1, re.search("(?P<login>ogin:\s*$)", "login: "), b"\r\r\n ogin:"),
             (
                 2,
                 re.search("(?P<passwd>assword:\s*$)", "password: "),
-                six.b("\r\r\n password:"),
+                b"\r\r\n password:",
             ),
             (
                 3,
                 re.search("(?P<shell>%|#\s*$)", "junos % "),
-                six.b("\r\r\nroot@device:~ # "),
+                b"\r\r\nroot@device:~ # ",
             ),
         ]
         self.dev = Console(host="1.1.1.1", user="lab", password="lab123", mode="Telnet")
@@ -87,16 +86,16 @@ def test_telnet_old_fact_warning(self, mock_warn):
     def test_login_bad_password(self, mock_write, mock_expect, mock_open):
         tty_netconf.open = MagicMock()
         mock_expect.side_effect = [
-            (1, re.search("(?P<login>ogin:\s*$)", "login: "), six.b("\r\r\n ogin:")),
+            (1, re.search("(?P<login>ogin:\s*$)", "login: "), b"\r\r\n ogin:"),
             (
                 2,
                 re.search("(?P<passwd>assword:\s*$)", "password: "),
-                six.b("\r\r\n password:"),
+                b"\r\r\n password:",
             ),
             (
                 3,
                 re.search("(?P<badpasswd>ogin incorrect)", "login incorrect"),
-                six.b("\r\r\nlogin incorrect"),
+                b"\r\r\nlogin incorrect",
             ),
         ]
         self.dev = Console(host="1.1.1.1", user="lab", password="lab123", mode="Telnet")
@@ -110,16 +109,16 @@ def test_with_context(self, mock_write, mock_expect, mock_open, mock_logout):
         tty_netconf.open = MagicMock()
 
         mock_expect.side_effect = [
-            (1, re.search("(?P<login>ogin:\s*$)", "login: "), six.b("\r\r\n ogin:")),
+            (1, re.search("(?P<login>ogin:\s*$)", "login: "), b"\r\r\n ogin:"),
             (
                 2,
                 re.search("(?P<passwd>assword:\s*$)", "password: "),
-                six.b("\r\r\n password:"),
+                b"\r\r\n password:",
             ),
             (
                 3,
                 re.search("(?P<shell>%|#\s*$)", "junos % "),
-                six.b("\r\r\nroot@device:~ # "),
+                b"\r\r\nroot@device:~ # ",
             ),
         ]
         with Console(
@@ -175,9 +174,9 @@ def test_console_tty_open_err(self, mock_login, mock_telnet):
     def test_console_serial(self, mock_write, mock_expect, mock_open):
         tty_netconf.open = MagicMock()
         mock_expect.side_effect = [
-            six.b("\r\r\n Login:"),
-            six.b("\r\r\n password:"),
-            six.b("\r\r\nroot@device:~ # "),
+            b"\r\r\n Login:",
+            b"\r\r\n password:",
+            b"\r\r\nroot@device:~ # ",
         ]
         self.dev = Console(host="1.1.1.1", user="lab", password="lab123", mode="serial")
         self.dev.open()
@@ -258,15 +257,13 @@ def test_load_console(
                 </policy-statement>
                 </policy-options>"""
 
-        mock_read_until.return_value = six.b(
-            """
+        mock_read_until.return_value = b"""
         <rpc-reply xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" xmlns:junos="http://xml.juniper.net/junos/15.2I0/junos">
             <load-configuration-results>
             <ok/>
             </load-configuration-results>
             </rpc-reply>
             ]]>]]>"""
-        )
         cu = Config(self.dev)
         op = cu.load(xml, format="xml")
         cu.commit()

tests/unit/transport/test_serial.py

@@ -5,7 +5,6 @@
 import nose2
 from mock import MagicMock, patch
 import sys
-import six
 
 from jnpr.junos.console import Console
 
@@ -90,12 +89,9 @@ def setUp(self, mock_read, mock_flush, mock_write, mock_serial_read, mock_open):
             ("shell", "shell"),
         ]
         mock_serial_read.side_effect = [
-            six.b(
-                "<!-- No zombies were killed during the creation of this user interface -->"
-            ),
-            six.b(""),
-            six.b(
-                """<!-- user root, class super-user -->
+            b"<!-- No zombies were killed during the creation of this user interface -->",
+            b"",
+            b"""<!-- user root, class super-user -->
 <hello xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
   <capabilities>
     <capability>urn:ietf:params:netconf:base:1.0</capability>
@@ -114,9 +110,8 @@ def setUp(self, mock_read, mock_flush, mock_write, mock_serial_read, mock_open):
   </capabilities>
   <session-id>7478</session-id>
 </hello>
-]]>]]>"""
-            ),
-            six.b(""),
+]]>]]>""",
+            b"",
         ]
         self.dev.open()
 
@@ -144,8 +139,7 @@ def test_tty_serial_win_rpc_call(self, mock_serial_close, mock_close):
         self.dev._tty.read = MagicMock()
         self.dev._tty.rawwrite = MagicMock()
         self.dev._tty.read.side_effect = [
-            six.b(
-                '<rpc-reply xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"'
+               b'<rpc-reply xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"'
                 ' xmlns:junos="http://xml.juniper.net/junos/15.1X49/junos">'
                 '<route-engine-information xmlns="http://xml.juniper.net/ju'
                 'nos/15.1X49/junos-chassis"><route-engine><status>OK</statu'
@@ -171,7 +165,6 @@ def test_tty_serial_win_rpc_call(self, mock_serial_close, mock_close):
                 "rage-five>0.08</load-average-five><load-average-fifteen>0."
                 "06</load-average-fifteen></route-engine></route-engine-inf"
                 "ormation></rpc-reply>]]>]]>"
-            )
         ]
         res = self.dev.rpc.get_route_engine_information()
         self.assertEqual(res.tag, "route-engine-information")

tests/unit/transport/test_tty_netconf.py

@@ -43,7 +43,7 @@ def test_open_at_shell_true(self, mock_rcv):
     @patch("jnpr.junos.transport.tty_netconf.timedelta")
     def test_open_RuntimeError(self, mock_delta, mock_rcv):
         mock_rcv.return_value = "]]>]]>"
-        self.tty_net._tty.read.return_value = six.b("testing")
+        self.tty_net._tty.read.return_value = b"testing"
         from datetime import timedelta
 
         mock_delta.return_value = timedelta(seconds=0.5)
@@ -56,7 +56,7 @@ def test_rpc(self, mock_rcv, mock_parse):
         mock_rcv.return_value = "]]>]]>"
         self.tty_net.rpc("get-interface-information")
         self.tty_net._tty.rawwrite.assert_called_with(
-            six.b("<rpc><get-interface-information/></rpc>")
+            b"<rpc><get-interface-information/></rpc>"
         )
 
     @patch("jnpr.junos.transport.tty_netconf.tty_netconf._receive")
@@ -103,7 +103,7 @@ def test_tty_netconf_receive_socket_error(self, mock_select):
     @patch("jnpr.junos.transport.tty_netconf.select.select")
     def test_tty_netconf_receive_empty_line(self, mock_select):
         rx = MagicMock()
-        rx.read_until.side_effect = iter([six.b(""), six.b("]]>]]>")])
+        rx.read_until.side_effect = iter([b"", b"]]>]]>"])
         mock_select.return_value = ([rx], [], [])
         self.assertEqual(self.tty_net._receive().tag, "error-in-receive")
 
@@ -126,30 +126,30 @@ def test_tty_netconf_receive_XMLSyntaxError(self, mock_select):
         rx = MagicMock()
 
         rx.read_until.side_effect = iter(
-            [six.b("<rpc-reply>ok<dummy></rpc-reply>"), six.b("\n]]>]]>")]
+            [b"<rpc-reply>ok<dummy></rpc-reply>", b"\n]]>]]>"]
         )
         mock_select.return_value = ([rx], [], [])
         self.assertEqual(
-            self.tty_net._receive(), six.b("<rpc-reply>ok<dummy/></rpc-reply>")
+            self.tty_net._receive(), b"<rpc-reply>ok<dummy/></rpc-reply>"
         )
 
     @patch("jnpr.junos.transport.tty_netconf.select.select")
     def test_tty_netconf_receive_XMLSyntaxError_eom_in_center(self, mock_select):
         rx = MagicMock()
         rx.read_until.side_effect = iter(
-            [six.b("<rpc-reply>ok</rpc-reply>"), six.b("]]>]]>\ndummy")]
+            [b"<rpc-reply>ok</rpc-reply>", b"]]>]]>\ndummy"]
         )
         mock_select.return_value = ([rx], [], [])
-        self.assertEqual(self.tty_net._receive(), six.b("<rpc-reply>ok</rpc-reply>"))
+        self.assertEqual(self.tty_net._receive(), b"<rpc-reply>ok</rpc-reply>")
 
     @patch("jnpr.junos.transport.tty_netconf.select.select")
     def test_tty_netconf_receive_xmn_error(self, mock_select):
         rx = MagicMock()
         rx.read_until.side_effect = iter(
             [
-                six.b("<message>ok</message>"),
-                six.b("\n</xnm:error>\n"),
-                six.b("]]>]]>\ndummy"),
+                b"<message>ok</message>",
+                b"\n</xnm:error>\n",
+                b"]]>]]>\ndummy",
             ]
         )
         mock_select.return_value = ([rx], [], [])

tests/unit/transport/test_tty_telnet.py

@@ -7,7 +7,6 @@
 import nose2
 from mock import MagicMock, patch
 from jnpr.junos.transport.tty_telnet import Telnet
-import six
 
 
 class TestTTYTelnet(unittest.TestCase):
@@ -62,7 +61,7 @@ def test_read_prompt_in_use_RuntimeError(self):
         self.tel_conn._tn.expect.return_value = (
             None,
             None,
-            six.b("port already in use"),
+            b"port already in use",
         )
         self.assertRaises(RuntimeError, self.tel_conn._login_state_machine)