Coverage for tests/test_cmdline/test_cli : 91%

import shlex 

import sys 

import contextlib 

import StringIO 

import nose 

from tests import util 

from ipalib import api, errors 

from ipapython.version import API_VERSION 

class TestCLIParsing(object): 

    """Tests that commandlines are correctly parsed to Command keyword args 

    """ 

    def check_command(self, commandline, expected_command_name, **kw_expected): 

        argv = shlex.split(commandline) 

        executioner = api.Backend.cli 

        cmd = executioner.get_command(argv) 

        kw_got = executioner.argv_to_keyword_arguments(cmd, argv[1:]) 

        util.assert_deepequal(expected_command_name, cmd.name, 'Command name') 

        util.assert_deepequal(kw_expected, kw_got) 

    def run_command(self, command_name, **kw): 

        """Run a command on the server""" 

        if not api.Backend.xmlclient.isconnected(): 

            api.Backend.xmlclient.connect(fallback=False) 

        try: 

            api.Command[command_name](**kw) 

        except errors.NetworkError: 

            raise nose.SkipTest('%r: Server not available: %r' % 

                                (self.__module__, api.env.xmlrpc_uri)) 

    @contextlib.contextmanager 

    def fake_stdin(self, string_in): 

        """Context manager that temporarily replaces stdin to read a string""" 

        old_stdin = sys.stdin 

        sys.stdin = StringIO.StringIO(string_in) 

        yield 

        sys.stdin = old_stdin 

    def test_ping(self): 

        self.check_command('ping', 'ping', 

            version=API_VERSION) 

    def test_user_show(self): 

        self.check_command('user-show admin', 'user_show', 

            uid=u'admin', 

            rights=False, 

            raw=False, 

            all=False, 

            version=API_VERSION) 

    def test_user_show_underscore(self): 

        self.check_command('user_show admin', 'user_show', 

            uid=u'admin', 

            rights=False, 

            raw=False, 

            all=False, 

            version=API_VERSION) 

    def test_group_add(self): 

        self.check_command('group-add tgroup1 --desc="Test group"', 

            'group_add', 

            cn=u'tgroup1', 

            description=u'Test group', 

            nonposix=False, 

            external=False, 

            raw=False, 

            all=False, 

            version=API_VERSION) 

    def test_sudocmdgroup_add_member(self): 

        # Test CSV splitting is done correctly 

        self.check_command( 

            # The following is as it would appear on the command line: 

            r'sudocmdgroup-add-member tcmdgroup1 --sudocmds=abc,\"de,f\",\\,g', 

            'sudocmdgroup_add_member', 

            cn=u'tcmdgroup1', 

            sudocmd=[u'abc', u'de,f', u'\\', u'g'], 

            raw=False, 

            all=False, 

            version=API_VERSION) 

    def test_group_add_nonposix(self): 

        self.check_command('group-add tgroup1 --desc="Test group" --nonposix', 

            'group_add', 

            cn=u'tgroup1', 

            description=u'Test group', 

            nonposix=True, 

            external=False, 

            raw=False, 

            all=False, 

            version=API_VERSION) 

    def test_group_add_gid(self): 

        self.check_command('group-add tgroup1 --desc="Test group" --gid=1234', 

            'group_add', 

            cn=u'tgroup1', 

            description=u'Test group', 

            gidnumber=u'1234', 

            nonposix=False, 

            external=False, 

            raw=False, 

            all=False, 

            version=API_VERSION) 

    def test_group_add_interactive(self): 

        with self.fake_stdin('Test group\n'): 

            self.check_command('group-add tgroup1', 'group_add', 

                cn=u'tgroup1', 

                description=u'Test group', 

                nonposix=False, 

                external=False, 

                raw=False, 

                all=False, 

                version=API_VERSION) 

    def test_dnsrecord_add(self): 

        self.check_command('dnsrecord-add test-example.com ns --a-rec=1.2.3.4', 

            'dnsrecord_add', 

            dnszoneidnsname=u'test-example.com', 

            idnsname=u'ns', 

            arecord=[u'1.2.3.4'], 

            structured=False, 

            force=False, 

            raw=False, 

            all=False, 

            version=API_VERSION) 

    def test_dnsrecord_del_all(self): 

        try: 

            self.run_command('dnszone_add', idnsname=u'test-example.com', 

                idnssoamname=u'ns.test-example.com', force=True) 

        except errors.NotFound: 

            raise nose.SkipTest('DNS is not configured') 

        try: 

            self.run_command('dnsrecord_add', 

                dnszoneidnsname=u'test-example.com', 

                idnsname=u'ns', arecord=u'1.2.3.4') 

            with self.fake_stdin('yes\n'): 

                self.check_command('dnsrecord_del test-example.com ns', 

                    'dnsrecord_del', 

                    dnszoneidnsname=u'test-example.com', 

                    idnsname=u'ns', 

                    del_all=True, 

                    structured=False, 

                    raw=False, 

                    all=False, 

                    version=API_VERSION) 

            with self.fake_stdin('YeS\n'): 

                self.check_command('dnsrecord_del test-example.com ns', 

                    'dnsrecord_del', 

                    dnszoneidnsname=u'test-example.com', 

                    idnsname=u'ns', 

                    del_all=True, 

                    structured=False, 

                    raw=False, 

                    all=False, 

                    version=API_VERSION) 

        finally: 

            self.run_command('dnszone_del', idnsname=u'test-example.com') 

    def test_dnsrecord_del_one_by_one(self): 

        try: 

            self.run_command('dnszone_add', idnsname=u'test-example.com', 

                idnssoamname=u'ns.test-example.com', force=True) 

        except errors.NotFound: 

            raise nose.SkipTest('DNS is not configured') 

        try: 

            records = (u'1 1 E3B72BA346B90570EED94BE9334E34AA795CED23', 

                       u'2 1 FD2693C1EFFC11A8D2BE57229212A04B45663791') 

            for record in records: 

                self.run_command('dnsrecord_add', 

                    dnszoneidnsname=u'test-example.com', idnsname=u'ns', 

                    sshfprecord=record) 

            with self.fake_stdin('no\nyes\nyes\n'): 

                self.check_command('dnsrecord_del test-example.com ns', 

                    'dnsrecord_del', 

                    dnszoneidnsname=u'test-example.com', 

                    idnsname=u'ns', 

                    del_all=False, 

                    sshfprecord=records, 

                    structured=False, 

                    raw=False, 

                    all=False, 

                    version=API_VERSION) 

        finally: 

            self.run_command('dnszone_del', idnsname=u'test-example.com') 

    def test_dnsrecord_add_ask_for_missing_fields(self): 

        sshfp_parts = (1, 1, u'E3B72BA346B90570EED94BE9334E34AA795CED23') 

        with self.fake_stdin('SSHFP\n%d\n%d\n%s' % sshfp_parts): 

            self.check_command('dnsrecord-add test-example.com sshfp', 

                'dnsrecord_add', 

                dnszoneidnsname=u'test-example.com', 

                idnsname=u'sshfp', 

                sshfp_part_fp_type=sshfp_parts[0], 

                sshfp_part_algorithm=sshfp_parts[1], 

                sshfp_part_fingerprint=sshfp_parts[2], 

                structured=False, 

                raw=False, 

                all=False, 

                force=False, 

                version=API_VERSION) 

        # NOTE: when a DNS record part is passed via command line, it is not 

        # converted to its base type when transfered via wire 

        with self.fake_stdin('%d\n%s' % (sshfp_parts[1], sshfp_parts[2])): 

            self.check_command('dnsrecord-add test-example.com sshfp ' \ 

                    '--sshfp-algorithm=%d' % sshfp_parts[0], 

                'dnsrecord_add', 

                dnszoneidnsname=u'test-example.com', 

                idnsname=u'sshfp', 

                sshfp_part_fp_type=sshfp_parts[0], 

                sshfp_part_algorithm=unicode(sshfp_parts[1]),   # passed via cmdline 

                sshfp_part_fingerprint=sshfp_parts[2], 

                structured=False, 

                raw=False, 

                all=False, 

                force=False, 

                version=API_VERSION) 

        with self.fake_stdin(sshfp_parts[2]): 

            self.check_command('dnsrecord-add test-example.com sshfp ' \ 

                    '--sshfp-algorithm=%d --sshfp-fp-type=%d' % (sshfp_parts[0], sshfp_parts[1]), 

                'dnsrecord_add', 

                dnszoneidnsname=u'test-example.com', 

                idnsname=u'sshfp', 

                sshfp_part_fp_type=unicode(sshfp_parts[0]),     # passed via cmdline 

                sshfp_part_algorithm=unicode(sshfp_parts[1]),   # passed via cmdline 

                sshfp_part_fingerprint=sshfp_parts[2], 

                structured=False, 

                raw=False, 

                all=False, 

                force=False, 

                version=API_VERSION)