<commit_before>from twisted.internet import defer, reactor
from twisted.trial.unittest import TestCase
from carnifex.sshprocess import SSHProcessInductor
from twisted.internet.protocol import ProcessProtocol
from twisted.internet.error import ProcessTerminated, ProcessDone

USER = None # None means to defaults to current user
HOST = 'localhost'
PORT = 22

SUCCEEDING_COMMAND = 'true'
FAILING_COMMAND = 'false'
ECHO_COMMAND = 'echo'

class InductorTest(TestCase):
    def setUp(self):
        self.inductor = SSHProcessInductor(reactor, HOST, PORT)

    def tearDown(self):
        inductor = self.inductor
        self.inductor = None
        inductor.disconnectAll()

    def test_execute_exitcode_0(self):
        executable = SUCCEEDING_COMMAND
        disconnectedDeferred = defer.Deferred()
        protocol = ProcessProtocol()
        protocol.processEnded = disconnectedDeferred.callback
        self.inductor.execute(protocol, executable)
        return self.assertFailure(disconnectedDeferred, ProcessDone)

    def test_execute_nonzero_exitcode(self):
        executable = FAILING_COMMAND
        disconnectedDeferred = defer.Deferred()
        protocol = ProcessProtocol()
        protocol.processEnded = disconnectedDeferred.callback
        self.inductor.execute(protocol, executable)
        return self.assertFailure(disconnectedDeferred, ProcessTerminated)

    def test_execute_echo_return(self):
        executable = ECHO_COMMAND
        echoText = "some text to be echoed"
        expectedResult = (echoText + '\n', '', 0)

        disconnectedDeferred = defer.Deferred()
        protocol = ProcessProtocol()
        protocol.processEnded = disconnectedDeferred.callback
        resultDeferred = self.inductor.run(executable, (executable, echoText))
        @resultDeferred.addCallback
        def checkResult(result):
            self.assertEqual(result, expectedResult)

        return resultDeferred<commit_msg>Move generic inductor test methods to a mixin<commit_after>from twisted.internet import defer
from twisted.trial.unittest import TestCase
from carnifex.sshprocess import SSHProcessInductor
from twisted.internet.protocol import ProcessProtocol
from twisted.internet.error import ProcessTerminated, ProcessDone

UID = None # indicate that we want to run processes as the current user.

SUCCEEDING_COMMAND = 'true' # `true`should return exitcode 0
FAILING_COMMAND = 'false' # `false`should return a nonzero exitcode
ECHO_COMMAND = 'echo' # 'echo` should echo back on stdout

# `echo` return an additional newline after the given text to echo.
echoTransform = lambda text: text + '\n'


class InductorTestMixin(object):
    def test_execute_exitcode_0(self):
        """Check that we get the expected reason failure object
        when a process exits with exit code 0.
        """
        disconnectedDeferred = defer.Deferred()
        protocol = ProcessProtocol()
        protocol.processEnded = disconnectedDeferred.callback
        self.inductor.execute(protocol, SUCCEEDING_COMMAND, uid=UID)
        return self.assertFailure(disconnectedDeferred, ProcessDone)

    def test_execute_nonzero_exitcode(self):
        """Check that we get the expected failure when a process exits
        with a nonzero exit code.
        """
        disconnectedDeferred = defer.Deferred()
        protocol = ProcessProtocol()
        protocol.processEnded = disconnectedDeferred.callback
        self.inductor.execute(protocol, FAILING_COMMAND, uid=UID)
        # Process should return a ProcessTerminated failure
        # when it exits with a nonzero exit code
        return self.assertFailure(disconnectedDeferred, ProcessTerminated)

    def test_execute_echo_return(self):
        """Check that we get the expected stdout, stderr and exit code
        """
        executable = ECHO_COMMAND
        echoText = "some text to be echoed"
        args = (executable, echoText)

        disconnectedDeferred = defer.Deferred()
        protocol = ProcessProtocol()
        protocol.processEnded = disconnectedDeferred.callback
        resultDeferred = self.inductor.run(executable, args, uid=self.uid)
        @resultDeferred.addCallback
        def checkResult((stdin, stderr, exitCode)):
            self.assertEqual(stdin, echoTransform(echoText), "stdout not as expected")
            self.assertEqual(stderr, '', "unexpected data on stderr")
            self.assertEqual(exitCode, 0, "nonzero exit code")

        return resultDeferred
class SSHProcessInductorTest(TestCase, InductorTestMixin):
    """This test need to authenticate the connect to an ssh server.
    """
    def setUp(self):
        from twisted.internet import reactor
        self.uid = None # indicate that we want to run as the current user.
        self.host = 'localhost'
        self.port = 22
        self.inductor = SSHProcessInductor(reactor, self.host, self.port)

    def tearDown(self):
        InductorTestMixin
        inductor = self.inductor
        self.inductor = None
        inductor.disconnectAll()
{'fraction_non_alphanumeric': 0.07303931351027738, 'fraction_numerical': 0.001995609658750748, 'mean_word_length': 3.3393939393939394, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4259508', 'n_tokens_mistral': 1382, 'n_tokens_neox': 1323, 'n_words': 376}