Get exit status implementation and tests (#99)

* Added get_exit_status functionality and tests
* Added get exit status with signal tests
* Updated changelog

Author: pkittenis

Changelog.rst

@@ -9,7 +9,7 @@ Changes
 
 * Added private key file formats to `ssh.key` and `ssh.key.Key.export_privkey_file_format` for exporting private key
   file with specified format - #96
-
+* Added `ssh.channel.Channel.get_exit_status` implementation and tests.
 
 
 1.1.1

ci/integration_tests/test_channel.py

@@ -123,6 +123,48 @@ def test_exit_code(self):
         status = chan.get_exit_status()
         self.assertEqual(status, 2)
 
+    def test_exit_state(self):
+        self._auth()
+        chan = self.session.channel_new()
+        self.assertEqual(chan.open_session(), 0)
+        self.assertEqual(chan.request_exec('exit 2'), 0)
+        self.assertEqual(chan.send_eof(), 0)
+        self.assertEqual(chan.close(), 0)
+        exit_code, signal, pcore_dumped = chan.get_exit_state()
+        self.assertEqual(exit_code, 2)
+
+    def test_exit_state_w_signal(self):
+        self._auth()
+        chan = self.session.channel_new()
+        my_sig = 'TERM'
+        chan.open_session()
+        chan.request_exec('sleep 5 && exit 0')
+        chan.send_eof()
+        chan.request_send_signal(my_sig)
+        chan.close()
+        exit_code, signal, pcore_dumped = chan.get_exit_state()
+        self.assertNotEqual(exit_code, 0)
+        self.assertEqual(signal, bytes(my_sig, 'utf-8'))
+
+    def test_exit_state_w_signal_non_blocking(self):
+        self._auth()
+        chan = self.session.channel_new()
+        my_sig = 'TERM'
+        chan.open_session()
+        chan.request_exec('sleep 5 && exit 0')
+        chan.send_eof()
+        chan.request_send_signal(my_sig)
+        chan.close()
+        self.session.set_blocking(0)
+        exit_code, signal, pcore_dumped = chan.get_exit_state()
+        while exit_code == SSH_AGAIN:
+            self.assertEqual(signal, b"")
+            self.assertFalse(pcore_dumped)
+            wait_socket(self.session, self.sock)
+            exit_code, signal, pcore_dumped = chan.get_exit_state()
+        self.assertNotEqual(exit_code, 0)
+        self.assertEqual(signal, bytes(my_sig, 'utf-8'))
+
     def test_long_running_execute(self):
         self._auth()
         chan = self.session.channel_new()

ssh/c_ssh.pxd

@@ -268,6 +268,10 @@ cdef extern from "libssh/libssh.h" nogil:
     int ssh_channel_change_pty_size(ssh_channel channel, int cols, int rows)
     int ssh_channel_close(ssh_channel channel)
     void ssh_channel_free(ssh_channel channel)
+    int ssh_channel_get_exit_state(ssh_channel channel,
+                                   uint32_t *pexit_code,
+                                   char **pexit_signal,
+                                   int *pcore_dumped)
     int ssh_channel_get_exit_status(ssh_channel channel)
     ssh_session ssh_channel_get_session(ssh_channel channel)
     int ssh_channel_is_closed(ssh_channel channel)