From a066bcb17e1b99a42a5834d1ace6feba7c9b60b7 Mon Sep 17 00:00:00 2001 From: Arthur de Jong Date: Fri, 16 Aug 2013 22:35:02 +0200 Subject: Implement tests for caching functionality --- tests/Makefile.am | 8 +- tests/test_pynslcd_cache.py | 459 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 464 insertions(+), 3 deletions(-) create mode 100755 tests/test_pynslcd_cache.py (limited to 'tests') diff --git a/tests/Makefile.am b/tests/Makefile.am index 184b79e..3c6bfce 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -19,8 +19,10 @@ # 02110-1301 USA TESTS = test_dict test_set test_tio test_expr test_getpeercred test_cfg \ - test_myldap.sh test_common test_nsscmds.sh test_pamcmds.sh \ - test_pycompile.sh + test_myldap.sh test_common test_nsscmds.sh test_pamcmds.sh +if HAVE_PYTHON +TESTS += test_pycompile.sh test_pynslcd_cache.py +endif AM_TESTS_ENVIRONMENT = PYTHON='@PYTHON@'; export PYTHON; @@ -30,7 +32,7 @@ check_PROGRAMS = test_dict test_set test_tio test_expr test_getpeercred \ EXTRA_DIST = nslcd-test.conf test_myldap.sh test_nsscmds.sh test_pamcmds.sh \ test_pycompile.sh in_testenv.sh test_pamcmds.expect \ - usernames.txt + usernames.txt test_pynslcd_cache.py CLEANFILES = $(EXTRA_PROGRAMS) diff --git a/tests/test_pynslcd_cache.py b/tests/test_pynslcd_cache.py new file mode 100755 index 0000000..5c15b01 --- /dev/null +++ b/tests/test_pynslcd_cache.py @@ -0,0 +1,459 @@ +#!/usr/bin/env python + +# test_pynslcd_cache.py - tests for the pynslcd caching functionality +# +# Copyright (C) 2013 Arthur de Jong +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library 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 +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA +# 02110-1301 USA + +import os +import os.path +import sys +import unittest + +# fix the Python path +sys.path.insert(1, os.path.abspath(os.path.join(sys.path[0], '..', 'pynslcd'))) +sys.path.insert(2, os.path.abspath(os.path.join('..', 'pynslcd'))) + + +# TODO: think about case-sesitivity of cache searches (have tests for that) + + +class TestAlias(unittest.TestCase): + + def setUp(self): + import alias + cache = alias.Cache() + cache.store('alias1', ['member1', 'member2']) + cache.store('alias2', ['member1', 'member3']) + cache.store('alias3', []) + self.cache = cache + + def test_by_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='alias1')), [ + ['alias1', ['member1', 'member2']], + ]) + + def test_by_member(self): + self.assertItemsEqual(self.cache.retrieve(dict(rfc822MailMember='member1')), [ + ['alias1', ['member1', 'member2']], + ['alias2', ['member1', 'member3']], + ]) + + def test_all(self): + self.assertItemsEqual(self.cache.retrieve({}), [ + ['alias1', ['member1', 'member2']], + ['alias2', ['member1', 'member3']], + ['alias3', []], + ]) + + +class TestEther(unittest.TestCase): + + def setUp(self): + import ether + cache = ether.Cache() + cache.store('name1', '0:18:8a:54:1a:11') + cache.store('name2', '0:18:8a:54:1a:22') + self.cache = cache + + def test_by_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='name1')), [ + ['name1', '0:18:8a:54:1a:11'], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='name2')), [ + ['name2', '0:18:8a:54:1a:22'], + ]) + + def test_by_ether(self): + # ideally we should also support alternate representations + self.assertItemsEqual(self.cache.retrieve(dict(macAddress='0:18:8a:54:1a:22')), [ + ['name2', '0:18:8a:54:1a:22'], + ]) + + def test_all(self): + self.assertItemsEqual(self.cache.retrieve({}), [ + ['name1', '0:18:8a:54:1a:11'], + ['name2', '0:18:8a:54:1a:22'], + ]) + + +class TestGroup(unittest.TestCase): + + def setUp(self): + import group + cache = group.Cache() + cache.store('group1', 'pass1', 10, ['user1', 'user2']) + cache.store('group2', 'pass2', 20, ['user1', 'user2', 'user3']) + cache.store('group3', 'pass3', 30, []) + cache.store('group4', 'pass4', 40, ['user2', ]) + self.cache = cache + + def test_by_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='group1')), [ + ['group1', 'pass1', 10, ['user1', 'user2']], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='group3')), [ + ['group3', 'pass3', 30, []], + ]) + + def test_by_gid(self): + self.assertItemsEqual(self.cache.retrieve(dict(gidNumber=10)), [ + ['group1', 'pass1', 10, ['user1', 'user2']], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(gidNumber=40)), [ + ['group4', 'pass4', 40, ['user2']], + ]) + + def test_all(self): + self.assertItemsEqual(self.cache.retrieve({}), [ + ['group1', 'pass1', 10, ['user1', 'user2']], + ['group2', 'pass2', 20, ['user1', 'user2', 'user3']], + ['group3', 'pass3', 30, []], + ['group4', 'pass4', 40, ['user2']], + ]) + + def test_bymember(self): + self.assertItemsEqual(self.cache.retrieve(dict(memberUid='user1')), [ + ['group1', 'pass1', 10, ['user1', 'user2']], + ['group2', 'pass2', 20, ['user1', 'user2', 'user3']], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(memberUid='user2')), [ + ['group1', 'pass1', 10, ['user1', 'user2']], + ['group2', 'pass2', 20, ['user1', 'user2', 'user3']], + ['group4', 'pass4', 40, ['user2']], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(memberUid='user3')), [ + ['group2', 'pass2', 20, ['user1', 'user2', 'user3']], + ]) + + +class TestHost(unittest.TestCase): + + def setUp(self): + import host + cache = host.Cache() + cache.store('hostname1', [], ['127.0.0.1', ]) + cache.store('hostname2', ['alias1', 'alias2'], ['127.0.0.2', '127.0.0.3']) + self.cache = cache + + def test_by_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='hostname1')), [ + ['hostname1', [], ['127.0.0.1']], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='hostname2')), [ + ['hostname2', ['alias1', 'alias2'], ['127.0.0.2', '127.0.0.3']], + ]) + + def test_by_alias(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='alias1')), [ + ['hostname2', ['alias1', 'alias2'], ['127.0.0.2', '127.0.0.3']], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='alias2')), [ + ['hostname2', ['alias1', 'alias2'], ['127.0.0.2', '127.0.0.3']], + ]) + + def test_by_address(self): + self.assertItemsEqual(self.cache.retrieve(dict(ipHostNumber='127.0.0.3')), [ + ['hostname2', ['alias1', 'alias2'], ['127.0.0.2', '127.0.0.3']], + ]) + + +class TestNetgroup(unittest.TestCase): + + def setUp(self): + import netgroup + cache = netgroup.Cache() + cache.store('netgroup1', ['(host1, user1,)', '(host1, user2,)', '(host2, user1,)'], ['netgroup2', ]) + cache.store('netgroup2', ['(host3, user1,)', '(host3, user3,)'], []) + self.cache = cache + + def test_by_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='netgroup1')), [ + ['netgroup1', ['(host1, user1,)', '(host1, user2,)', '(host2, user1,)'], ['netgroup2', ]], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='netgroup2')), [ + ['netgroup2', ['(host3, user1,)', '(host3, user3,)'], []], + ]) + + +class TestNetwork(unittest.TestCase): + + def setUp(self): + import network + cache = network.Cache() + cache.store('networkname1', [], ['127.0.0.1', ]) + cache.store('networkname2', ['alias1', 'alias2'], ['127.0.0.2', '127.0.0.3']) + self.cache = cache + + def test_by_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='networkname1')), [ + ['networkname1', [], ['127.0.0.1']], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='networkname2')), [ + ['networkname2', ['alias1', 'alias2'], ['127.0.0.2', '127.0.0.3']], + ]) + + def test_by_alias(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='alias1')), [ + ['networkname2', ['alias1', 'alias2'], ['127.0.0.2', '127.0.0.3']], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='alias2')), [ + ['networkname2', ['alias1', 'alias2'], ['127.0.0.2', '127.0.0.3']], + ]) + + def test_by_address(self): + self.assertItemsEqual(self.cache.retrieve(dict(ipNetworkNumber='127.0.0.3')), [ + ['networkname2', ['alias1', 'alias2'], ['127.0.0.2', '127.0.0.3']], + ]) + + +class TestPasswd(unittest.TestCase): + + def setUp(self): + import passwd + cache = passwd.Cache() + cache.store('name', 'passwd', 100, 200, 'gecos', '/home/user', '/bin/bash') + cache.store('name2', 'passwd2', 101, 202, 'gecos2', '/home/user2', '/bin/bash') + self.cache = cache + + def test_by_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(uid='name')), [ + [u'name', u'passwd', 100, 200, u'gecos', u'/home/user', u'/bin/bash'], + ]) + + def test_by_unknown_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(uid='notfound')), []) + + def test_by_number(self): + self.assertItemsEqual(self.cache.retrieve(dict(uidNumber=100)), [ + [u'name', u'passwd', 100, 200, u'gecos', u'/home/user', u'/bin/bash'], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(uidNumber=101)), [ + ['name2', 'passwd2', 101, 202, 'gecos2', '/home/user2', '/bin/bash'], + ]) + + def test_all(self): + self.assertItemsEqual(self.cache.retrieve({}), [ + [u'name', u'passwd', 100, 200, u'gecos', u'/home/user', u'/bin/bash'], + [u'name2', u'passwd2', 101, 202, u'gecos2', u'/home/user2', u'/bin/bash'], + ]) + + +class TestProtocol(unittest.TestCase): + + def setUp(self): + import protocol + cache = protocol.Cache() + cache.store('protocol1', ['alias1', 'alias2'], 100) + cache.store('protocol2', ['alias3', ], 200) + cache.store('protocol3', [], 300) + self.cache = cache + + def test_by_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='protocol1')), [ + ['protocol1', ['alias1', 'alias2'], 100], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='protocol2')), [ + ['protocol2', ['alias3', ], 200], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='protocol3')), [ + ['protocol3', [], 300], + ]) + + def test_by_unknown_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='notfound')), []) + + def test_by_number(self): + self.assertItemsEqual(self.cache.retrieve(dict(ipProtocolNumber=100)), [ + ['protocol1', ['alias1', 'alias2'], 100], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(ipProtocolNumber=200)), [ + ['protocol2', ['alias3', ], 200], + ]) + + def test_by_alias(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='alias1')), [ + ['protocol1', ['alias1', 'alias2'], 100], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='alias3')), [ + ['protocol2', ['alias3', ], 200], + ]) + + def test_all(self): + self.assertItemsEqual(self.cache.retrieve({}), [ + ['protocol1', ['alias1', 'alias2'], 100], + ['protocol2', ['alias3'], 200], + ['protocol3', [], 300], + ]) + + +class TestRpc(unittest.TestCase): + + def setUp(self): + import rpc + cache = rpc.Cache() + cache.store('rpc1', ['alias1', 'alias2'], 100) + cache.store('rpc2', ['alias3', ], 200) + cache.store('rpc3', [], 300) + self.cache = cache + + def test_by_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='rpc1')), [ + ['rpc1', ['alias1', 'alias2'], 100], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='rpc2')), [ + ['rpc2', ['alias3', ], 200], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='rpc3')), [ + ['rpc3', [], 300], + ]) + + def test_by_unknown_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='notfound')), []) + + def test_by_number(self): + self.assertItemsEqual(self.cache.retrieve(dict(oncRpcNumber=100)), [ + ['rpc1', ['alias1', 'alias2'], 100], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(oncRpcNumber=200)), [ + ['rpc2', ['alias3', ], 200], + ]) + + def test_by_alias(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='alias1')), [ + ['rpc1', ['alias1', 'alias2'], 100], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='alias3')), [ + ['rpc2', ['alias3', ], 200], + ]) + + def test_all(self): + self.assertItemsEqual(self.cache.retrieve({}), [ + ['rpc1', ['alias1', 'alias2'], 100], + ['rpc2', ['alias3'], 200], + ['rpc3', [], 300], + ]) + + +class TestService(unittest.TestCase): + + def setUp(self): + import service + cache = service.Cache() + cache.store('service1', ['alias1', 'alias2'], 100, 'tcp') + cache.store('service1', ['alias1', 'alias2'], 100, 'udp') + cache.store('service2', ['alias3', ], 200, 'udp') + cache.store('service3', [], 300, 'udp') + self.cache = cache + + def test_by_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='service1')), [ + ['service1', ['alias1', 'alias2'], 100, 'tcp'], + ['service1', ['alias1', 'alias2'], 100, 'udp'], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='service2')), [ + ['service2', ['alias3', ], 200, 'udp'], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='service3')), [ + ['service3', [], 300, 'udp'], + ]) + + def test_by_name_and_protocol(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='service1', ipServiceProtocol='udp')), [ + ['service1', ['alias1', 'alias2'], 100, 'udp'], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='service1', ipServiceProtocol='tcp')), [ + ['service1', ['alias1', 'alias2'], 100, 'tcp'], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='service2', ipServiceProtocol='udp')), [ + ['service2', ['alias3', ], 200, 'udp'], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='service2', ipServiceProtocol='tcp')), [ + ]) + + def test_by_unknown_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='notfound')), []) + + def test_by_number(self): + self.assertItemsEqual(self.cache.retrieve(dict(ipServicePort=100)), [ + ['service1', ['alias1', 'alias2'], 100, 'tcp'], + ['service1', ['alias1', 'alias2'], 100, 'udp'], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(ipServicePort=200)), [ + ['service2', ['alias3', ], 200, 'udp'], + ]) + + def test_by_number_and_protocol(self): + self.assertItemsEqual(self.cache.retrieve(dict(ipServicePort=100, ipServiceProtocol='udp')), [ + ['service1', ['alias1', 'alias2'], 100, 'udp'], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(ipServicePort=100, ipServiceProtocol='tcp')), [ + ['service1', ['alias1', 'alias2'], 100, 'tcp'], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(ipServicePort=200, ipServiceProtocol='udp')), [ + ['service2', ['alias3', ], 200, 'udp'], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(ipServicePort=200, ipServiceProtocol='tcp')), [ + ]) + + def test_by_alias(self): + self.assertItemsEqual(self.cache.retrieve(dict(cn='alias1')), [ + ['service1', ['alias1', 'alias2'], 100, 'udp'], + ['service1', ['alias1', 'alias2'], 100, 'tcp'], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(cn='alias3')), [ + ['service2', ['alias3', ], 200, 'udp'], + ]) + + def test_all(self): + self.assertItemsEqual(self.cache.retrieve({}), [ + ['service1', ['alias1', 'alias2'], 100, 'tcp'], + ['service1', ['alias1', 'alias2'], 100, 'udp'], + ['service2', ['alias3', ], 200, 'udp'], + ['service3', [], 300, 'udp'], + ]) + + +class Testshadow(unittest.TestCase): + + def setUp(self): + import shadow + cache = shadow.Cache() + cache.store('name', 'passwd', 15639, 0, 7, -1, -1, -1, 0) + cache.store('name2', 'passwd2', 15639, 0, 7, -1, -1, -1, 0) + self.cache = cache + + def test_by_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(uid='name')), [ + [u'name', u'passwd', 15639, 0, 7, -1, -1, -1, 0], + ]) + self.assertItemsEqual(self.cache.retrieve(dict(uid='name2')), [ + [u'name2', u'passwd2', 15639, 0, 7, -1, -1, -1, 0], + ]) + + def test_by_unknown_name(self): + self.assertItemsEqual(self.cache.retrieve(dict(uid='notfound')), []) + + def test_all(self): + self.assertItemsEqual(self.cache.retrieve({}), [ + [u'name', u'passwd', 15639, 0, 7, -1, -1, -1, 0], + [u'name2', u'passwd2', 15639, 0, 7, -1, -1, -1, 0], + ]) + + +if __name__ == '__main__': + unittest.main() -- cgit v1.2.3