valitydev/salt
14
0
Fork 0
mirror of https://github.com/valitydev/salt.git synced 2024-11-07 08:58:59 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity
c7cb00923e
salt/tests/unit/modules/test_memcached.py
rallytime 3273bbdab7
Merge branch '2017.7' into '2018.3' 
Conflicts:
  - doc/ref/configuration/master.rst
  - doc/ref/modules/all/index.rst
  - doc/topics/grains/index.rst
  - doc/topics/releases/2016.3.4.rst
  - doc/topics/spm/spm_formula.rst
  - doc/topics/tutorials/cron.rst
  - doc/topics/tutorials/index.rst
  - doc/topics/tutorials/stormpath.rst
  - salt/engines/slack.py
  - salt/log/handlers/fluent_mod.py
  - salt/modules/cyg.py
  - salt/modules/junos.py
  - salt/modules/namecheap_dns.py
  - salt/modules/namecheap_domains.py
  - salt/modules/namecheap_ns.py
  - salt/modules/namecheap_ssl.py
  - salt/modules/namecheap_users.py
  - salt/modules/reg.py
  - salt/modules/tomcat.py
  - salt/modules/vault.py
  - salt/modules/win_file.py
  - salt/modules/zpool.py
  - salt/output/highstate.py
  - salt/renderers/pass.py
  - salt/runners/cache.py
  - salt/states/boto_apigateway.py
  - salt/states/boto_iam.py
  - salt/states/boto_route53.py
  - salt/states/msteams.py
  - salt/states/reg.py
  - salt/states/win_iis.py
  - tests/integration/modules/test_cmdmod.py
  - tests/integration/states/test_user.py
  - tests/support/helpers.py
  - tests/unit/cloud/clouds/test_openstack.py
  - tests/unit/fileserver/test_gitfs.py
  - tests/unit/modules/test_junos.py
  - tests/unit/pillar/test_git.py
  - tests/unit/states/test_win_path.py
  - tests/unit/test_pillar.py
  - tests/unit/utils/test_format_call.py
  - tests/unit/utils/test_utils.py
  - tests/unit/utils/test_warnings.py
2018-06-01 14:54:12 -04:00

468 lines
13 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
'''
:codeauthor: Jayesh Kariya <jayeshk@saltstack.com>
'''
# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals
# Import Salt Testing Libs
from tests.support.unit import TestCase, skipIf
from tests.support.mock import (
MagicMock,
patch,
NO_MOCK,
NO_MOCK_REASON
)
# Import Salt Libs
import salt.modules.memcached as memcached
from salt.exceptions import CommandExecutionError, SaltInvocationError
from salt.ext.six import integer_types
@skipIf(NO_MOCK, NO_MOCK_REASON)
class MemcachedTestCase(TestCase):
'''
Test cases for salt.modules.memcached
'''
# 'status' function tests: 2
def test_status(self):
'''
Test if it gets memcached status
'''
class MockMemcache(object):
"""
Mock of memcache
"""
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return [('127.0.0.1:11211 (1)', {})]
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertDictEqual(memcached.status(),
{'127.0.0.1:11211 (1)': {}})
def test_status_false(self):
'''
Test if it gets memcached status
'''
class MockMemcache(object):
"""
Mock of memcache
"""
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return []
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertFalse(memcached.status())
# 'get' function tests: 1
def test_get(self):
'''
Test if it retrieve value for a key
'''
class MockMemcache(object):
"""
Mock of memcache
"""
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return [('127.0.0.1:11211 (1)', {})]
@staticmethod
def get(key):
"""
Mock of get method
"""
return key
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertEqual(memcached.get('salt'), 'salt')
# 'set_' function tests: 1
def test_set(self):
'''
Test if it set a key on the memcached server
'''
class MockMemcache(object):
"""
Mock of memcache
"""
def __init__(self):
self.key = None
self.value = None
self.time = None
self.min_compress_len = None
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return [('127.0.0.1:11211 (1)', {})]
def set(self, key, value, time, min_compress_len):
"""
Mock of set method
"""
self.key = key
self.value = value
self.time = time
self.min_compress_len = min_compress_len
return True
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertTrue(memcached.set_('salt', '1111'))
self.assertRaises(SaltInvocationError, memcached.set_,
'salt', '1111', time='0.1')
self.assertRaises(SaltInvocationError, memcached.set_,
'salt', '1111', min_compress_len='0.1')
# 'delete' function tests: 1
def test_delete(self):
'''
Test if it delete a key from memcache server
'''
class MockMemcache(object):
"""
Mock of memcache
"""
def __init__(self):
self.key = None
self.time = None
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return [('127.0.0.1:11211 (1)', {})]
def delete(self, key, time):
"""
Mock of delete method
"""
self.key = key
self.time = time
return True
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertTrue(memcached.delete('salt'))
self.assertRaises(SaltInvocationError, memcached.delete,
'salt', '1111', time='0.1')
# 'add' function tests: 1
def test_add(self):
'''
Test if it add a key from memcache server
'''
class MockMemcache(object):
"""
Mock of memcache
"""
def __init__(self):
self.key = None
self.value = None
self.time = None
self.min_compress_len = None
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return [('127.0.0.1:11211 (1)', {})]
def add(self, key, value, time, min_compress_len):
"""
Mock of add method
"""
self.key = key
self.value = value
self.time = time
self.min_compress_len = min_compress_len
return True
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertTrue(memcached.add('salt', '1111'))
self.assertRaises(SaltInvocationError, memcached.add,
'salt', '1111', time='0.1')
self.assertRaises(SaltInvocationError, memcached.add,
'salt', '1111', min_compress_len='0.1')
# 'replace' function tests: 1
def test_replace(self):
'''
Test if it replace a key from memcache server
'''
class MockMemcache(object):
"""
Mock of memcache
"""
def __init__(self):
self.key = None
self.value = None
self.time = None
self.min_compress_len = None
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return [('127.0.0.1:11211 (1)', {})]
def replace(self, key, value, time, min_compress_len):
"""
Mock of replace method
"""
self.key = key
self.value = value
self.time = time
self.min_compress_len = min_compress_len
return True
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertTrue(memcached.replace('salt', '1111'))
self.assertRaises(SaltInvocationError, memcached.replace,
'salt', '1111', time='0.1')
self.assertRaises(SaltInvocationError, memcached.replace,
'salt', '1111', min_compress_len='0.1')
# 'increment' function tests: 3
def test_increment(self):
'''
Test if it increment the value of a key
'''
class MockMemcache(object):
"""
Mock of memcache
"""
def __init__(self):
self.key = None
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return [('127.0.0.1:11211 (1)', {})]
def get(self, key):
"""
Mock of get method
"""
self.key = key
return 1
def incr(self, key, delta):
"""
Mock of incr method
"""
self.key = key
if not isinstance(delta, integer_types):
raise SaltInvocationError('Delta value must be an integer')
return key
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertEqual(memcached.increment('salt'), 'salt')
self.assertRaises(SaltInvocationError, memcached.increment,
'salt', delta='sa')
def test_increment_exist(self):
'''
Test if it increment the value of a key
'''
class MockMemcache(object):
"""
Mock of memcache
"""
def __init__(self):
self.key = None
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return [('127.0.0.1:11211 (1)', {})]
def get(self, key):
"""
Mock of get method
"""
self.key = key
return key
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertRaises(CommandExecutionError, memcached.increment,
'salt')
def test_increment_none(self):
'''
Test if it increment the value of a key
'''
class MockMemcache(object):
"""
Mock of memcache
"""
def __init__(self):
self.key = None
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return [('127.0.0.1:11211 (1)', {})]
def get(self, key):
"""
Mock of get method
"""
self.key = key
return None
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertRaises(CommandExecutionError, memcached.increment,
'salt')
# 'decrement' function tests: 3
def test_decrement(self):
'''
Test if it decrement the value of a key
'''
class MockMemcache(object):
"""
Mock of memcache
"""
def __init__(self):
self.key = None
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return [('127.0.0.1:11211 (1)', {})]
def get(self, key):
"""
Mock of get method
"""
self.key = key
return 1
def decr(self, key, delta):
"""
Mock of decr method
"""
self.key = key
if not isinstance(delta, integer_types):
raise SaltInvocationError('Delta value must be an integer')
return key
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertEqual(memcached.decrement('salt'), 'salt')
self.assertRaises(SaltInvocationError, memcached.decrement,
'salt', delta='sa')
def test_decrement_exist(self):
'''
Test if it decrement the value of a key
'''
class MockMemcache(object):
"""
Mock of memcache
"""
def __init__(self):
self.key = None
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return [('127.0.0.1:11211 (1)', {})]
def get(self, key):
"""
Mock of get method
"""
self.key = key
return key
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertRaises(CommandExecutionError, memcached.decrement,
'salt')
def test_decrement_none(self):
'''
Test if it decrement the value of a key
'''
class MockMemcache(object):
"""
Mock of memcache
"""
def __init__(self):
self.key = None
@staticmethod
def get_stats():
"""
Mock of stats method
"""
return [('127.0.0.1:11211 (1)', {})]
def get(self, key):
"""
Mock of get method
"""
self.key = key
return None
with patch.object(memcached, '_connect',
MagicMock(return_value=MockMemcache())):
self.assertRaises(CommandExecutionError, memcached.decrement,
'salt')
Reference in New Issue View Git Blame Copy Permalink