mirror of
				https://github.com/python/cpython.git
				synced 2025-11-03 23:21:29 +00:00 
			
		
		
		
	Add a subprocess test of remapping standard file descriptors (issue #1187).
This commit is contained in:
		
							parent
							
								
									4d41968904
								
							
						
					
					
						commit
						95aaeee59a
					
				
					 1 changed files with 48 additions and 0 deletions
				
			
		| 
						 | 
				
			
			@ -955,6 +955,54 @@ def test_close_fds_0_1_2(self):
 | 
			
		|||
        # all standard fds closed.
 | 
			
		||||
        self.check_close_std_fds([0, 1, 2])
 | 
			
		||||
 | 
			
		||||
    def test_remapping_std_fds(self):
 | 
			
		||||
        # open up some temporary files
 | 
			
		||||
        temps = [mkstemp() for i in range(3)]
 | 
			
		||||
        try:
 | 
			
		||||
            temp_fds = [fd for fd, fname in temps]
 | 
			
		||||
 | 
			
		||||
            # unlink the files -- we won't need to reopen them
 | 
			
		||||
            for fd, fname in temps:
 | 
			
		||||
                os.unlink(fname)
 | 
			
		||||
 | 
			
		||||
            # write some data to what will become stdin, and rewind
 | 
			
		||||
            os.write(temp_fds[1], b"STDIN")
 | 
			
		||||
            os.lseek(temp_fds[1], 0, 0)
 | 
			
		||||
 | 
			
		||||
            # move the standard file descriptors out of the way
 | 
			
		||||
            saved_fds = [os.dup(fd) for fd in range(3)]
 | 
			
		||||
            try:
 | 
			
		||||
                # duplicate the file objects over the standard fd's
 | 
			
		||||
                for fd, temp_fd in enumerate(temp_fds):
 | 
			
		||||
                    os.dup2(temp_fd, fd)
 | 
			
		||||
 | 
			
		||||
                # now use those files in the "wrong" order, so that subprocess
 | 
			
		||||
                # has to rearrange them in the child
 | 
			
		||||
                p = subprocess.Popen([sys.executable, "-c",
 | 
			
		||||
                    'import sys; got = sys.stdin.read();'
 | 
			
		||||
                    'sys.stdout.write("got %s"%got); sys.stderr.write("err")'],
 | 
			
		||||
                    stdin=temp_fds[1],
 | 
			
		||||
                    stdout=temp_fds[2],
 | 
			
		||||
                    stderr=temp_fds[0])
 | 
			
		||||
                p.wait()
 | 
			
		||||
            finally:
 | 
			
		||||
                # restore the original fd's underneath sys.stdin, etc.
 | 
			
		||||
                for std, saved in enumerate(saved_fds):
 | 
			
		||||
                    os.dup2(saved, std)
 | 
			
		||||
                    os.close(saved)
 | 
			
		||||
 | 
			
		||||
            for fd in temp_fds:
 | 
			
		||||
                os.lseek(fd, 0, 0)
 | 
			
		||||
 | 
			
		||||
            out = os.read(temp_fds[2], 1024)
 | 
			
		||||
            err = support.strip_python_stderr(os.read(temp_fds[0], 1024))
 | 
			
		||||
            self.assertEqual(out, b"got STDIN")
 | 
			
		||||
            self.assertEqual(err, b"err")
 | 
			
		||||
 | 
			
		||||
        finally:
 | 
			
		||||
            for fd in temp_fds:
 | 
			
		||||
                os.close(fd)
 | 
			
		||||
 | 
			
		||||
    def test_surrogates_error_message(self):
 | 
			
		||||
        def prepare():
 | 
			
		||||
            raise ValueError("surrogate:\uDCff")
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
		Loading…
	
	Add table
		Add a link
		
	
		Reference in a new issue