salt/tests/unit/beacons/test_wtmp.py
Erik Johnson 80dcb44a0d
Merge remote-tracking branch 'upstream/2018.3' into merge-fluorine
Conflicts:
- salt/utils/mac_utils.py
- tests/unit/config/test_config.py
- tests/unit/loader/test_globals.py
- tests/unit/ssh/test_ssh.py
- tests/unit/templates/test_jinja.py
- tests/unit/utils/vmware/test_host.py
- tests/unit/utils/vmware/test_license.py
2018-08-17 07:52:59 -05:00

211 lines
10 KiB
Python

# coding: utf-8
# Python libs
from __future__ import absolute_import
import datetime
import logging
# Salt testing libs
from tests.support.unit import skipIf, TestCase
from tests.support.mock import NO_MOCK, NO_MOCK_REASON, patch, MagicMock, mock_open
from tests.support.mixins import LoaderModuleMockMixin
# Salt libs
import salt.beacons.wtmp as wtmp
from salt.ext import six
# pylint: disable=import-error
try:
import dateutil.parser as dateutil_parser # pylint: disable=unused-import
_TIME_SUPPORTED = True
except ImportError:
_TIME_SUPPORTED = False
raw = b'\x07\x00\x00\x00H\x18\x00\x00pts/14\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00s/14gareth\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00::1\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x13I\xc5YZf\x05\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
pack = (7, 6216, b'pts/14\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00', b's/14', b'gareth\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00', b'::1\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00', 0, 0, 0, 1506101523, 353882, 0, 0, 0, 16777216)
log = logging.getLogger(__name__)
@skipIf(NO_MOCK, NO_MOCK_REASON)
class WTMPBeaconTestCase(TestCase, LoaderModuleMockMixin):
'''
Test case for salt.beacons.[s]
'''
def setup_loader_modules(self):
return {
wtmp: {
'__context__': {'wtmp.loc': 2},
'__salt__': {},
}
}
def test_non_list_config(self):
config = {}
ret = wtmp.validate(config)
self.assertEqual(ret, (False, 'Configuration for wtmp beacon must'
' be a list.'))
def test_empty_config(self):
config = [{}]
ret = wtmp.validate(config)
self.assertEqual(ret, (True, 'Valid beacon configuration'))
def test_no_match(self):
config = [{'users': {'gareth': {'time_range': {'end': '09-22-2017 5pm',
'start': '09-22-2017 3pm'}}}}
]
ret = wtmp.validate(config)
self.assertEqual(ret, (True, 'Valid beacon configuration'))
with patch('salt.utils.files.fopen', mock_open(b'')) as m_open:
ret = wtmp.beacon(config)
call_args = next(six.itervalues(m_open.filehandles))[0].call.args
assert call_args == (wtmp.WTMP, 'rb'), call_args
assert ret == [], ret
def test_invalid_users(self):
config = [{'users': ['gareth']}]
ret = wtmp.validate(config)
self.assertEqual(ret, (False, 'User configuration for wtmp beacon must be a dictionary.'))
def test_invalid_groups(self):
config = [{'groups': ['docker']}]
ret = wtmp.validate(config)
self.assertEqual(ret, (False, 'Group configuration for wtmp beacon must be a dictionary.'))
def test_default_invalid_time_range(self):
config = [{'defaults': {'time_range': {'start': '3pm'}}}]
ret = wtmp.validate(config)
self.assertEqual(ret, (False, 'The time_range parameter for wtmp beacon must contain start & end options.'))
def test_users_invalid_time_range(self):
config = [{'users': {'gareth': {'time_range': {'start': '3pm'}}}}]
ret = wtmp.validate(config)
self.assertEqual(ret, (False, 'The time_range parameter for wtmp beacon must contain start & end options.'))
def test_groups_invalid_time_range(self):
config = [{'groups': {'docker': {'time_range': {'start': '3pm'}}}}]
ret = wtmp.validate(config)
self.assertEqual(ret, (False, 'The time_range parameter for wtmp beacon must contain start & end options.'))
def test_match(self):
with patch('salt.utils.files.fopen',
mock_open(read_data=raw)):
with patch('struct.unpack',
MagicMock(return_value=pack)):
config = [{'users': {'gareth': {}}}]
ret = wtmp.validate(config)
self.assertEqual(ret, (True, 'Valid beacon configuration'))
_expected = [{'PID': 6216,
'action': 'login',
'line': 'pts/14',
'session': 0,
'time': 0,
'exit_status': 0,
'inittab': 's/14',
'type': 7,
'addr': 1506101523,
'hostname': '::1',
'user': 'gareth'}]
ret = wtmp.beacon(config)
log.debug('{}'.format(ret))
self.assertEqual(ret, _expected)
@skipIf(not _TIME_SUPPORTED, 'dateutil.parser is missing.')
def test_match_time(self):
with patch('salt.utils.files.fopen',
mock_open(read_data=raw)):
mock_now = datetime.datetime(2017, 9, 22, 16, 0, 0, 0)
with patch('datetime.datetime', MagicMock()), \
patch('datetime.datetime.now',
MagicMock(return_value=mock_now)):
with patch('struct.unpack',
MagicMock(return_value=pack)):
config = [{'users': {'gareth': {'time': {'end': '09-22-2017 5pm',
'start': '09-22-2017 3pm'}}}}
]
ret = wtmp.validate(config)
self.assertEqual(ret, (True, 'Valid beacon configuration'))
_expected = [{'PID': 6216,
'action': 'login',
'line': 'pts/14',
'session': 0,
'time': 0,
'exit_status': 0,
'inittab': 's/14',
'type': 7,
'addr': 1506101523,
'hostname': '::1',
'user': 'gareth'}]
ret = wtmp.beacon(config)
self.assertEqual(ret, _expected)
def test_match_group(self):
for groupadd in ('salt.modules.aix_group',
'salt.modules.mac_group',
'salt.modules.pw_group',
'salt.modules.solaris_group',
'salt.modules.win_groupadd'):
mock_group_info = {'passwd': 'x',
'gid': 100,
'name': 'users',
'members': ['gareth']}
with patch('salt.utils.files.fopen',
mock_open(read_data=raw)):
with patch('time.time',
MagicMock(return_value=1506121200)):
with patch('struct.unpack',
MagicMock(return_value=pack)):
with patch('{0}.info'.format(groupadd),
new=MagicMock(return_value=mock_group_info)):
config = [{'group': {'users': {'time': {'end': '09-22-2017 5pm',
'start': '09-22-2017 3pm'}}}}
]
ret = wtmp.validate(config)
self.assertEqual(ret,
(True, 'Valid beacon configuration'))
_expected = [{'PID': 6216,
'action': 'login',
'line': 'pts/14',
'session': 0,
'time': 0,
'exit_status': 0,
'inittab': 's/14',
'type': 7,
'addr': 1506101523,
'hostname': '::1',
'user': 'gareth'}]
ret = wtmp.beacon(config)
self.assertEqual(ret, _expected)