Correctly detect if environment is interactive

src/rkd/process.py

@@ -45,20 +45,20 @@ def __init__(self):
         self.has_exited = False
 
 
-def is_a_tty() -> bool:
-    return os.isatty(sys.stdout.fileno())
-
-
 def check_call(command: str, script: Optional[str] = ''):
     if os.getenv('RKD_COMPAT_SUBPROCESS') == 'true':
         subprocess.check_call(command, shell=True)
         return
 
     os.environ['PYTHONUNBUFFERED'] = "1"
 
-    old_tty = termios.tcgetattr(sys.stdin)
+    try:
+        old_tty = termios.tcgetattr(sys.stdin)
+    except termios.error:
+        old_tty = None
+
+    is_interactive_session = old_tty is not None
     process_state = ProcessState()
-    is_interactive_session = is_a_tty()
 
     try:
         if is_interactive_session:
@@ -74,7 +74,8 @@ def check_call(command: str, script: Optional[str] = ''):
                                    bufsize=64, close_fds=ON_POSIX, universal_newlines=False, preexec_fn=os.setsid)
 
         out_buffer = TextBuffer(buffer_size=1024 * 10)
-        fd_thread = Thread(target=push_output, args=(process, primary_fd, out_buffer, process_state))
+        fd_thread = Thread(target=push_output,
+                           args=(process, primary_fd, out_buffer, process_state, is_interactive_session))
         fd_thread.daemon = True
         fd_thread.start()
 
@@ -91,9 +92,14 @@ def check_call(command: str, script: Optional[str] = ''):
         )
 
 
-def push_output(process, primary_fd, out_buffer: TextBuffer, process_state: ProcessState):
+def push_output(process, primary_fd, out_buffer: TextBuffer, process_state: ProcessState, is_interactive_session: bool):
+    to_select = [primary_fd]
+
+    if is_interactive_session:
+        to_select = [sys.stdin] + to_select
+
     while process.poll() is None:
-        r, w, e = select.select([sys.stdin, primary_fd], [], [])
+        r, w, e = select.select(to_select, [], [])
 
         if sys.stdin in r:
             d = os.read(sys.stdin.fileno(), 10240)

test/test_taskutil.py

@@ -149,6 +149,24 @@ def test_sh_rkd_in_rkd_shows_first_lines_on_error(self):
                 self.assertIn('Bartolomeo Vanzetti', out.getvalue(),
                               msg='Expected that output will be shown for std_redirect=%s' % std_redirect)
 
+    def test_non_interactive_session_returns_output(self):
+        """Checks functionally if process.py is implementing a fall-back for non-interactive sessions
+
+        "true |" part enforces the console to be non-interactive, which should cause
+        "termios.error: (25, 'Inappropriate ioctl for device')" that should be handled and interactive mode
+        should be turned off for stdin
+        """
+
+        task = InitTask()
+        task._io = IO()
+        io = IO()
+        out = StringIO()
+
+        with io.capture_descriptors(stream=out, enable_standard_out=False):
+            task.sh(''' true | python3 -m rkd --silent :sh -c 'echo "Strajk Kobiet! Jebac PiS!"' ''')
+
+        self.assertIn('Strajk Kobiet! Jebac PiS!', out.getvalue())
+
     def test_dollar_symbols_are_escaped_in_shell_commands(self):
         """Check that in envrionment variable there can be defined a value that contains dollar symbols"""