Back to Repositories

Testing Command Output Processing Workflow in thefuck

This test suite validates the rerun output reader functionality in thefuck, focusing on command output capture and process management. It ensures reliable command execution and output handling across different scenarios including timeout and encoding cases.

Test Coverage Overview

The test suite provides comprehensive coverage of the rerun output reader component.

Key areas tested include:

Basic command output capture and encoding handling
Unicode and special character processing
Process timeout management
Platform-specific behavior (Windows vs Unix)
Error handling for access denied scenarios

Implementation Analysis

The implementation uses pytest’s class-based test structure with extensive mocking to isolate system interactions.

Notable patterns include:

Setup/teardown methods for consistent test state
Mock patching for process and system calls
Platform-specific test skipping
Timeout and process termination handling

Technical Details

Testing tools and configuration:

pytest as the primary testing framework
mock library for test doubles
psutil for process management
Platform-specific conditionals via pytest.mark.skipif
Unicode handling for international character support

Best Practices Demonstrated

The test suite exemplifies several testing best practices.

Notable examples include:

Proper test isolation through mock usage
Consistent resource cleanup in teardown
Comprehensive error case coverage
Clear test method naming conventions
Effective use of pytest fixtures and decorators

nvbn/thefuck

tests/output_readers/test_rerun.py

            
# -*- encoding: utf-8 -*-

import pytest
import sys
from mock import Mock, patch
from psutil import AccessDenied, TimeoutExpired

from thefuck.output_readers import rerun


class TestRerun(object):
    def setup_method(self, test_method):
        self.patcher = patch('thefuck.output_readers.rerun.Process')
        process_mock = self.patcher.start()
        self.proc_mock = process_mock.return_value = Mock()

    def teardown_method(self, test_method):
        self.patcher.stop()

    @patch('thefuck.output_readers.rerun._wait_output', return_value=False)
    @patch('thefuck.output_readers.rerun.Popen')
    def test_get_output(self, popen_mock, wait_output_mock):
        popen_mock.return_value.stdout.read.return_value = b'output'
        assert rerun.get_output('', '') is None
        wait_output_mock.assert_called_once()

    @patch('thefuck.output_readers.rerun.Popen')
    def test_get_output_invalid_continuation_byte(self, popen_mock):
        output = b'ls: illegal option -- \xc3
usage: ls [-@ABC...] [file ...]
'
        expected = u'ls: illegal option -- \ufffd
usage: ls [-@ABC...] [file ...]
'
        popen_mock.return_value.stdout.read.return_value = output
        actual = rerun.get_output('', '')
        assert actual == expected

    @pytest.mark.skipif(sys.platform == 'win32', reason="skip when running on Windows")
    @patch('thefuck.output_readers.rerun._wait_output')
    def test_get_output_unicode_misspell(self, wait_output_mock):
        rerun.get_output(u'pácman', u'pácman')
        wait_output_mock.assert_called_once()

    def test_wait_output_is_slow(self, settings):
        assert rerun._wait_output(Mock(), True)
        self.proc_mock.wait.assert_called_once_with(settings.wait_slow_command)

    def test_wait_output_is_not_slow(self, settings):
        assert rerun._wait_output(Mock(), False)
        self.proc_mock.wait.assert_called_once_with(settings.wait_command)

    @patch('thefuck.output_readers.rerun._kill_process')
    def test_wait_output_timeout(self, kill_process_mock):
        self.proc_mock.wait.side_effect = TimeoutExpired(3)
        self.proc_mock.children.return_value = []
        assert not rerun._wait_output(Mock(), False)
        kill_process_mock.assert_called_once_with(self.proc_mock)

    @patch('thefuck.output_readers.rerun._kill_process')
    def test_wait_output_timeout_children(self, kill_process_mock):
        self.proc_mock.wait.side_effect = TimeoutExpired(3)
        self.proc_mock.children.return_value = [Mock()] * 2
        assert not rerun._wait_output(Mock(), False)
        assert kill_process_mock.call_count == 3

    def test_kill_process(self):
        proc = Mock()
        rerun._kill_process(proc)
        proc.kill.assert_called_once_with()

    @patch('thefuck.output_readers.rerun.logs')
    def test_kill_process_access_denied(self, logs_mock):
        proc = Mock()
        proc.kill.side_effect = AccessDenied()
        rerun._kill_process(proc)
        proc.kill.assert_called_once_with()
        logs_mock.debug.assert_called_once()