Python click.testing 模块，CliRunner() 实例源码

def test_cli_with_login_username_only(self, mocked_method):
        mocked = mock.MagicMock()
        mocked.login.return_value = {"Status": "Login Succeeded"}
        mocked.push.return_value = [
            {"stream": "In process"},
            {"status": "Successfully pushed"}]
        mocked_method.return_value = mocked
        with FakeProjectDirectory() as tmpdir:
            add_sh_fake_config(tmpdir)
            runner = CliRunner()
            result = runner.invoke(
                cli, ["dev", "--version", "test", "--apikey", "apikey"])
            assert result.exit_code == 0
            mocked.login.assert_called_with(
                email=None, password=' ',
                reauth=False, registry='registry', username='apikey')
            mocked.push.assert_called_with(
                'registry/user/project:test', decode=True,
                insecure_registry=False, stream=True)

def testSQLiteHelp(self):
    """testing the main of the sqlite"""
    runner = CliRunner()
    result = runner.invoke(main.entry_point, ['sqlite', '--help'])
    expected_output = ('Usage: entry_point sqlite [OPTIONS]\n'
                       '\n'
                       'Options:\n'
                       '  --path TEXT       The path to plaso\n'
                       '  --name TEXT       The plugin name\n'
                       '  --testfile TEXT   The testfile path\n'
                       '  --sql / --no-sql  The output example flag for the '
                       'SQL Query for the plugin.\n'
                       '  --help            Show this message and exit.\n')

    self.assertEqual(expected_output, str(result.output))
    self.assertEqual(0, result.exit_code)

def test_password_empty_file():
    """Test the encrypt module's CLI function with an empty YAML file."""
    runner = CliRunner()
    with runner.isolated_filesystem():
        initial_data = {'language': 'python'}
        with open('file.yml', 'w') as file:
            ordered_dump(initial_data, file)

        result = runner.invoke(cli, ['mandeep', 'Travis-Encrypt', 'file.yml'],
                               'SUPER_SECURE_PASSWORD')
        assert not result.exception

        with open('file.yml') as file:
            config = ordered_load(file)

            assert 'password' in config
            assert 'secure' in config['password']
            assert config['language'] == 'python'
            assert base64.b64decode(config['password']['secure'])

def test_password_nonempty_file():
    """Test the encrypt module's CLI function with a nonempty YAML file.

    The YAML file includes information that needs to be overwritten."""
    runner = CliRunner()
    with runner.isolated_filesystem():

        initial_data = OrderedDict([('language', 'python'), ('dist', 'trusty'),
                                    ('password', {'secure': 'SUPER_INSECURE_PASSWORD'})])

        with open('file.yml', 'w') as file:
            ordered_dump(initial_data, file)

        result = runner.invoke(cli, ['mandeep', 'Travis-Encrypt', 'file.yml'],
                               'SUPER_SECURE_PASSWORD')
        assert not result.exception

        with open('file.yml') as file:
            config = ordered_load(file)

        assert config['language'] == 'python'
        assert config['dist'] == 'trusty'
        assert base64.b64decode(config['password']['secure'])
        assert ['language', 'dist', 'password'] == [key for key in config.keys()]

def test_deploy_empty_file():
    """Test the encrypt module's CLI function with the --deploy flag and an empty YAML file."""
    runner = CliRunner()
    with runner.isolated_filesystem():
        initial_data = {'language': 'python'}
        with open('file.yml', 'w') as file:
            ordered_dump(initial_data, file)

        result = runner.invoke(cli, ['--deploy', 'mandeep', 'Travis-Encrypt', 'file.yml'],
                               'SUPER_SECURE_PASSWORD')
        assert not result.exception

        with open('file.yml') as file:
            config = ordered_load(file)

        assert config['language'] == 'python'
        assert base64.b64decode(config['deploy']['password']['secure'])

def test_deploy_nonempty_file():
    """Test the encrypt module's CLI function with the --deploy flag and a nonempty YAML file.

    The YAML file includes information that needs to be overwritten."""
    runner = CliRunner()
    with runner.isolated_filesystem():

        initial_data = OrderedDict([('language', 'python'), ('dist', 'trusty'),
                                    ('deploy', {'password': {'secure': 'SUPER_INSECURE_PASSWORD'}})])

        with open('file.yml', 'w') as file:
            ordered_dump(initial_data, file)

        result = runner.invoke(cli, ['--deploy', 'mandeep', 'Travis-Encrypt', 'file.yml'],
                               'SUPER_SECURE_PASSWORD')

        assert not result.exception

        with open('file.yml') as file:
            config = ordered_load(file)

        assert config['language'] == 'python'
        assert config['dist'] == 'trusty'
        assert base64.b64decode(config['deploy']['password']['secure'])
        assert ['language', 'dist', 'deploy'] == [key for key in config.keys()]

def test_environment_variable_nonempty_file():
    """Test the encrypt module's CLI function with the --env flag and a nonempty YAML file.

    The YAML file includes information that needs to be overwritten."""
    runner = CliRunner()
    with runner.isolated_filesystem():

        initial_data = OrderedDict([('language', 'python'), ('dist', 'trusty'),
                                    ('env', {'global': {'secure': 'API_KEY="SUPER_INSECURE_KEY"'}})])

        with open('file.yml', 'w') as file:
            ordered_dump(initial_data, file)

        result = runner.invoke(cli, ['--env', 'mandeep', 'Travis-Encrypt', 'file.yml'],
                               'SUPER_SECURE_API_KEY')

        assert not result.exception

        with open('file.yml') as file:
            config = ordered_load(file)

        assert config['language'] == 'python'
        assert config['dist'] == 'trusty'
        assert base64.b64decode(config['env']['global']['secure'])
        assert ['language', 'dist', 'env'] == [key for key in config.keys()]

def test_cli_with_insecure_registry(self, mocked_method):
        mocked = mock.MagicMock()
        mocked.login.return_value = {"Status": "Login Succeeded"}
        mocked.push.return_value = [
            {"stream": "In process"},
            {"status": "Successfully pushed"}]
        mocked_method.return_value = mocked
        with FakeProjectDirectory() as tmpdir:
            add_sh_fake_config(tmpdir)
            runner = CliRunner()
            result = runner.invoke(
                cli, ["dev", "--version", "test", "--insecure"])
            assert result.exit_code == 0
            assert not mocked.login.called
            mocked.push.assert_called_with(
                'registry/user/project:test', decode=True,
                insecure_registry=True, stream=True)

def test_cli_insecure_registry(self, list_mocked, post_mocked, get_mocked):
        list_mocked.return_value = ['a1f', 'abc', 'spi-der']
        post_req = mock.Mock()
        post_req.headers = {'location': 'https://status-url'}
        post_mocked.return_value = post_req

        with FakeProjectDirectory() as tmpdir:
            add_scrapy_fake_config(tmpdir)
            add_sh_fake_config(tmpdir)
            add_fake_setup_py(tmpdir)

            runner = CliRunner()
            result = runner.invoke(
                cli, ["dev", "--version", "test", "--insecure"])
            assert result.exit_code == 0
            post_mocked.assert_called_with(
                'https://app.scrapinghub.com/api/releases/deploy.json',
                allow_redirects=False, auth=('abcdef', ''),
                data={'project': 12345,
                      'version': u'test',
                      'pull_insecure_registry': True,
                      'image_url': 'registry/user/project:test',
                      'spiders': 'a1f,abc,spi-der',
                      'scripts': 'scriptA.py,scriptB.py'}, timeout=300)
            get_mocked.assert_called_with('https://status-url', timeout=300)

def test_cli(self, test_mock, mocked_method):
        mocked = mock.MagicMock()
        mocked.build.return_value = [
            {"stream": "all is ok"},
            {"stream": "Successfully built 12345"}]
        mocked_method.return_value = mocked
        with FakeProjectDirectory() as tmpdir:
            add_scrapy_fake_config(tmpdir)
            add_sh_fake_config(tmpdir)
            add_fake_dockerfile(tmpdir)
            setup_py_path = os.path.join(tmpdir, 'setup.py')
            assert not os.path.isfile(setup_py_path)
            runner = CliRunner()
            result = runner.invoke(cli, ["dev", "-d"])
            assert result.exit_code == 0
            mocked.build.assert_called_with(
                decode=True, path=tmpdir, tag='registry/user/project:1.0')
            assert os.path.isfile(setup_py_path)
            test_mock.assert_called_with("dev", None)

def test_cli_custom_version(self, test_mock, mocked_method):
        mocked = mock.MagicMock()
        mocked.build.return_value = [
            {"stream": "all is ok"},
            {"stream": "Successfully built 12345"}]
        mocked_method.return_value = mocked
        with FakeProjectDirectory() as tmpdir:
            add_scrapy_fake_config(tmpdir)
            add_sh_fake_config(tmpdir)
            add_fake_dockerfile(tmpdir)
            runner = CliRunner()
            result = runner.invoke(cli, ["dev", "--version", "test"])
            assert result.exit_code == 0
            mocked.build.assert_called_with(
                decode=True, path=tmpdir, tag='registry/user/project:test')
            test_mock.assert_called_with("dev", "test")

def test_cli_container_error(self, get_mocked, get_client_mock):
        client_mock = mock.Mock()
        client_mock.create_container.return_value = {'Id': '1234'}
        client_mock.wait.return_value = 66
        get_client_mock.return_value = client_mock

        get_settings_mock = mock.Mock()
        get_settings_mock.json.return_value = {}
        get_mocked.return_value = get_settings_mock

        with FakeProjectDirectory() as tmpdir:
            add_sh_fake_config(tmpdir)
            runner = CliRunner()
            result = runner.invoke(cli, ["dev", "-d", "--version", "test"])
            assert result.exit_code == 1
            assert 'list cmd exited with code 66' in result.output

def test_cli(self, get_mocked, get_client_mock):
        client_mock = mock.Mock()
        client_mock.create_container.return_value = {'Id': '1234'}
        client_mock.wait.return_value = 0
        client_mock.logs.return_value = b'abc\ndef\ndsd'
        get_client_mock.return_value = client_mock

        get_settings_mock = mock.Mock()
        get_settings_mock.json.return_value = {}
        get_mocked.return_value = get_settings_mock

        with FakeProjectDirectory() as tmpdir:
            add_sh_fake_config(tmpdir)
            runner = CliRunner()
            result = runner.invoke(cli, [
                "dev", "-d", "-s", "--version", "test"])
            assert result.exit_code == 0
            assert result.output.endswith('abc\ndef\ndsd\n')
        get_mocked.assert_called_with(
            'https://app.scrapinghub.com/api/settings/get.json',
            allow_redirects=False, auth=('abcdef', ''),
            params={'project': 12345}, timeout=300)

def test_sub_command_options_other(push):
    """Test additional option values for all sub commands.

    :param bool push: Run push sub command instead of build.
    """
    if push:
        args = ['push', 'docs', 'gh-pages', '.']
    else:
        args = ['build', 'docs', join('docs', '_build', 'html')]

    # Defined.
    args += ['-p', 'tags', '-s', 'semver', '-s', 'time']
    if push:
        args += ['-e' 'one', '-e', 'two', '-e', 'three', '-e', 'four']
    result = CliRunner().invoke(cli, args)
    config = result.exception.args[0]
    assert config.priority == 'tags'
    assert config.sort == ('semver', 'time')
    if push:
        assert config.grm_exclude == ('one', 'two', 'three', 'four')

def setup_runner_resource(config_file, table_name):

    # setup

    runner = CliRunner()

    api = get_api(config_file=config_file)

    prefix = ['--config-file', config_file]

    try:
        api.manager.delete_table(table_name)
    except KeyError:
        pass

    api.manager.create_archive_table(table_name)

    # yield fixture

    yield runner, api, config_file, prefix

    # teardown

    api.manager.delete_table(table_name)

def test_insufficient_configuration(manager_with_spec, tempdir):
    '''
    Test writing an api with required user config to a config file and then
    running configure without sufficient user_config.
    '''

    # Create an insufficiently specified API
    api = DataAPI(username='My Name')

    # Attach a manager which requires username, contact
    api.attach_manager(manager_with_spec)

    # Export the API to a file
    config_file = os.path.join(tempdir, '.datafs.yml')
    to_config_file(api=api, config_file=config_file, profile='conftest')

    runner = CliRunner()

    prefix = ['--config-file', config_file, '--profile', 'conftest']

    # Test the configuration and make sure an exception is raised
    result = runner.invoke(cli, prefix + ['configure'])
    assert result.exception

def test_delete_profile_missing_profile_name(self):
        runner = CliRunner()
        result = runner.invoke(delete_profile)
        assert result.exit_code == 1
        assert result.output == '--profile_name required\nAborted!\n'

def test_set_profile_missing_profile_name(self):
        runner = CliRunner()
        result = runner.invoke(set_profile)
        assert result.exit_code == 1
        assert result.output == '--profile_name required\nAborted!\n'

def test_delete_application_cache_missing_application_name(self):
        runner = CliRunner()
        result = runner.invoke(delete_application_cache)
        assert result.exit_code == 1
        assert result.output == '--application_name required\nAborted!\n'

def test_delete_application_cache_missing_platform_name(self):
        runner = CliRunner()
        result = runner.invoke(delete_application_cache, ['--application_name', 'toto'])
        assert result.exit_code == 1
        assert result.output == '--platform_name required\nAborted!\n'

def test_delete_release_module_cache_missing_module_name(self):
        runner = CliRunner()
        result = runner.invoke(delete_release_module_cache)
        assert result.exit_code == 1
        assert result.output == '--module_name required\nAborted!\n'

def test_delete_release_template_package_cache_missing_template_name(self):
        runner = CliRunner()
        result = runner.invoke(delete_release_template_package_cache)
        assert result.exit_code == 1
        assert result.output == '--template_name required\nAborted!\n'

def test_delete_release_template_package_cache_missing_template_version(self):
        runner = CliRunner()
        result = runner.invoke(delete_release_template_package_cache, ['--template_name', 'toto'])
        assert result.exit_code == 1
        assert result.output == '--template_version required\nAborted!\n'

def test_delete_workingcopy_module_cache_missing_module_name(self):
        runner = CliRunner()
        result = runner.invoke(delete_workingcopy_module_cache)
        assert result.exit_code == 1
        assert result.output == '--module_name required\nAborted!\n'

def test_delete_workingcopy_module_cache_missing_module_version(self):
        runner = CliRunner()
        result = runner.invoke(delete_workingcopy_module_cache, ['--module_name', 'toto'])
        assert result.exit_code == 1
        assert result.output == '--module_version required\nAborted!\n'

def test_delete_workingcopy_template_package_cache_missing_template_name(self):
        runner = CliRunner()
        result = runner.invoke(delete_workingcopy_template_package_cache)
        assert result.exit_code == 1
        assert result.output == '--template_name required\nAborted!\n'

def test_regenerate_application_cache_missing_template_version(self):
        runner = CliRunner()
        result = runner.invoke(regenerate_application_cache)
        assert result.exit_code == 1
        assert result.output == '--application_name required\nAborted!\n'

def test_regenerate_application_cache_missing_platform_name(self):
        runner = CliRunner()
        result = runner.invoke(regenerate_application_cache, ['--application_name', 'toto'])
        assert result.exit_code == 1
        assert result.output == '--platform_name required\nAborted!\n'

def test_regenerate_module_cache_missing_module_name(self):
        runner = CliRunner()
        result = runner.invoke(regenerate_module_cache)
        assert result.exit_code == 1
        assert result.output == '--module_name required\nAborted!\n'

def test_regenerate_module_cache_missing_module_version(self):
        runner = CliRunner()
        result = runner.invoke(regenerate_module_cache, ['--module_name', 'toto'])
        assert result.exit_code == 1
        assert result.output == '--module_version required\nAborted!\n'

def test_regenerate_template_package_cache_missing_template_name(self):
        runner = CliRunner()
        result = runner.invoke(regenerate_template_package_cache)
        assert result.exit_code == 1
        assert result.output == '--template_name required\nAborted!\n'

def test_get_files_missing_application_name(self):
        runner = CliRunner()
        result = runner.invoke(get_files)
        assert result.output == '--application_name required\nAborted!\n'

def test_get_files_missing_platform_name(self):
        runner = CliRunner()
        result = runner.invoke(get_files, ['--application_name', 'toto'])
        assert result.output == '--platform_name required\nAborted!\n'

def test_get_files_missing_path(self):
        runner = CliRunner()
        result = runner.invoke(get_files, ['--application_name', 'toto', '--platform_name', 'titi'])
        assert result.output == '--path required\nAborted!\n'

def test_get_files_missing_module_name(self):
        runner = CliRunner()
        result = runner.invoke(get_files, ['--application_name', 'toto', '--platform_name', 'titi', '--path', '#'])
        assert result.output == '--module_name required\nAborted!\n'

def test_get_files_missing_module_version(self):
        runner = CliRunner()
        result = runner.invoke(get_files,
                               ['--application_name', 'toto', '--platform_name', 'titi', '--path', '#', '--module_name',
                                'tata'])
        assert result.output == '--module_version required\nAborted!\n'

def test_get_application_missing_application_name(self):
        runner = CliRunner()
        result = runner.invoke(get_application)
        assert result.exit_code == 1
        assert result.output == '--application_name required\nAborted!\n'

def test_get_application_from_module_missing_module(self):
        runner = CliRunner()
        result = runner.invoke(get_application_from_module)
        assert result.exit_code == 1
        assert result.output == '--module required\nAborted!\n'

def test_get_application_from_module_missing_version(self):
        runner = CliRunner()
        result = runner.invoke(get_application_from_module, ['--module', 'toto'])
        assert result.exit_code == 1
        assert result.output == '--version required\nAborted!\n'

def test_get_application_from_module_missing_type(self):
        runner = CliRunner()
        result = runner.invoke(get_application_from_module, ['--module', 'toto', '--version', 'titi'])
        assert result.exit_code == 1
        assert result.output == '--type required\nAborted!\n'

def test_perform_search_applications_missing_name(self):
        runner = CliRunner()
        result = runner.invoke(perform_search_applications)
        assert result.exit_code == 1
        assert result.output == '--name required\nAborted!\n'

def test_create_template_package_missing_from_package_version_when_from_package_name_is_given(self):
        runner = CliRunner()
        result = runner.invoke(create_template_package, ['--from_package_name', 'toto'])
        assert result.output == '--from_package_version required when --from_package_name is given\nAborted!\n'

def test_create_template_package_missing_body(self):
        runner = CliRunner()
        result = runner.invoke(create_template_package)
        assert result.output == '--body required\nAborted!\n'

def test_create_template_package_workingcopy_missing_package_name(self):
        runner = CliRunner()
        result = runner.invoke(create_template_package_workingcopy)
        assert result.output == '--package_name required\nAborted!\n'

def test_create_template_package_workingcopy_missing_package_version(self):
        runner = CliRunner()
        result = runner.invoke(create_template_package_workingcopy, ['--package_name', 'toto'])
        assert result.output == '--package_version required\nAborted!\n'

def test_create_template_package_workingcopy_missing_template_name(self):
        runner = CliRunner()
        result = runner.invoke(create_template_package_workingcopy,
                               ['--package_name', 'toto', '--package_version', '1.0.0'])
        assert result.output == '--template_name required\nAborted!\n'

def test_delete_template_package_release_missing_package_name(self):
        runner = CliRunner()
        result = runner.invoke(delete_template_package_release)
        assert result.output == '--package_name required\nAborted!\n'

def test_delete_template_package_release_missing_package_version(self):
        runner = CliRunner()
        result = runner.invoke(delete_template_package_release, ['--package_name', 'toto'])
        assert result.output == '--package_version required\nAborted!\n'

def test_delete_template_package_workingcopy_missing_package_name(self):
        runner = CliRunner()
        result = runner.invoke(delete_template_package_workingcopy)
        assert result.output == '--package_name required\nAborted!\n'

def test_delete_template_package_workingcopy_missing_package_version(self):
        runner = CliRunner()
        result = runner.invoke(delete_template_package_workingcopy, ['--package_name', 'toto'])
        assert result.output == '--package_version required\nAborted!\n'