Repository: Azure/WALinuxAgent
Pull request #498 (2016-11-29 17:45:30)

Adds BIG-IP integration to WAAgent
This patch includes the implementation portion of BIG-IP integration
with WAAgent.


Patch:
diff --git a/azurelinuxagent/common/osutil/bigip.py b/azurelinuxagent/common/osutil/bigip.py
new file mode 100644
index 00000000..501ded6a
--- /dev/null
+++ b/azurelinuxagent/common/osutil/bigip.py
@@ -0,0 +1,350 @@
+# Copyright 2016 F5 Networks Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Requires Python 2.4+ and Openssl 1.0+
+#
+
+import array
+import fcntl
+import os
+import platform
+import re
+import socket
+import struct
+import time
+
+try:
+    # WAAgent > 2.1.3
+    import azurelinuxagent.common.logger as logger
+    import azurelinuxagent.common.utils.shellutil as shellutil
+
+    from azurelinuxagent.common.exception import OSUtilError
+    from azurelinuxagent.common.osutil.default import DefaultOSUtil
+except ImportError:
+    # WAAgent <= 2.1.3
+    import azurelinuxagent.logger as logger
+    import azurelinuxagent.utils.shellutil as shellutil
+
+    from azurelinuxagent.exception import OSUtilError
+    from azurelinuxagent.distro.default.osutil import DefaultOSUtil
+
+
+class BigIpOSUtil(DefaultOSUtil):
+    def __init__(self):
+        super(BigIpOSUtil, self).__init__()
+
+    def _wait_until_mcpd_is_initialized(self):
+        """Wait for mcpd to become available
+
+        All configuration happens in mcpd so we need to wait that this is
+        available before we go provisioning the system. I call this method
+        at the first opportunity I have (during the DVD mounting call).
+        This ensures that the rest of the provisioning does not need to wait
+        for mcpd to be available unless it absolutely wants to.
+
+        :return bool: Returns True upon success
+        :raises OSUtilError: Raises exception if mcpd does not come up within
+                             roughly 50 minutes (100 * 30 seconds)
+        """
+        for retries in range(1, 100):
+            # Retry until mcpd completes startup:
+            logger.info("Checking to see if mcpd is up")
+            rc = shellutil.run("/usr/bin/tmsh -a show sys mcp-state field-fmt 2>/dev/null | grep phase | grep running", chk_err=False)
+            if rc == 0:
+                logger.info("mcpd is up!")
+                break
+            time.sleep(30)
+
+        if rc is 0:
+            return True
+
+        raise OSUtilError(
+            "mcpd hasn't completed initialization! Cannot proceed!"
+        )
+
+    def _save_sys_config(self):
+        cmd = "/usr/bin/tmsh save sys config"
+        rc = shellutil.run(cmd)
+        if rc != 0:
+            logger.error("WARNING: Cannot save sys config on 1st boot.")
+        return rc
+
+    def restart_ssh_service(self):
+        return shellutil.run("/usr/bin/bigstart restart sshd", chk_err=False)
+
+    def stop_agent_service(self):
+        return shellutil.run("/sbin/service waagent stop", chk_err=False)
+
+    def start_agent_service(self):
+        return shellutil.run("/sbin/service waagent start", chk_err=False)
+
+    def register_agent_service(self):
+        return shellutil.run("/sbin/chkconfig --add waagent", chk_err=False)
+
+    def unregister_agent_service(self):
+        return shellutil.run("/sbin/chkconfig --del waagent", chk_err=False)
+
+    def get_dhcp_pid(self):
+        ret = shellutil.run_get_output("/sbin/pidof dhclient")
+        return ret[1] if ret[0] == 0 else None
+
+    def set_hostname(self, hostname):
+        """Set the static hostname of the device
+
+        Normally, tmsh is used to set the hostname for the system. For our
+        purposes at this time though, I would hesitate to trust this function.
+
+        Azure(Stack) uses the name that you provide in the Web UI or ARM (for
+        example) as the value of the hostname argument to this method. The
+        problem is that there is nowhere in the UI that specifies the
+        restrictions and checks that tmsh has for the hostname.
+
+        For example, if you set the name "bigip1" in the Web UI, Azure(Stack)
+        considers that a perfectly valid name. When WAAgent gets around to
+        running though, tmsh will reject that value because it is not a fully
+        qualified domain name. The proper value should have been bigip.xxx.yyy
+
+        WAAgent will not fail if this command fails, but the hostname will not
+        be what the user set either. Currently we do not set the hostname when
+        WAAgent starts up, so I am passing on setting it here too.
+
+        :param hostname: The hostname to set on the device
+        """
+        return None
+
+    def set_dhcp_hostname(self, hostname):
+        """Sets the DHCP hostname
+
+        See `set_hostname` for an explanation of why I pass here
+
+        :param hostname: The hostname to set on the device
+        """
+        return None
+
+    def useradd(self, username, expiration=None):
+        """Create user account using tmsh
+
+        Our policy is to create two accounts when booting a BIG-IP instance.
+        The first account is the one that the user specified when they did
+        the instance creation. The second one is the admin account that is,
+        or should be, built in to the system.
+
+        :param username: The username that you want to add to the system
+        :param expiration: The expiration date to use. We do not use this
+                           value.
+        """
+        userentry = self.get_userentry(username)
+        if userentry is not None:
+            logger.info("User {0} already exists, skip useradd", username)
+            return None
+
+        cmd = "/usr/bin/tmsh create auth user %s partition-access add { all-partitions { role admin } } shell bash" % (username)
+        retcode, out = shellutil.run_get_output(cmd, log_cmd=True, chk_err=True)
+        if retcode != 0:
+            raise OSUtilError(("Failed to create user account:{0}, "
+                               "retcode:{1}, "
+                               "output:{2}").format(username, retcode, out))
+        self._save_sys_config()
+        return retcode
+
+    def chpasswd(self, username, password, crypt_id=6, salt_len=10):
+        """Change a user's password with tmsh
+
+        Since we are creating the user specified account and additionally
+        changing the password of the built-in 'admin' account, both must
+        be modified in this method.
+
+        :param username: The username whose password to change
+        :param password: The unencrypted password to set for the user
+        :param crypt_id: If encrypting the password, the crypt_id that was used
+        :param salt_len: If encrypting the password, the length of the salt
+                         value used to do it.
+        """
+
+        if self.is_sys_user(username):
+            raise OSUtilError(("User {0} is a system user. "
+                               "Will not set passwd.").format(username))
+
+        # Start by setting the password of the user provided account
+        cmd = ("/usr/bin/tmsh modify auth user {0} password '{1}'"
+               "").format(username, password)
+        ret, output = shellutil.run_get_output(cmd, log_cmd=False, chk_err=True)
+        if ret != 0:
+            raise OSUtilError(("Failed to set password for {0}: {1}"
+                               "").format(username, output))
+
+        # Next, set the password of the built-in 'admin' account to be have
+        # the same password as the user provided account
+        userentry = self.get_userentry('admin')
+        if userentry is None:
+            raise OSUtilError("The 'admin' user account was not found!")
+
+        cmd = ("/usr/bin/tmsh modify auth user 'admin' password '{0}'").format(password)
+        ret, output = shellutil.run_get_output(cmd, log_cmd=False, chk_err=True)
+        if ret != 0:
+            raise OSUtilError(("Failed to set password for 'admin': {0}").format(output))
+        self._save_sys_config()
+        return ret
+
+    def get_dvd_device(self, dev_dir='/dev'):
+        """Find BIG-IP's CD/DVD device
+
+        This device is almost certainly /dev/cdrom so I added the ? to this pattern.
+        Note that this method will return upon the first device found, but in my
+        tests with 12.1.1 it will also find /dev/sr0 on occasion. This is NOT the
+        correct CD/DVD device though.
+
+        :todo: Consider just always returning "/dev/cdrom" here if that device device
+               exists on all platforms that are supported on Azure(Stack)
+        :param dev_dir: The root directory from which to look for devices
+        """
+        patten = r'(sr[0-9]|hd[c-z]|cdrom[0-9]?)'
+        for dvd in [re.match(patten, dev) for dev in os.listdir(dev_dir)]:
+            if dvd is not None:
+                return "/dev/{0}".format(dvd.group(0))
+        raise OSUtilError("Failed to get dvd device")
+
+    def mount_dvd(self, max_retry=6, chk_err=True):
+        """Mount the DVD containing the provisioningiso.iso file
+
+        This is the _first_ hook that WAAgent provides for us, so this is the
+        point where we should wait for mcpd to load. I am just overloading
+        this method to add the mcpd wait. Then I proceed with the stock code.
+
+        :param max_retry: Maximum number of retries waagent will make when
+                          mounting the provisioningiso.iso DVD
+        :param chk_err: Whether to check for errors or not in the mounting
+                        commands
+        """
+        self._wait_until_mcpd_is_initialized()
+        return super(BigIpOSUtil, self).mount_dvd(max_retry, chk_err)
+
+    def eject_dvd(self, chk_err=True):
+        """Runs the eject command to eject the provisioning DVD
+
+        BIG-IP does not include an eject command. It is sufficient to just
+        umount the DVD disk. But I will log that we do not support this for
+        future reference.
+
+        :param chk_err: Whether or not to check for errors raised by the eject
+                        command
+        """
+        logger.warn("Eject is not supported on this platform")
+
+    def set_admin_access_to_ip(self, dest_ip):
+        """Sets admin access to an IP address
+
+        This method is primarily used to limit which user account is allowed to
+        communicate with the Azure(Stack) metadata service. This service is at
+        the address 169.254.169.254 and includes information about the device
+        that "normal" users should not be allowed to see.
+
+        We cannot use this iptables command that comes with the default class
+        because we do not ship the 'ipt_owner' iptables extension with BIG-IP.
+
+        This should not be a problem though as the only people who should have
+        access to BIG-IP are people who are root anyways. Our system is not
+        a "general purpose" user system. So for those reasons I am dropping
+        that requirement from our implementation.
+
+        :param dest_ip: The IP address that you want to allow admin access for
+        """
+        self._set_accept_admin_access_to_ip(dest_ip)
+        self._set_drop_admin_access_to_ip(dest_ip)
+
+    def _set_accept_admin_access_to_ip(self, dest_ip):
+        """Sets the "accept" IP Tables rules
+
+        I broke this out to a separate function so that I could more easily
+        test it in the tests/common/osutil/test_default.py code
+
+        :param dest_ip:
+        :return:
+        """
+        # This allows root to access dest_ip
+        rm_old = "iptables -D OUTPUT -d {0} -j ACCEPT"
+        rule = "iptables -A OUTPUT -d {0} -j ACCEPT"
+        shellutil.run(rm_old.format(dest_ip), chk_err=False)
+        shellutil.run(rule.format(dest_ip))
+
+    def _set_drop_admin_access_to_ip(self, dest_ip):
+        """Sets the "drop" IP Tables rules
+
+        I broke this out to a separate function so that I could more easily
+        test it in the tests/common/osutil/test_default.py code
+
+        :param dest_ip:
+        :return:
+        """
+        # This blocks all other users to access dest_ip
+        rm_old = "iptables -D OUTPUT -d {0} -j DROP"
+        rule = "iptables -A OUTPUT -d {0} -j DROP"
+        shellutil.run(rm_old.format(dest_ip), chk_err=False)
+        shellutil.run(rule.format(dest_ip))
+
+    def get_first_if(self):
+        """Return the interface name, and ip addr of the management interface.
+
+        We need to add a struct_size check here because, curiously, our 64bit
+        platform is identified by python in Azure(Stack) as 32 bit and without
+        adjusting the struct_size, we can't get the information we need.
+
+        I believe this may be caused by only python i686 being shipped with
+        BIG-IP instead of python x86_64??
+        """
+        iface = ''
+        expected = 16  # how many devices should I expect...
+
+        python_arc = platform.architecture()[0]
+        if python_arc == '64bit':
+            struct_size = 40  # for 64bit the size is 40 bytes
+        else:
+            struct_size = 32  # for 32bit the size is 32 bytes
+        sock = socket.socket(socket.AF_INET,
+                             socket.SOCK_DGRAM,
+                             socket.IPPROTO_UDP)
+        buff = array.array('B', b'\0' * (expected * struct_size))
+        param = struct.pack('iL',
+                            expected*struct_size,
+                            buff.buffer_info()[0])
+        ret = fcntl.ioctl(sock.fileno(), 0x8912, param)
+        retsize = (struct.unpack('iL', ret)[0])
+        if retsize == (expected * struct_size):
+            logger.warn(('SIOCGIFCONF returned more than {0} up '
+                         'network interfaces.'), expected)
+        sock = buff.tostring()
+        for i in range(0, struct_size * expected, struct_size):
+            iface = sock[i:i+16].split(b'\0', 1)[0]
+            if iface == b'lo':
+                continue
+            else:
+                break
+        return iface.decode('latin-1'), socket.inet_ntoa(sock[i+20:i+24])
+
+    def device_for_ide_port(self, port_id):
+        """Return device name attached to ide port 'n'.
+
+        Include a wait in here because BIG-IP may not have yet initialized
+        this list of devices.
+
+        :param port_id:
+        :return:
+        """
+        for retries in range(1, 100):
+            # Retry until devices are ready
+            if os.path.exists("/sys/bus/vmbus/devices/"):
+                break
+            else:
+                time.sleep(10)
+        return super(BigIpOSUtil, self).device_for_ide_port(port_id)
diff --git a/azurelinuxagent/common/osutil/factory.py b/azurelinuxagent/common/osutil/factory.py
index cdfcd747..2718ba10 100644
--- a/azurelinuxagent/common/osutil/factory.py
+++ b/azurelinuxagent/common/osutil/factory.py
@@ -30,6 +30,7 @@ from .suse import SUSEOSUtil, SUSE11OSUtil
 from .ubuntu import UbuntuOSUtil, Ubuntu12OSUtil, Ubuntu14OSUtil, \
                     UbuntuSnappyOSUtil
 from .alpine import AlpineOSUtil
+from .bigip import BigIpOSUtil
 
 def get_osutil(distro_name=DISTRO_NAME, distro_version=DISTRO_VERSION,
                distro_full_name=DISTRO_FULL_NAME):
@@ -70,6 +71,8 @@ def get_osutil(distro_name=DISTRO_NAME, distro_version=DISTRO_VERSION,
             return RedhatOSUtil()
     elif distro_name == "freebsd":
         return FreeBSDOSUtil()
+    elif distro_name == "bigip":
+        return BigIpOSUtil()
     else:
         logger.warn("Unable to load distro implementation for {0}.", distro_name)
         logger.warn("Use default distro implementation instead.")
diff --git a/azurelinuxagent/common/version.py b/azurelinuxagent/common/version.py
index 6643bc26..45d81ae0 100644
--- a/azurelinuxagent/common/version.py
+++ b/azurelinuxagent/common/version.py
@@ -26,6 +26,36 @@ from azurelinuxagent.common.utils.flexible_version import FlexibleVersion
 from azurelinuxagent.common.future import ustr
 
 
+"""
+Add this workaround for detecting F5 products because BIG-IP/IQ/etc do not show
+their version info in the /etc/product-version location. Instead, the version
+and product information is contained in the /VERSION file.
+"""
+def get_f5_platform():
+    result = [None,None,None,None]
+    f5_version = re.compile("^Version: (\d+\.\d+\.\d+)")
+    f5_product = re.compile("^Product: ([\w-]+)")
+
+    with open('/VERSION', 'r') as fh:
+        content = fh.readlines()
+        for line in content:
+            version_matches = f5_version.match(line)
+            product_matches = f5_product.match(line)
+            if version_matches:
+                result[1] = version_matches.group(1)
+            elif product_matches:
+                result[3] = product_matches.group(1)
+                if result[3] == "BIG-IP":
+                    result[0] = "bigip"
+                    result[2] = "bigip"
+                elif result[3] == "BIG-IQ":
+                    result[0] = "bigiq"
+                    result[2] = "bigiq"
+                elif result[3] == "iWorkflow":
+                    result[0] = "iworkflow"
+                    result[2] = "iworkflow"
+    return result
+
 def get_distro():
     if 'FreeBSD' in platform.system():
         release = re.sub('\-.*\Z', '', ustr(platform.release()))
@@ -140,32 +170,3 @@ def is_snappy():
 if is_snappy():
     DISTRO_FULL_NAME = "Snappy Ubuntu Core"
 
-"""
-Add this workaround for detecting F5 products because BIG-IP/IQ/etc do not show
-their version info in the /etc/product-version location. Instead, the version
-and product information is contained in the /VERSION file.
-"""
-def get_f5_platform():
-    result = [None,None,None,None]
-    f5_version = re.compile("^Version: (\d+\.\d+\.\d+)")
-    f5_product = re.compile("^Product: ([\w-]+)")
-
-    with open('/VERSION', 'r') as fh:
-        content = fh.readlines()
-        for line in content:
-            version_matches = f5_version.match(line)
-            product_matches = f5_product.match(line)
-            if version_matches:
-                result[1] = version_matches.group(1)
-            elif product_matches:
-                result[3] = product_matches.group(1)
-                if result[3] == "BIG-IP":
-                    result[0] = "bigip"
-                    result[2] = "bigip"
-                elif result[3] == "BIG-IQ":
-                    result[0] = "bigiq"
-                    result[2] = "bigiq"
-                elif result[3] == "iWorkflow":
-                    result[0] = "iworkflow"
-                    result[2] = "iworkflow"
-    return result


Test patch:
diff --git a/tests/common/osutil/test_bigip.py b/tests/common/osutil/test_bigip.py
new file mode 100644
index 00000000..e5d130b7
--- /dev/null
+++ b/tests/common/osutil/test_bigip.py
@@ -0,0 +1,288 @@
+# Copyright 2016 F5 Networks Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Requires Python 2.4+ and Openssl 1.0+
+#
+
+import os
+import socket
+import time
+
+import azurelinuxagent.common.osutil.bigip as osutil
+import azurelinuxagent.common.osutil.default as default
+import azurelinuxagent.common.utils.shellutil as shellutil
+
+from azurelinuxagent.common.exception import OSUtilError
+from azurelinuxagent.common.exception import UpdateError
+from tests.tools import *
+
+
+class TestBigIpOSUtil_wait_until_mcpd_is_initialized(AgentTestCase):
+
+    @patch.object(shellutil, "run", return_value=0)
+    @patch.object(logger, "info", return_value=None)
+    def test_success(self, *args):
+        result = osutil.BigIpOSUtil._wait_until_mcpd_is_initialized(
+            osutil.BigIpOSUtil()
+        )
+        self.assertEqual(result, True)
+
+        # There are two logger calls in the mcpd wait function. The second
+        # occurs after mcpd is found to be "up"
+        self.assertEqual(args[0].call_count, 2)
+
+    @patch.object(shellutil, "run", return_value=1)
+    @patch.object(logger, "info", return_value=None)
+    @patch.object(time, "sleep", return_value=None)
+    def test_failure(self, *args):
+        self.assertRaises(
+            OSUtilError,
+            osutil.BigIpOSUtil._wait_until_mcpd_is_initialized,
+            osutil.BigIpOSUtil()
+        )
+
+
+class TestBigIpOSUtil_save_sys_config(AgentTestCase):
+
+    @patch.object(shellutil, "run", return_value=0)
+    @patch.object(logger, "error", return_value=None)
+    def test_success(self, *args):
+        result = osutil.BigIpOSUtil._save_sys_config(osutil.BigIpOSUtil())
+        self.assertEqual(result, 0)
+        self.assertEqual(args[0].call_count, 0)
+
+    @patch.object(shellutil, "run", return_value=1)
+    @patch.object(logger, "error", return_value=None)
+    def test_failure(self, *args):
+        result = osutil.BigIpOSUtil._save_sys_config(osutil.BigIpOSUtil())
+        self.assertEqual(result, 1)
+        self.assertEqual(args[0].call_count, 1)
+
+
+class TestBigIpOSUtil_get_dhcp_pid(AgentTestCase):
+
+    @patch.object(shellutil, "run_get_output", return_value=(0, 8623))
+    def test_success(self, *args):
+        result = osutil.BigIpOSUtil.get_dhcp_pid(osutil.BigIpOSUtil())
+        self.assertEqual(result, 8623)
+
+    @patch.object(shellutil, "run_get_output", return_value=(1, 'foo'))
+    def test_failure(self, *args):
+        result = osutil.BigIpOSUtil.get_dhcp_pid(osutil.BigIpOSUtil())
+        self.assertEqual(result, None)
+
+
+class TestBigIpOSUtil_useradd(AgentTestCase):
+
+    @patch.object(osutil.BigIpOSUtil, 'get_userentry', return_value=None)
+    @patch.object(shellutil, "run_get_output")
+    def test_success(self, *args):
+        args[0].return_value = (0, None)
+        result = osutil.BigIpOSUtil.useradd(
+            osutil.BigIpOSUtil(), 'foo', expiration=None
+        )
+        self.assertEqual(result, 0)
+
+    @patch.object(osutil.BigIpOSUtil, 'get_userentry', return_value=None)
+    def test_user_already_exists(self, *args):
+        args[0].return_value = 'admin'
+        result = osutil.BigIpOSUtil.useradd(
+            osutil.BigIpOSUtil(), 'admin', expiration=None
+        )
+        self.assertEqual(result, None)
+
+    @patch.object(shellutil, "run", return_value=1)
+    def test_failure(self, *args):
+        self.assertRaises(
+            OSUtilError,
+            osutil.BigIpOSUtil.useradd,
+            osutil.BigIpOSUtil(), 'foo', expiration=None
+        )
+
+
+class TestBigIpOSUtil_chpasswd(AgentTestCase):
+
+    @patch.object(shellutil, "run_get_output", return_value=(0, None))
+    @patch.object(osutil.BigIpOSUtil, 'get_userentry', return_value=True)
+    @patch.object(osutil.BigIpOSUtil, 'is_sys_user', return_value=False)
+    @patch.object(osutil.BigIpOSUtil, '_save_sys_config', return_value=None)
+    def test_success(self, *args):
+        result = osutil.BigIpOSUtil.chpasswd(
+            osutil.BigIpOSUtil(), 'admin', 'password', crypt_id=6, salt_len=10
+        )
+        self.assertEqual(result, 0)
+        self.assertEqual(args[0].call_count, 1)
+        self.assertEqual(args[0].call_count, 1)
+
+    @patch.object(osutil.BigIpOSUtil, 'is_sys_user', return_value=True)
+    def test_is_sys_user(self, *args):
+        self.assertRaises(
+            OSUtilError,
+            osutil.BigIpOSUtil.chpasswd,
+            osutil.BigIpOSUtil(), 'admin', 'password', crypt_id=6, salt_len=10
+        )
+
+    @patch.object(shellutil, "run_get_output", return_value=(1, None))
+    @patch.object(osutil.BigIpOSUtil, 'is_sys_user', return_value=False)
+    def test_failed_to_set_user_password(self, *args):
+        self.assertRaises(
+            OSUtilError,
+            osutil.BigIpOSUtil.chpasswd,
+            osutil.BigIpOSUtil(), 'admin', 'password', crypt_id=6, salt_len=10
+        )
+
+    @patch.object(shellutil, "run_get_output", return_value=(0, None))
+    @patch.object(osutil.BigIpOSUtil, 'is_sys_user', return_value=False)
+    @patch.object(osutil.BigIpOSUtil, 'get_userentry', return_value=None)
+    def test_failed_to_get_user_entry(self, *args):
+        self.assertRaises(
+            OSUtilError,
+            osutil.BigIpOSUtil.chpasswd,
+            osutil.BigIpOSUtil(), 'admin', 'password', crypt_id=6, salt_len=10
+        )
+
+
+class TestBigIpOSUtil_get_dvd_device(AgentTestCase):
+
+    @patch.object(os, "listdir", return_value=['tty1','cdrom0'])
+    def test_success(self, *args):
+        result = osutil.BigIpOSUtil.get_dvd_device(
+            osutil.BigIpOSUtil(), '/dev'
+        )
+        self.assertEqual(result, '/dev/cdrom0')
+
+    @patch.object(os, "listdir", return_value=['foo', 'bar'])
+    def test_failure(self, *args):
+        self.assertRaises(
+            OSUtilError,
+            osutil.BigIpOSUtil.get_dvd_device,
+            osutil.BigIpOSUtil(), '/dev'
+        )
+
+
+class TestBigIpOSUtil_restart_ssh_service(AgentTestCase):
+
+    @patch.object(shellutil, "run", return_value=0)
+    def test_success(self, *args):
+        result = osutil.BigIpOSUtil.restart_ssh_service(osutil.BigIpOSUtil())
+        self.assertEqual(result, 0)
+
+
+class TestBigIpOSUtil_stop_agent_service(AgentTestCase):
+
+    @patch.object(shellutil, "run", return_value=0)
+    def test_success(self, *args):
+        result = osutil.BigIpOSUtil.stop_agent_service(osutil.BigIpOSUtil())
+        self.assertEqual(result, 0)
+
+
+class TestBigIpOSUtil_start_agent_service(AgentTestCase):
+
+    @patch.object(shellutil, "run", return_value=0)
+    def test_success(self, *args):
+        result = osutil.BigIpOSUtil.start_agent_service(osutil.BigIpOSUtil())
+        self.assertEqual(result, 0)
+
+
+class TestBigIpOSUtil_register_agent_service(AgentTestCase):
+
+    @patch.object(shellutil, "run", return_value=0)
+    def test_success(self, *args):
+        result = osutil.BigIpOSUtil.register_agent_service(osutil.BigIpOSUtil())
+        self.assertEqual(result, 0)
+
+
+class TestBigIpOSUtil_unregister_agent_service(AgentTestCase):
+
+    @patch.object(shellutil, "run", return_value=0)
+    def test_success(self, *args):
+        result = osutil.BigIpOSUtil.unregister_agent_service(osutil.BigIpOSUtil())
+        self.assertEqual(result, 0)
+
+
+class TestBigIpOSUtil_set_hostname(AgentTestCase):
+
+    @patch.object(os.path, "exists", return_value=False)
+    def test_success(self, *args):
+        result = osutil.BigIpOSUtil.set_hostname(osutil.BigIpOSUtil(), None)
+        self.assertEqual(args[0].call_count, 0)
+        self.assertEqual(result, None)
+
+
+class TestBigIpOSUtil_set_dhcp_hostname(AgentTestCase):
+
+    @patch.object(os.path, "exists", return_value=False)
+    def test_success(self, *args):
+        result = osutil.BigIpOSUtil.set_dhcp_hostname(osutil.BigIpOSUtil(), None)
+        self.assertEqual(args[0].call_count, 0)
+        self.assertEqual(result, None)
+
+
+class TestBigIpOSUtil_get_first_if(AgentTestCase):
+
+    def test_success(self):
+        ifname, ipaddr = osutil.BigIpOSUtil().get_first_if()
+        self.assertTrue(ifname.startswith('eth'))
+        self.assertTrue(ipaddr is not None)
+        try:
+            socket.inet_aton(ipaddr)
+        except socket.error:
+            self.fail("not a valid ip address")
+
+
+class TestBigIpOSUtil_mount_dvd(AgentTestCase):
+
+    @patch.object(shellutil, "run", return_value=0)
+    @patch.object(time, "sleep", return_value=None)
+    @patch.object(osutil.BigIpOSUtil,
+                  '_wait_until_mcpd_is_initialized', return_value=None)
+    @patch.object(default.DefaultOSUtil, 'mount_dvd', return_value=None)
+    def test_success(self, *args):
+        osutil.BigIpOSUtil.mount_dvd(
+            osutil.BigIpOSUtil(), max_retry=6, chk_err=True
+        )
+        self.assertEqual(args[0].call_count, 1)
+        self.assertEqual(args[1].call_count, 1)
+
+
+class TestBigIpOSUtil_set_admin_access_to_ip(AgentTestCase):
+
+    @patch.object(shellutil, "run", return_value=0)
+    @patch.object(osutil.BigIpOSUtil,
+                  '_set_accept_admin_access_to_ip', return_value=None)
+    @patch.object(osutil.BigIpOSUtil,
+                  '_set_drop_admin_access_to_ip', return_value=None)
+    def test_success(self, *args):
+        osutil.BigIpOSUtil.set_admin_access_to_ip(
+            osutil.BigIpOSUtil(), '192.168.10.10'
+        )
+        self.assertEqual(args[0].call_count, 1)
+        self.assertEqual(args[1].call_count, 1)
+
+    @patch.object(shellutil, "run", return_value=0)
+    def test_accept_access(self, *args):
+        osutil.BigIpOSUtil._set_accept_admin_access_to_ip(
+            osutil.BigIpOSUtil(), '192.168.10.10'
+        )
+        self.assertEqual(args[0].call_count, 2)
+
+    @patch.object(shellutil, "run", return_value=0)
+    def test_drop_access(self, *args):
+        osutil.BigIpOSUtil._set_drop_admin_access_to_ip(
+            osutil.BigIpOSUtil(), '192.168.10.10'
+        )
+        self.assertEqual(args[0].call_count, 2)
+
+if __name__ == '__main__':
+    unittest.main()