summaryrefslogtreecommitdiff
path: root/tests/test_osutil.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_osutil.py')
-rw-r--r--tests/test_osutil.py174
1 files changed, 174 insertions, 0 deletions
diff --git a/tests/test_osutil.py b/tests/test_osutil.py
new file mode 100644
index 0000000..95b8e17
--- /dev/null
+++ b/tests/test_osutil.py
@@ -0,0 +1,174 @@
+# Copyright 2014 Microsoft Corporation
+#
+# 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+
+#
+# Implements parts of RFC 2131, 1541, 1497 and
+# http://msdn.microsoft.com/en-us/library/cc227282%28PROT.10%29.aspx
+# http://msdn.microsoft.com/en-us/library/cc227259%28PROT.13%29.aspx
+
+import tests.env as env
+from tests.tools import *
+import uuid
+import unittest
+import os
+import shutil
+import time
+import azurelinuxagent.utils.fileutil as fileutil
+import azurelinuxagent.utils.shellutil as shellutil
+import azurelinuxagent.conf as conf
+from azurelinuxagent.utils.osutil import OSUTIL, OSUtilError
+import test
+
+class TestOSUtil(unittest.TestCase):
+ def test_current_distro(self):
+ self.assertNotEquals(None, OSUTIL)
+
+mount_list_sample="""\
+/dev/sda1 on / type ext4 (rw)
+proc on /proc type proc (rw)
+sysfs on /sys type sysfs (rw)
+devpts on /dev/pts type devpts (rw,gid=5,mode=620)
+tmpfs on /dev/shm type tmpfs (rw,rootcontext="system_u:object_r:tmpfs_t:s0")
+none on /proc/sys/fs/binfmt_misc type binfmt_misc (rw)
+/dev/sdb1 on /mnt/resource type ext4 (rw)
+"""
+
+class TestCurrOS(unittest.TestCase):
+#class TestCurrOS(object):
+ def test_get_paths(self):
+ self.assertNotEquals(None, OSUTIL.get_home())
+ self.assertNotEquals(None, OSUTIL.get_lib_dir())
+ self.assertNotEquals(None, OSUTIL.get_agent_pid_file_path())
+ self.assertNotEquals(None, OSUTIL.get_conf_file_path())
+ self.assertNotEquals(None, OSUTIL.get_dvd_mount_point())
+ self.assertNotEquals(None, OSUTIL.get_ovf_env_file_path_on_dvd())
+
+ @mock(fileutil, 'write_file', MockFunc())
+ @mock(fileutil, 'append_file', MockFunc())
+ @mock(fileutil, 'chmod', MockFunc())
+ @mock(fileutil, 'read_file', MockFunc(retval=''))
+ @mock(shellutil, 'run', MockFunc())
+ @mock(shellutil, 'run_get_output', MockFunc(retval=[0, '']))
+ def test_update_user_account(self):
+ OSUTIL.useradd('foo')
+ OSUTIL.chpasswd('foo', 'bar')
+ OSUTIL.del_account('foo')
+
+ @mock(fileutil, 'read_file', MockFunc(retval='root::::'))
+ @mock(fileutil, 'write_file', MockFunc())
+ def test_delete_root_password(self):
+ OSUTIL.del_root_password()
+ self.assertEquals('root:*LOCK*:14600::::::',
+ fileutil.write_file.args[1])
+
+ def test_cert_operation(self):
+ if os.path.isfile('/tmp/test.prv'):
+ os.remove('/tmp/test.prv')
+ shutil.copyfile(os.path.join(env.test_root, 'test.prv'),
+ '/tmp/test.prv')
+ if os.path.isfile('/tmp/test.crt'):
+ os.remove('/tmp/test.crt')
+ shutil.copyfile(os.path.join(env.test_root, 'test.crt'),
+ '/tmp/test.crt')
+ pub1 = OSUTIL.get_pubkey_from_prv('/tmp/test.prv')
+ pub2 = OSUTIL.get_pubkey_from_crt('/tmp/test.crt')
+ self.assertEquals(pub1, pub2)
+ thumbprint = OSUTIL.get_thumbprint_from_crt('/tmp/test.crt')
+ self.assertEquals('33B0ABCE4673538650971C10F7D7397E71561F35', thumbprint)
+
+ def test_selinux(self):
+ if not OSUTIL.is_selinux_system():
+ return
+ isrunning = OSUTIL.is_selinux_enforcing()
+ if not OSUTIL.is_selinux_enforcing():
+ OSUTIL.set_selinux_enforce(0)
+ self.assertEquals(False, OSUTIL.is_selinux_enforcing())
+ OSUTIL.set_selinux_enforce(1)
+ self.assertEquals(True, OSUTIL.is_selinux_enforcing())
+ if os.path.isfile('/tmp/abc'):
+ os.remove('/tmp/abc')
+ fileutil.write_file('/tmp/abc', '')
+ OSUTIL.set_selinux_context('/tmp/abc','unconfined_u:object_r:ssh_home_t:s')
+ OSUTIL.set_selinux_enforce(1 if isrunning else 0)
+
+ @mock(shellutil, 'run_get_output', MockFunc(retval=[0, '']))
+ @mock(fileutil, 'write_file', MockFunc())
+ def test_network_operation(self):
+ OSUTIL.start_network()
+ OSUTIL.allow_dhcp_broadcast()
+ OSUTIL.gen_transport_cert()
+ mac = OSUTIL.get_mac_addr()
+ self.assertNotEquals(None, mac)
+ OSUTIL.is_missing_default_route()
+ OSUTIL.set_route_for_dhcp_broadcast('api')
+ OSUTIL.remove_route_for_dhcp_broadcast('api')
+ OSUTIL.route_add('', '', '')
+ OSUTIL.get_dhcp_pid()
+ OSUTIL.set_hostname('api')
+ OSUTIL.publish_hostname('api')
+
+ @mock(OSUTIL, 'get_home', MockFunc(retval='/tmp/home'))
+ @mock(OSUTIL, 'get_pubkey_from_prv', MockFunc(retval=''))
+ @mock(fileutil, 'chowner', MockFunc())
+ def test_deploy_key(self):
+ if os.path.isdir('/tmp/home'):
+ shutil.rmtree('/tmp/home')
+ fileutil.write_file('/tmp/foo.prv', '')
+ OSUTIL.deploy_ssh_keypair("foo", ('$HOME/.ssh/id_rsa', 'foo'))
+ OSUTIL.deploy_ssh_pubkey("foo", ('$HOME/.ssh/authorized_keys', None,
+ 'ssh-rsa asdf'))
+ OSUTIL.deploy_ssh_pubkey("foo", ('$HOME/.ssh/authorized_keys', 'foo',
+ 'ssh-rsa asdf'))
+ self.assertRaises(OSUtilError, OSUTIL.deploy_ssh_pubkey, "foo",
+ ('$HOME/.ssh/authorized_keys', 'foo','hehe-rsa asdf'))
+ self.assertTrue(os.path.isfile('/tmp/home/.ssh/id_rsa'))
+ self.assertTrue(os.path.isfile('/tmp/home/.ssh/id_rsa.pub'))
+ self.assertTrue(os.path.isfile('/tmp/home/.ssh/authorized_keys'))
+
+ @mock(shellutil, 'run_get_output', MockFunc(retval=[0, '']))
+ @mock(OSUTIL, 'get_sshd_conf_file_path', MockFunc(retval='/tmp/sshd_config'))
+ def test_ssh_operation(self):
+ shellutil.run_get_output.retval=[0,
+ '2048 f1:fe:14:66:9d:46:9a:60:8b:8c:'
+ '80:43:39:1c:20:9e root@api (RSA)']
+ sshd_conf = OSUTIL.get_sshd_conf_file_path()
+ self.assertEquals('/tmp/sshd_config', sshd_conf)
+ if os.path.isfile(sshd_conf):
+ os.remove(sshd_conf)
+ shutil.copyfile(os.path.join(env.test_root, 'sshd_config'), sshd_conf)
+ OSUTIL.set_ssh_client_alive_interval()
+ OSUTIL.conf_sshd(True)
+ self.assertTrue(simple_file_grep(sshd_conf,
+ 'PasswordAuthentication no'))
+ self.assertTrue(simple_file_grep(sshd_conf,
+ 'ChallengeResponseAuthentication no'))
+ self.assertTrue(simple_file_grep(sshd_conf,
+ 'ClientAliveInterval 180'))
+
+ @mock(shellutil, 'run_get_output', MockFunc(retval=[0, '']))
+ @mock(OSUTIL, 'get_dvd_device', MockFunc(retval=[0, 'abc']))
+ @mock(OSUTIL, 'get_mount_point', MockFunc(retval='/tmp/cdrom'))
+ def test_mount(self):
+ OSUTIL.mount_dvd()
+ OSUTIL.umount_dvd()
+ mount_point = OSUTIL.get_mount_point(mount_list_sample, '/dev/sda')
+ self.assertNotEquals(None, mount_point)
+
+ def test_getdvd(self):
+ fileutil.write_file("/tmp/sr0", '')
+ OSUTIL.get_dvd_device(dev_dir='/tmp')
+
+if __name__ == '__main__':
+ unittest.main()