mirror of
https://github.com/clearlinux/WALinuxAgent.git
synced 2026-09-05 13:22:05 +00:00
175 lines
7.3 KiB
Python
175 lines
7.3 KiB
Python
# 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()
|