36d7c0c403
These were discovered on python3 with fetch code that fails on errors. Probably could be provoked with particular sets of arguments to stat as well.
1035 lines
43 KiB
Python
1035 lines
43 KiB
Python
# -*- coding: utf-8 -*-
|
|
# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
|
|
# (c) 2016 Toshio Kuratomi <tkuratomi@ansible.com>
|
|
#
|
|
# This file is part of Ansible
|
|
#
|
|
# Ansible is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# Ansible is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
# Make coding more python3-ish
|
|
from __future__ import (absolute_import, division)
|
|
__metaclass__ = type
|
|
|
|
import errno
|
|
import json
|
|
import os
|
|
import sys
|
|
from io import BytesIO, StringIO
|
|
|
|
from units.mock.procenv import ModuleTestCase, swap_stdin_and_argv
|
|
|
|
from ansible.compat.tests import unittest
|
|
from ansible.compat.tests.mock import patch, MagicMock, mock_open, Mock, call
|
|
from ansible.module_utils.six.moves import builtins
|
|
|
|
realimport = builtins.__import__
|
|
|
|
class TestModuleUtilsBasic(ModuleTestCase):
|
|
|
|
def clear_modules(self, mods):
|
|
for mod in mods:
|
|
if mod in sys.modules:
|
|
del sys.modules[mod]
|
|
|
|
@patch.object(builtins, '__import__')
|
|
def test_module_utils_basic_import_syslog(self, mock_import):
|
|
def _mock_import(name, *args, **kwargs):
|
|
if name == 'syslog':
|
|
raise ImportError
|
|
return realimport(name, *args, **kwargs)
|
|
|
|
self.clear_modules(['syslog', 'ansible.module_utils.basic'])
|
|
mod = builtins.__import__('ansible.module_utils.basic')
|
|
self.assertTrue(mod.module_utils.basic.HAS_SYSLOG)
|
|
|
|
self.clear_modules(['syslog', 'ansible.module_utils.basic'])
|
|
mock_import.side_effect = _mock_import
|
|
mod = builtins.__import__('ansible.module_utils.basic')
|
|
self.assertFalse(mod.module_utils.basic.HAS_SYSLOG)
|
|
|
|
@patch.object(builtins, '__import__')
|
|
def test_module_utils_basic_import_selinux(self, mock_import):
|
|
def _mock_import(name, *args, **kwargs):
|
|
if name == 'selinux':
|
|
raise ImportError
|
|
return realimport(name, *args, **kwargs)
|
|
|
|
try:
|
|
self.clear_modules(['selinux', 'ansible.module_utils.basic'])
|
|
mod = builtins.__import__('ansible.module_utils.basic')
|
|
self.assertTrue(mod.module_utils.basic.HAVE_SELINUX)
|
|
except ImportError:
|
|
# no selinux on test system, so skip
|
|
pass
|
|
|
|
self.clear_modules(['selinux', 'ansible.module_utils.basic'])
|
|
mock_import.side_effect = _mock_import
|
|
mod = builtins.__import__('ansible.module_utils.basic')
|
|
self.assertFalse(mod.module_utils.basic.HAVE_SELINUX)
|
|
|
|
@patch.object(builtins, '__import__')
|
|
def test_module_utils_basic_import_json(self, mock_import):
|
|
def _mock_import(name, *args, **kwargs):
|
|
if name == 'json':
|
|
raise ImportError
|
|
elif name == 'simplejson':
|
|
return MagicMock()
|
|
return realimport(name, *args, **kwargs)
|
|
|
|
self.clear_modules(['json', 'ansible.module_utils.basic'])
|
|
mod = builtins.__import__('ansible.module_utils.basic')
|
|
|
|
self.clear_modules(['json', 'ansible.module_utils.basic'])
|
|
mock_import.side_effect = _mock_import
|
|
mod = builtins.__import__('ansible.module_utils.basic')
|
|
|
|
# FIXME: doesn't work yet
|
|
#@patch.object(builtins, 'bytes')
|
|
#def test_module_utils_basic_bytes(self, mock_bytes):
|
|
# mock_bytes.side_effect = NameError()
|
|
# from ansible.module_utils import basic
|
|
|
|
@patch.object(builtins, '__import__')
|
|
@unittest.skipIf(sys.version_info[0] >= 3, "literal_eval is available in every version of Python3")
|
|
def test_module_utils_basic_import_literal_eval(self, mock_import):
|
|
def _mock_import(name, *args, **kwargs):
|
|
try:
|
|
fromlist = kwargs.get('fromlist', args[2])
|
|
except IndexError:
|
|
fromlist = []
|
|
if name == 'ast' and 'literal_eval' in fromlist:
|
|
raise ImportError
|
|
return realimport(name, *args, **kwargs)
|
|
|
|
mock_import.side_effect = _mock_import
|
|
self.clear_modules(['ast', 'ansible.module_utils.basic'])
|
|
mod = builtins.__import__('ansible.module_utils.basic')
|
|
self.assertEqual(mod.module_utils.basic.literal_eval("'1'"), "1")
|
|
self.assertEqual(mod.module_utils.basic.literal_eval("1"), 1)
|
|
self.assertEqual(mod.module_utils.basic.literal_eval("-1"), -1)
|
|
self.assertEqual(mod.module_utils.basic.literal_eval("(1,2,3)"), (1,2,3))
|
|
self.assertEqual(mod.module_utils.basic.literal_eval("[1]"), [1])
|
|
self.assertEqual(mod.module_utils.basic.literal_eval("True"), True)
|
|
self.assertEqual(mod.module_utils.basic.literal_eval("False"), False)
|
|
self.assertEqual(mod.module_utils.basic.literal_eval("None"), None)
|
|
#self.assertEqual(mod.module_utils.basic.literal_eval('{"a": 1}'), dict(a=1))
|
|
self.assertRaises(ValueError, mod.module_utils.basic.literal_eval, "asdfasdfasdf")
|
|
|
|
@patch.object(builtins, '__import__')
|
|
def test_module_utils_basic_import_systemd_journal(self, mock_import):
|
|
def _mock_import(name, *args, **kwargs):
|
|
try:
|
|
fromlist = kwargs.get('fromlist', args[2])
|
|
except IndexError:
|
|
fromlist = []
|
|
if name == 'systemd' and 'journal' in fromlist:
|
|
raise ImportError
|
|
return realimport(name, *args, **kwargs)
|
|
|
|
self.clear_modules(['systemd', 'ansible.module_utils.basic'])
|
|
mod = builtins.__import__('ansible.module_utils.basic')
|
|
self.assertTrue(mod.module_utils.basic.has_journal)
|
|
|
|
self.clear_modules(['systemd', 'ansible.module_utils.basic'])
|
|
mock_import.side_effect = _mock_import
|
|
mod = builtins.__import__('ansible.module_utils.basic')
|
|
self.assertFalse(mod.module_utils.basic.has_journal)
|
|
|
|
def test_module_utils_basic_get_platform(self):
|
|
with patch('platform.system', return_value='foo'):
|
|
from ansible.module_utils.basic import get_platform
|
|
self.assertEqual(get_platform(), 'foo')
|
|
|
|
def test_module_utils_basic_get_distribution(self):
|
|
from ansible.module_utils.basic import get_distribution
|
|
|
|
with patch('platform.system', return_value='Foo'):
|
|
self.assertEqual(get_distribution(), None)
|
|
|
|
with patch('platform.system', return_value='Linux'):
|
|
with patch('platform.linux_distribution', return_value=["foo"]):
|
|
self.assertEqual(get_distribution(), "Foo")
|
|
|
|
with patch('os.path.isfile', return_value=True):
|
|
with patch('platform.linux_distribution', side_effect=[("AmazonFooBar",)]):
|
|
self.assertEqual(get_distribution(), "Amazonfoobar")
|
|
|
|
with patch('platform.linux_distribution', side_effect=(("",), ("AmazonFooBam",))):
|
|
self.assertEqual(get_distribution(), "Amazon")
|
|
|
|
with patch('platform.linux_distribution', side_effect=[("",),("",)]):
|
|
self.assertEqual(get_distribution(), "OtherLinux")
|
|
|
|
def _dist(distname='', version='', id='', supported_dists=(), full_distribution_name=1):
|
|
if supported_dists != ():
|
|
return ("Bar", "2", "Two")
|
|
else:
|
|
return ("", "", "")
|
|
|
|
with patch('platform.linux_distribution', side_effect=_dist):
|
|
self.assertEqual(get_distribution(), "Bar")
|
|
|
|
with patch('platform.linux_distribution', side_effect=Exception("boo")):
|
|
with patch('platform.dist', return_value=("bar", "2", "Two")):
|
|
self.assertEqual(get_distribution(), "Bar")
|
|
|
|
def test_module_utils_basic_get_distribution_version(self):
|
|
from ansible.module_utils.basic import get_distribution_version
|
|
|
|
with patch('platform.system', return_value='Foo'):
|
|
self.assertEqual(get_distribution_version(), None)
|
|
|
|
with patch('platform.system', return_value='Linux'):
|
|
with patch('platform.linux_distribution', return_value=("foo", "1", "One")):
|
|
self.assertEqual(get_distribution_version(), "1")
|
|
|
|
with patch('os.path.isfile', return_value=True):
|
|
def _dist(distname='', version='', id='', supported_dists=(), full_distribution_name=1):
|
|
if supported_dists != ():
|
|
return ("AmazonFooBar", "2", "")
|
|
else:
|
|
return ("", "", "")
|
|
|
|
with patch('platform.linux_distribution', side_effect=_dist):
|
|
self.assertEqual(get_distribution_version(), "2")
|
|
|
|
with patch('platform.linux_distribution', side_effect=Exception("boo")):
|
|
with patch('platform.dist', return_value=("bar", "3", "Three")):
|
|
self.assertEqual(get_distribution_version(), "3")
|
|
|
|
def test_module_utils_basic_load_platform_subclass(self):
|
|
class LinuxTest:
|
|
pass
|
|
|
|
class Foo(LinuxTest):
|
|
platform = "Linux"
|
|
distribution = None
|
|
|
|
class Bar(LinuxTest):
|
|
platform = "Linux"
|
|
distribution = "Bar"
|
|
|
|
from ansible.module_utils.basic import load_platform_subclass
|
|
|
|
# match just the platform class, not a specific distribution
|
|
with patch('ansible.module_utils.basic.get_platform', return_value="Linux"):
|
|
with patch('ansible.module_utils.basic.get_distribution', return_value=None):
|
|
self.assertIs(type(load_platform_subclass(LinuxTest)), Foo)
|
|
|
|
# match both the distribution and platform class
|
|
with patch('ansible.module_utils.basic.get_platform', return_value="Linux"):
|
|
with patch('ansible.module_utils.basic.get_distribution', return_value="Bar"):
|
|
self.assertIs(type(load_platform_subclass(LinuxTest)), Bar)
|
|
|
|
# if neither match, the fallback should be the top-level class
|
|
with patch('ansible.module_utils.basic.get_platform', return_value="Foo"):
|
|
with patch('ansible.module_utils.basic.get_distribution', return_value=None):
|
|
self.assertIs(type(load_platform_subclass(LinuxTest)), LinuxTest)
|
|
|
|
def test_module_utils_basic_json_dict_converters(self):
|
|
from ansible.module_utils.basic import json_dict_unicode_to_bytes, json_dict_bytes_to_unicode
|
|
|
|
test_data = dict(
|
|
item1 = u"Fóo",
|
|
item2 = [u"Bár", u"Bam"],
|
|
item3 = dict(sub1=u"Súb"),
|
|
item4 = (u"föo", u"bär", u"©"),
|
|
item5 = 42,
|
|
)
|
|
res = json_dict_unicode_to_bytes(test_data)
|
|
res2 = json_dict_bytes_to_unicode(res)
|
|
|
|
self.assertEqual(test_data, res2)
|
|
|
|
def test_module_utils_basic_get_module_path(self):
|
|
from ansible.module_utils.basic import get_module_path
|
|
with patch('os.path.realpath', return_value='/path/to/foo/'):
|
|
self.assertEqual(get_module_path(), '/path/to/foo')
|
|
|
|
def test_module_utils_basic_ansible_module_creation(self):
|
|
from ansible.module_utils import basic
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec=dict(),
|
|
)
|
|
|
|
arg_spec = dict(
|
|
foo = dict(required=True),
|
|
bar = dict(),
|
|
bam = dict(),
|
|
baz = dict(),
|
|
)
|
|
mut_ex = (('bar', 'bam'),)
|
|
req_to = (('bam', 'baz'),)
|
|
|
|
# should test ok
|
|
args = json.dumps(dict(ANSIBLE_MODULE_ARGS={"foo": "hello"}))
|
|
|
|
with swap_stdin_and_argv(stdin_data=args):
|
|
basic._ANSIBLE_ARGS = None
|
|
am = basic.AnsibleModule(
|
|
argument_spec = arg_spec,
|
|
mutually_exclusive = mut_ex,
|
|
required_together = req_to,
|
|
no_log=True,
|
|
check_invalid_arguments=False,
|
|
add_file_common_args=True,
|
|
supports_check_mode=True,
|
|
)
|
|
|
|
# FIXME: add asserts here to verify the basic config
|
|
|
|
# fail, because a required param was not specified
|
|
args = json.dumps(dict(ANSIBLE_MODULE_ARGS={}))
|
|
|
|
with swap_stdin_and_argv(stdin_data=args):
|
|
basic._ANSIBLE_ARGS = None
|
|
self.assertRaises(
|
|
SystemExit,
|
|
basic.AnsibleModule,
|
|
argument_spec = arg_spec,
|
|
mutually_exclusive = mut_ex,
|
|
required_together = req_to,
|
|
no_log=True,
|
|
check_invalid_arguments=False,
|
|
add_file_common_args=True,
|
|
supports_check_mode=True,
|
|
)
|
|
|
|
# fail because of mutually exclusive parameters
|
|
args = json.dumps(dict(ANSIBLE_MODULE_ARGS={"foo":"hello", "bar": "bad", "bam": "bad"}))
|
|
|
|
with swap_stdin_and_argv(stdin_data=args):
|
|
basic._ANSIBLE_ARGS = None
|
|
self.assertRaises(
|
|
SystemExit,
|
|
basic.AnsibleModule,
|
|
argument_spec = arg_spec,
|
|
mutually_exclusive = mut_ex,
|
|
required_together = req_to,
|
|
no_log=True,
|
|
check_invalid_arguments=False,
|
|
add_file_common_args=True,
|
|
supports_check_mode=True,
|
|
)
|
|
|
|
# fail because a param required due to another param was not specified
|
|
args = json.dumps(dict(ANSIBLE_MODULE_ARGS={"bam": "bad"}))
|
|
|
|
with swap_stdin_and_argv(stdin_data=args):
|
|
basic._ANSIBLE_ARGS = None
|
|
self.assertRaises(
|
|
SystemExit,
|
|
basic.AnsibleModule,
|
|
argument_spec = arg_spec,
|
|
mutually_exclusive = mut_ex,
|
|
required_together = req_to,
|
|
no_log=True,
|
|
check_invalid_arguments=False,
|
|
add_file_common_args=True,
|
|
supports_check_mode=True,
|
|
)
|
|
|
|
def test_module_utils_basic_ansible_module_type_check(self):
|
|
from ansible.module_utils import basic
|
|
|
|
arg_spec = dict(
|
|
foo = dict(type='float'),
|
|
foo2 = dict(type='float'),
|
|
foo3 = dict(type='float'),
|
|
bar = dict(type='int'),
|
|
bar2 = dict(type='int'),
|
|
)
|
|
|
|
# should test ok
|
|
args = json.dumps(dict(ANSIBLE_MODULE_ARGS={
|
|
"foo": 123.0, # float
|
|
"foo2": 123, # int
|
|
"foo3": "123", # string
|
|
"bar": 123, # int
|
|
"bar2": "123", # string
|
|
}))
|
|
|
|
with swap_stdin_and_argv(stdin_data=args):
|
|
basic._ANSIBLE_ARGS = None
|
|
am = basic.AnsibleModule(
|
|
argument_spec = arg_spec,
|
|
no_log=True,
|
|
check_invalid_arguments=False,
|
|
add_file_common_args=True,
|
|
supports_check_mode=True,
|
|
)
|
|
|
|
# fail, because bar does not accept floating point numbers
|
|
args = json.dumps(dict(ANSIBLE_MODULE_ARGS={"bar": 123.0}))
|
|
|
|
with swap_stdin_and_argv(stdin_data=args):
|
|
basic._ANSIBLE_ARGS = None
|
|
self.assertRaises(
|
|
SystemExit,
|
|
basic.AnsibleModule,
|
|
argument_spec = arg_spec,
|
|
no_log=True,
|
|
check_invalid_arguments=False,
|
|
add_file_common_args=True,
|
|
supports_check_mode=True,
|
|
)
|
|
|
|
def test_module_utils_basic_ansible_module_load_file_common_arguments(self):
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
am.selinux_mls_enabled = MagicMock()
|
|
am.selinux_mls_enabled.return_value = True
|
|
am.selinux_default_context = MagicMock()
|
|
am.selinux_default_context.return_value = 'unconfined_u:object_r:default_t:s0'.split(':', 3)
|
|
|
|
# with no params, the result should be an empty dict
|
|
res = am.load_file_common_arguments(params=dict())
|
|
self.assertEqual(res, dict())
|
|
|
|
base_params = dict(
|
|
path = '/path/to/file',
|
|
mode = 0o600,
|
|
owner = 'root',
|
|
group = 'root',
|
|
seuser = '_default',
|
|
serole = '_default',
|
|
setype = '_default',
|
|
selevel = '_default',
|
|
)
|
|
|
|
extended_params = base_params.copy()
|
|
extended_params.update(dict(
|
|
follow = True,
|
|
foo = 'bar',
|
|
))
|
|
|
|
final_params = base_params.copy()
|
|
final_params.update(dict(
|
|
path = '/path/to/real_file',
|
|
secontext=['unconfined_u', 'object_r', 'default_t', 's0'],
|
|
attributes=None,
|
|
))
|
|
|
|
# with the proper params specified, the returned dictionary should represent
|
|
# only those params which have something to do with the file arguments, excluding
|
|
# other params and updated as required with proper values which may have been
|
|
# massaged by the method
|
|
with patch('os.path.islink', return_value=True):
|
|
with patch('os.path.realpath', return_value='/path/to/real_file'):
|
|
res = am.load_file_common_arguments(params=extended_params)
|
|
self.assertEqual(res, final_params)
|
|
|
|
def test_module_utils_basic_ansible_module_selinux_mls_enabled(self):
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
basic.HAVE_SELINUX = False
|
|
self.assertEqual(am.selinux_mls_enabled(), False)
|
|
|
|
basic.HAVE_SELINUX = True
|
|
basic.selinux = Mock()
|
|
with patch.dict('sys.modules', {'selinux': basic.selinux}):
|
|
with patch('selinux.is_selinux_mls_enabled', return_value=0):
|
|
self.assertEqual(am.selinux_mls_enabled(), False)
|
|
with patch('selinux.is_selinux_mls_enabled', return_value=1):
|
|
self.assertEqual(am.selinux_mls_enabled(), True)
|
|
delattr(basic, 'selinux')
|
|
|
|
def test_module_utils_basic_ansible_module_selinux_initial_context(self):
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
am.selinux_mls_enabled = MagicMock()
|
|
am.selinux_mls_enabled.return_value = False
|
|
self.assertEqual(am.selinux_initial_context(), [None, None, None])
|
|
am.selinux_mls_enabled.return_value = True
|
|
self.assertEqual(am.selinux_initial_context(), [None, None, None, None])
|
|
|
|
def test_module_utils_basic_ansible_module_selinux_enabled(self):
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
# we first test the cases where the python selinux lib is
|
|
# not installed, which has two paths: one in which the system
|
|
# does have selinux installed (and the selinuxenabled command
|
|
# is present and returns 0 when run), or selinux is not installed
|
|
basic.HAVE_SELINUX = False
|
|
am.get_bin_path = MagicMock()
|
|
am.get_bin_path.return_value = '/path/to/selinuxenabled'
|
|
am.run_command = MagicMock()
|
|
am.run_command.return_value=(0, '', '')
|
|
self.assertRaises(SystemExit, am.selinux_enabled)
|
|
am.get_bin_path.return_value = None
|
|
self.assertEqual(am.selinux_enabled(), False)
|
|
|
|
# finally we test the case where the python selinux lib is installed,
|
|
# and both possibilities there (enabled vs. disabled)
|
|
basic.HAVE_SELINUX = True
|
|
basic.selinux = Mock()
|
|
with patch.dict('sys.modules', {'selinux': basic.selinux}):
|
|
with patch('selinux.is_selinux_enabled', return_value=0):
|
|
self.assertEqual(am.selinux_enabled(), False)
|
|
with patch('selinux.is_selinux_enabled', return_value=1):
|
|
self.assertEqual(am.selinux_enabled(), True)
|
|
delattr(basic, 'selinux')
|
|
|
|
def test_module_utils_basic_ansible_module_selinux_default_context(self):
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
am.selinux_initial_context = MagicMock(return_value=[None, None, None, None])
|
|
am.selinux_enabled = MagicMock(return_value=True)
|
|
|
|
# we first test the cases where the python selinux lib is not installed
|
|
basic.HAVE_SELINUX = False
|
|
self.assertEqual(am.selinux_default_context(path='/foo/bar'), [None, None, None, None])
|
|
|
|
# all following tests assume the python selinux bindings are installed
|
|
basic.HAVE_SELINUX = True
|
|
|
|
basic.selinux = Mock()
|
|
|
|
with patch.dict('sys.modules', {'selinux': basic.selinux}):
|
|
# next, we test with a mocked implementation of selinux.matchpathcon to simulate
|
|
# an actual context being found
|
|
with patch('selinux.matchpathcon', return_value=[0, 'unconfined_u:object_r:default_t:s0']):
|
|
self.assertEqual(am.selinux_default_context(path='/foo/bar'), ['unconfined_u', 'object_r', 'default_t', 's0'])
|
|
|
|
# we also test the case where matchpathcon returned a failure
|
|
with patch('selinux.matchpathcon', return_value=[-1, '']):
|
|
self.assertEqual(am.selinux_default_context(path='/foo/bar'), [None, None, None, None])
|
|
|
|
# finally, we test where an OSError occurred during matchpathcon's call
|
|
with patch('selinux.matchpathcon', side_effect=OSError):
|
|
self.assertEqual(am.selinux_default_context(path='/foo/bar'), [None, None, None, None])
|
|
|
|
delattr(basic, 'selinux')
|
|
|
|
def test_module_utils_basic_ansible_module_selinux_context(self):
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
am.selinux_initial_context = MagicMock(return_value=[None, None, None, None])
|
|
am.selinux_enabled = MagicMock(return_value=True)
|
|
|
|
# we first test the cases where the python selinux lib is not installed
|
|
basic.HAVE_SELINUX = False
|
|
self.assertEqual(am.selinux_context(path='/foo/bar'), [None, None, None, None])
|
|
|
|
# all following tests assume the python selinux bindings are installed
|
|
basic.HAVE_SELINUX = True
|
|
|
|
basic.selinux = Mock()
|
|
|
|
with patch.dict('sys.modules', {'selinux': basic.selinux}):
|
|
# next, we test with a mocked implementation of selinux.lgetfilecon_raw to simulate
|
|
# an actual context being found
|
|
with patch('selinux.lgetfilecon_raw', return_value=[0, 'unconfined_u:object_r:default_t:s0']):
|
|
self.assertEqual(am.selinux_context(path='/foo/bar'), ['unconfined_u', 'object_r', 'default_t', 's0'])
|
|
|
|
# we also test the case where matchpathcon returned a failure
|
|
with patch('selinux.lgetfilecon_raw', return_value=[-1, '']):
|
|
self.assertEqual(am.selinux_context(path='/foo/bar'), [None, None, None, None])
|
|
|
|
# finally, we test where an OSError occurred during matchpathcon's call
|
|
e = OSError()
|
|
e.errno = errno.ENOENT
|
|
with patch('selinux.lgetfilecon_raw', side_effect=e):
|
|
self.assertRaises(SystemExit, am.selinux_context, path='/foo/bar')
|
|
|
|
e = OSError()
|
|
with patch('selinux.lgetfilecon_raw', side_effect=e):
|
|
self.assertRaises(SystemExit, am.selinux_context, path='/foo/bar')
|
|
|
|
delattr(basic, 'selinux')
|
|
|
|
def test_module_utils_basic_ansible_module_is_special_selinux_path(self):
|
|
from ansible.module_utils import basic
|
|
|
|
args = json.dumps(dict(ANSIBLE_MODULE_ARGS={'_ansible_selinux_special_fs': "nfs,nfsd,foos"}))
|
|
|
|
with swap_stdin_and_argv(stdin_data=args):
|
|
basic._ANSIBLE_ARGS = None
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
def _mock_find_mount_point(path):
|
|
if path.startswith('/some/path'):
|
|
return '/some/path'
|
|
elif path.startswith('/weird/random/fstype'):
|
|
return '/weird/random/fstype'
|
|
return '/'
|
|
|
|
am.find_mount_point = MagicMock(side_effect=_mock_find_mount_point)
|
|
am.selinux_context = MagicMock(return_value=['foo_u', 'foo_r', 'foo_t', 's0'])
|
|
|
|
m = mock_open()
|
|
m.side_effect = OSError
|
|
|
|
with patch.object(builtins, 'open', m, create=True):
|
|
self.assertEqual(am.is_special_selinux_path('/some/path/that/should/be/nfs'), (False, None))
|
|
|
|
mount_data = [
|
|
'/dev/disk1 / ext4 rw,seclabel,relatime,data=ordered 0 0\n',
|
|
'1.1.1.1:/path/to/nfs /some/path nfs ro 0 0\n',
|
|
'whatever /weird/random/fstype foos rw 0 0\n',
|
|
]
|
|
|
|
# mock_open has a broken readlines() implementation apparently...
|
|
# this should work by default but doesn't, so we fix it
|
|
m = mock_open(read_data=''.join(mount_data))
|
|
m.return_value.readlines.return_value = mount_data
|
|
|
|
with patch.object(builtins, 'open', m, create=True):
|
|
self.assertEqual(am.is_special_selinux_path('/some/random/path'), (False, None))
|
|
self.assertEqual(am.is_special_selinux_path('/some/path/that/should/be/nfs'), (True, ['foo_u', 'foo_r', 'foo_t', 's0']))
|
|
self.assertEqual(am.is_special_selinux_path('/weird/random/fstype/path'), (True, ['foo_u', 'foo_r', 'foo_t', 's0']))
|
|
|
|
def test_module_utils_basic_ansible_module_user_and_group(self):
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
mock_stat = MagicMock()
|
|
mock_stat.st_uid = 0
|
|
mock_stat.st_gid = 0
|
|
|
|
with patch('os.lstat', return_value=mock_stat):
|
|
self.assertEqual(am.user_and_group('/path/to/file'), (0, 0))
|
|
|
|
def test_module_utils_basic_ansible_module_find_mount_point(self):
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
def _mock_ismount(path):
|
|
if path == b'/':
|
|
return True
|
|
return False
|
|
|
|
with patch('os.path.ismount', side_effect=_mock_ismount):
|
|
self.assertEqual(am.find_mount_point('/root/fs/../mounted/path/to/whatever'), '/')
|
|
|
|
def _mock_ismount(path):
|
|
if path == b'/subdir/mount':
|
|
return True
|
|
if path == b'/':
|
|
return True
|
|
return False
|
|
|
|
with patch('os.path.ismount', side_effect=_mock_ismount):
|
|
self.assertEqual(am.find_mount_point('/subdir/mount/path/to/whatever'), '/subdir/mount')
|
|
|
|
def test_module_utils_basic_ansible_module_set_context_if_different(self):
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
basic.HAVE_SELINUX = False
|
|
|
|
am.selinux_enabled = MagicMock(return_value=False)
|
|
self.assertEqual(am.set_context_if_different('/path/to/file', ['foo_u', 'foo_r', 'foo_t', 's0'], True), True)
|
|
self.assertEqual(am.set_context_if_different('/path/to/file', ['foo_u', 'foo_r', 'foo_t', 's0'], False), False)
|
|
|
|
basic.HAVE_SELINUX = True
|
|
|
|
am.selinux_enabled = MagicMock(return_value=True)
|
|
am.selinux_context = MagicMock(return_value=['bar_u', 'bar_r', None, None])
|
|
am.is_special_selinux_path = MagicMock(return_value=(False, None))
|
|
|
|
basic.selinux = Mock()
|
|
with patch.dict('sys.modules', {'selinux': basic.selinux}):
|
|
with patch('selinux.lsetfilecon', return_value=0) as m:
|
|
self.assertEqual(am.set_context_if_different('/path/to/file', ['foo_u', 'foo_r', 'foo_t', 's0'], False), True)
|
|
m.assert_called_with('/path/to/file', 'foo_u:foo_r:foo_t:s0')
|
|
m.reset_mock()
|
|
am.check_mode = True
|
|
self.assertEqual(am.set_context_if_different('/path/to/file', ['foo_u', 'foo_r', 'foo_t', 's0'], False), True)
|
|
self.assertEqual(m.called, False)
|
|
am.check_mode = False
|
|
|
|
with patch('selinux.lsetfilecon', return_value=1) as m:
|
|
self.assertRaises(SystemExit, am.set_context_if_different, '/path/to/file', ['foo_u', 'foo_r', 'foo_t', 's0'], True)
|
|
|
|
with patch('selinux.lsetfilecon', side_effect=OSError) as m:
|
|
self.assertRaises(SystemExit, am.set_context_if_different, '/path/to/file', ['foo_u', 'foo_r', 'foo_t', 's0'], True)
|
|
|
|
am.is_special_selinux_path = MagicMock(return_value=(True, ['sp_u', 'sp_r', 'sp_t', 's0']))
|
|
|
|
with patch('selinux.lsetfilecon', return_value=0) as m:
|
|
self.assertEqual(am.set_context_if_different('/path/to/file', ['foo_u', 'foo_r', 'foo_t', 's0'], False), True)
|
|
m.assert_called_with('/path/to/file', 'sp_u:sp_r:sp_t:s0')
|
|
|
|
delattr(basic, 'selinux')
|
|
|
|
def test_module_utils_basic_ansible_module_set_owner_if_different(self):
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
self.assertEqual(am.set_owner_if_different('/path/to/file', None, True), True)
|
|
self.assertEqual(am.set_owner_if_different('/path/to/file', None, False), False)
|
|
|
|
am.user_and_group = MagicMock(return_value=(500, 500))
|
|
|
|
with patch('os.lchown', return_value=None) as m:
|
|
self.assertEqual(am.set_owner_if_different('/path/to/file', 0, False), True)
|
|
m.assert_called_with(b'/path/to/file', 0, -1)
|
|
|
|
def _mock_getpwnam(*args, **kwargs):
|
|
mock_pw = MagicMock()
|
|
mock_pw.pw_uid = 0
|
|
return mock_pw
|
|
|
|
m.reset_mock()
|
|
with patch('pwd.getpwnam', side_effect=_mock_getpwnam):
|
|
self.assertEqual(am.set_owner_if_different('/path/to/file', 'root', False), True)
|
|
m.assert_called_with(b'/path/to/file', 0, -1)
|
|
|
|
with patch('pwd.getpwnam', side_effect=KeyError):
|
|
self.assertRaises(SystemExit, am.set_owner_if_different, '/path/to/file', 'root', False)
|
|
|
|
m.reset_mock()
|
|
am.check_mode = True
|
|
self.assertEqual(am.set_owner_if_different('/path/to/file', 0, False), True)
|
|
self.assertEqual(m.called, False)
|
|
am.check_mode = False
|
|
|
|
with patch('os.lchown', side_effect=OSError) as m:
|
|
self.assertRaises(SystemExit, am.set_owner_if_different, '/path/to/file', 'root', False)
|
|
|
|
def test_module_utils_basic_ansible_module_set_group_if_different(self):
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
self.assertEqual(am.set_group_if_different('/path/to/file', None, True), True)
|
|
self.assertEqual(am.set_group_if_different('/path/to/file', None, False), False)
|
|
|
|
am.user_and_group = MagicMock(return_value=(500, 500))
|
|
|
|
with patch('os.lchown', return_value=None) as m:
|
|
self.assertEqual(am.set_group_if_different('/path/to/file', 0, False), True)
|
|
m.assert_called_with(b'/path/to/file', -1, 0)
|
|
|
|
def _mock_getgrnam(*args, **kwargs):
|
|
mock_gr = MagicMock()
|
|
mock_gr.gr_gid = 0
|
|
return mock_gr
|
|
|
|
m.reset_mock()
|
|
with patch('grp.getgrnam', side_effect=_mock_getgrnam):
|
|
self.assertEqual(am.set_group_if_different('/path/to/file', 'root', False), True)
|
|
m.assert_called_with(b'/path/to/file', -1, 0)
|
|
|
|
with patch('grp.getgrnam', side_effect=KeyError):
|
|
self.assertRaises(SystemExit, am.set_group_if_different, '/path/to/file', 'root', False)
|
|
|
|
m.reset_mock()
|
|
am.check_mode = True
|
|
self.assertEqual(am.set_group_if_different('/path/to/file', 0, False), True)
|
|
self.assertEqual(m.called, False)
|
|
am.check_mode = False
|
|
|
|
with patch('os.lchown', side_effect=OSError) as m:
|
|
self.assertRaises(SystemExit, am.set_group_if_different, '/path/to/file', 'root', False)
|
|
|
|
@patch('tempfile.mkstemp')
|
|
@patch('os.umask')
|
|
@patch('shutil.copyfileobj')
|
|
@patch('shutil.move')
|
|
@patch('shutil.copy2')
|
|
@patch('os.rename')
|
|
@patch('pwd.getpwuid')
|
|
@patch('os.getuid')
|
|
@patch('os.environ')
|
|
@patch('os.getlogin')
|
|
@patch('os.chown')
|
|
@patch('os.chmod')
|
|
@patch('os.stat')
|
|
@patch('os.path.exists')
|
|
@patch('os.close')
|
|
def test_module_utils_basic_ansible_module_atomic_move(
|
|
self,
|
|
_os_close,
|
|
_os_path_exists,
|
|
_os_stat,
|
|
_os_chmod,
|
|
_os_chown,
|
|
_os_getlogin,
|
|
_os_environ,
|
|
_os_getuid,
|
|
_pwd_getpwuid,
|
|
_os_rename,
|
|
_shutil_copy2,
|
|
_shutil_move,
|
|
_shutil_copyfileobj,
|
|
_os_umask,
|
|
_tempfile_mkstemp):
|
|
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
environ = dict()
|
|
_os_environ.__getitem__ = environ.__getitem__
|
|
_os_environ.__setitem__ = environ.__setitem__
|
|
|
|
am.selinux_enabled = MagicMock()
|
|
am.selinux_context = MagicMock()
|
|
am.selinux_default_context = MagicMock()
|
|
am.set_context_if_different = MagicMock()
|
|
|
|
# test destination does not exist, no selinux, login name = 'root',
|
|
# no environment, os.rename() succeeds
|
|
_os_path_exists.side_effect = [False, False]
|
|
_os_getlogin.return_value = 'root'
|
|
_os_getuid.return_value = 0
|
|
_pwd_getpwuid.return_value = ('root', '', 0, 0, '', '', '')
|
|
_os_rename.return_value = None
|
|
_os_umask.side_effect = [18, 0]
|
|
am.selinux_enabled.return_value = False
|
|
_os_chmod.reset_mock()
|
|
_os_chown.reset_mock()
|
|
am.set_context_if_different.reset_mock()
|
|
am.atomic_move('/path/to/src', '/path/to/dest')
|
|
_os_rename.assert_called_with(b'/path/to/src', b'/path/to/dest')
|
|
self.assertEqual(_os_chmod.call_args_list, [call(b'/path/to/dest', basic.DEFAULT_PERM & ~18)])
|
|
|
|
# same as above, except selinux_enabled
|
|
_os_path_exists.side_effect = [False, False]
|
|
_os_getlogin.return_value = 'root'
|
|
_os_getuid.return_value = 0
|
|
_pwd_getpwuid.return_value = ('root', '', 0, 0, '', '', '')
|
|
_os_rename.return_value = None
|
|
_os_umask.side_effect = [18, 0]
|
|
mock_context = MagicMock()
|
|
am.selinux_default_context.return_value = mock_context
|
|
am.selinux_enabled.return_value = True
|
|
_os_chmod.reset_mock()
|
|
_os_chown.reset_mock()
|
|
am.set_context_if_different.reset_mock()
|
|
am.selinux_default_context.reset_mock()
|
|
am.atomic_move('/path/to/src', '/path/to/dest')
|
|
_os_rename.assert_called_with(b'/path/to/src', b'/path/to/dest')
|
|
self.assertEqual(_os_chmod.call_args_list, [call(b'/path/to/dest', basic.DEFAULT_PERM & ~18)])
|
|
self.assertEqual(am.selinux_default_context.call_args_list, [call('/path/to/dest')])
|
|
self.assertEqual(am.set_context_if_different.call_args_list, [call('/path/to/dest', mock_context, False)])
|
|
|
|
# now with dest present, no selinux, also raise OSError when using
|
|
# os.getlogin() to test corner case with no tty
|
|
_os_path_exists.side_effect = [True, True]
|
|
_os_getlogin.side_effect = OSError()
|
|
_os_getuid.return_value = 0
|
|
_pwd_getpwuid.return_value = ('root', '', 0, 0, '', '', '')
|
|
_os_rename.return_value = None
|
|
_os_umask.side_effect = [18, 0]
|
|
environ['LOGNAME'] = 'root'
|
|
stat1 = MagicMock()
|
|
stat1.st_mode = 0o0644
|
|
stat1.st_uid = 0
|
|
stat1.st_gid = 0
|
|
_os_stat.side_effect = [stat1,]
|
|
am.selinux_enabled.return_value = False
|
|
_os_chmod.reset_mock()
|
|
_os_chown.reset_mock()
|
|
am.set_context_if_different.reset_mock()
|
|
am.atomic_move('/path/to/src', '/path/to/dest')
|
|
_os_rename.assert_called_with(b'/path/to/src', b'/path/to/dest')
|
|
|
|
# dest missing, selinux enabled
|
|
_os_path_exists.side_effect = [True, True]
|
|
_os_getlogin.return_value = 'root'
|
|
_os_getuid.return_value = 0
|
|
_pwd_getpwuid.return_value = ('root', '', 0, 0, '', '', '')
|
|
_os_rename.return_value = None
|
|
_os_umask.side_effect = [18, 0]
|
|
stat1 = MagicMock()
|
|
stat1.st_mode = 0o0644
|
|
stat1.st_uid = 0
|
|
stat1.st_gid = 0
|
|
_os_stat.side_effect = [stat1,]
|
|
mock_context = MagicMock()
|
|
am.selinux_context.return_value = mock_context
|
|
am.selinux_enabled.return_value = True
|
|
_os_chmod.reset_mock()
|
|
_os_chown.reset_mock()
|
|
am.set_context_if_different.reset_mock()
|
|
am.selinux_default_context.reset_mock()
|
|
am.atomic_move('/path/to/src', '/path/to/dest')
|
|
_os_rename.assert_called_with(b'/path/to/src', b'/path/to/dest')
|
|
self.assertEqual(am.selinux_context.call_args_list, [call('/path/to/dest')])
|
|
self.assertEqual(am.set_context_if_different.call_args_list, [call('/path/to/dest', mock_context, False)])
|
|
|
|
# now testing with exceptions raised
|
|
# have os.stat raise OSError which is not EPERM
|
|
_os_stat.side_effect = OSError()
|
|
_os_path_exists.side_effect = [True, True]
|
|
_os_getlogin.return_value = 'root'
|
|
_os_getuid.return_value = 0
|
|
_pwd_getpwuid.return_value = ('root', '', 0, 0, '', '', '')
|
|
_os_rename.return_value = None
|
|
_os_umask.side_effect = [18, 0]
|
|
self.assertRaises(OSError, am.atomic_move, '/path/to/src', '/path/to/dest')
|
|
|
|
# and now have os.stat return EPERM, which should not fail
|
|
_os_stat.side_effect = OSError(errno.EPERM, 'testing os stat with EPERM')
|
|
_os_path_exists.side_effect = [True, True]
|
|
_os_getlogin.return_value = 'root'
|
|
_os_getuid.return_value = 0
|
|
_pwd_getpwuid.return_value = ('root', '', 0, 0, '', '', '')
|
|
_os_rename.return_value = None
|
|
_os_umask.side_effect = [18, 0]
|
|
# FIXME: we don't assert anything here yet
|
|
am.atomic_move('/path/to/src', '/path/to/dest')
|
|
|
|
# now we test os.rename() raising errors...
|
|
# first we test with a bad errno to verify it bombs out
|
|
_os_path_exists.side_effect = [False, False]
|
|
_os_getlogin.return_value = 'root'
|
|
_os_getuid.return_value = 0
|
|
_pwd_getpwuid.return_value = ('root', '', 0, 0, '', '', '')
|
|
_os_umask.side_effect = [18, 0]
|
|
_os_rename.side_effect = OSError(errno.EIO, 'failing with EIO')
|
|
self.assertRaises(SystemExit, am.atomic_move, '/path/to/src', '/path/to/dest')
|
|
|
|
# next we test with EPERM so it continues to the alternate code for moving
|
|
# test with mkstemp raising an error first
|
|
_os_path_exists.side_effect = [False, False]
|
|
_os_getlogin.return_value = 'root'
|
|
_os_getuid.return_value = 0
|
|
_os_close.return_value = None
|
|
_pwd_getpwuid.return_value = ('root', '', 0, 0, '', '', '')
|
|
_os_umask.side_effect = [18, 0]
|
|
_os_rename.side_effect = [OSError(errno.EPERM, 'failing with EPERM'), None]
|
|
_tempfile_mkstemp.return_value = None
|
|
_tempfile_mkstemp.side_effect = OSError()
|
|
am.selinux_enabled.return_value = False
|
|
self.assertRaises(SystemExit, am.atomic_move, '/path/to/src', '/path/to/dest')
|
|
|
|
# then test with it creating a temp file
|
|
_os_path_exists.side_effect = [False, False, False]
|
|
_os_getlogin.return_value = 'root'
|
|
_os_getuid.return_value = 0
|
|
_pwd_getpwuid.return_value = ('root', '', 0, 0, '', '', '')
|
|
_os_umask.side_effect = [18, 0]
|
|
_os_rename.side_effect = [OSError(errno.EPERM, 'failing with EPERM'), None]
|
|
mock_stat1 = MagicMock()
|
|
mock_stat2 = MagicMock()
|
|
mock_stat3 = MagicMock()
|
|
_os_stat.return_value = [mock_stat1, mock_stat2, mock_stat3]
|
|
_os_stat.side_effect = None
|
|
_tempfile_mkstemp.return_value = (None, '/path/to/tempfile')
|
|
_tempfile_mkstemp.side_effect = None
|
|
am.selinux_enabled.return_value = False
|
|
# FIXME: we don't assert anything here yet
|
|
am.atomic_move('/path/to/src', '/path/to/dest')
|
|
|
|
# same as above, but with selinux enabled
|
|
_os_path_exists.side_effect = [False, False, False]
|
|
_os_getlogin.return_value = 'root'
|
|
_os_getuid.return_value = 0
|
|
_pwd_getpwuid.return_value = ('root', '', 0, 0, '', '', '')
|
|
_os_umask.side_effect = [18, 0]
|
|
_os_rename.side_effect = [OSError(errno.EPERM, 'failing with EPERM'), None]
|
|
_tempfile_mkstemp.return_value = (None, None)
|
|
mock_context = MagicMock()
|
|
am.selinux_default_context.return_value = mock_context
|
|
am.selinux_enabled.return_value = True
|
|
am.atomic_move('/path/to/src', '/path/to/dest')
|
|
|
|
def test_module_utils_basic_ansible_module__symbolic_mode_to_octal(self):
|
|
|
|
from ansible.module_utils import basic
|
|
basic._ANSIBLE_ARGS = None
|
|
|
|
am = basic.AnsibleModule(
|
|
argument_spec = dict(),
|
|
)
|
|
|
|
mock_stat = MagicMock()
|
|
|
|
# FIXME: trying many more combinations here would be good
|
|
# directory, give full perms to all, then one group at a time
|
|
mock_stat.st_mode = 0o040000
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'a+rwx'), 0o0777)
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'u+rwx,g+rwx,o+rwx'), 0o0777)
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'o+rwx'), 0o0007)
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'g+rwx'), 0o0070)
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'u+rwx'), 0o0700)
|
|
|
|
# same as above, but in reverse so removing permissions
|
|
mock_stat.st_mode = 0o040777
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'a-rwx'), 0o0000)
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'u-rwx,g-rwx,o-rwx'), 0o0000)
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'o-rwx'), 0o0770)
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'g-rwx'), 0o0707)
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'u-rwx'), 0o0077)
|
|
|
|
# now using absolute assignment
|
|
mock_stat.st_mode = 0o040000
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'a=rwx'), 0o0777)
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'u=rwx,g=rwx,o=rwx'), 0o0777)
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'o=rwx'), 0o0007)
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'g=rwx'), 0o0070)
|
|
self.assertEqual(am._symbolic_mode_to_octal(mock_stat, 'u=rwx'), 0o0700)
|
|
|
|
# invalid modes
|
|
mock_stat.st_mode = 0o040000
|
|
self.assertRaises(ValueError, am._symbolic_mode_to_octal, mock_stat, 'a=foo')
|