Readme.md

Tests for file handle inheritance issues

1. test_write.py uses subprocess.call without redirection

• This fails on Python 2.x because the close_fds argument to call() defaults to False and file handles are inheritable by default
• This succeeds on Python 3.x because the close_fds argument to call() defaults to True when redirection is not used

2. test_write_redirect.py uses subprocess.call with stdout redirection

• This fails on Python 2.x because the close_fds argument to call() defaults to False and file handles are inheritable by default
• This fails on Python 3.x (x < 4) because the close_fds argument to call() defaults to False when redirection is used and file handles are inheritable by default
• This succeeds on Python 3.4 because file handles are non-inheritable by default

child.c

#include <windows.h>

int main(int argc, char**argv)
{
    HANDLE handle = CreateFile(argv[1], GENERIC_READ, FILE_SHARE_READ, NULL,
                               OPEN_EXISTING, 0, NULL);
    int size = GetFileSize(handle, NULL);
    CloseHandle(handle);

    return size;
}

test_write.py

#!/usr/bin/env python
from __future__ import print_function
import sys
import os
import subprocess
import tempfile
import threading
import time

def do_work():
    indata = b"[numthreads(32,32,1)] void main() {}"
    infd, infilename = tempfile.mkstemp()
    os.close(infd)

    try:
        with open(infilename, 'wb') as infile:
            infile.write(indata)
            infile.flush()
            os.fsync(infile.fileno())

        cmdline = ["child", infilename]
    
        retcode = subprocess.call(cmdline)
    
        if retcode != len(indata):
            print("Failure, got {} but expected {}".format(retcode, len(indata)))
    finally:
        time.sleep(0.5)
        os.remove(infilename)

def start_thread():
    thread = threading.Thread(target=do_work)
    thread.start()
    return thread

def main():
    num_threads = 10
    print("Spawning {} threads".format(num_threads))
    threads = [ start_thread() for x in range(num_threads) ]
    results = [ thread.join() for thread in threads ]
        

if __name__ == "__main__":
    main()

test_write_redirect.py

#!/usr/bin/env python
from __future__ import print_function
import sys
import os
import subprocess
import tempfile
import threading
import time

def do_work():
    indata = b"[numthreads(32,32,1)] void main() {}"
    infd, infilename = tempfile.mkstemp()
    os.close(infd)

    try:
        with open(infilename, 'wb') as infile:
            infile.write(indata)
            infile.flush()
            os.fsync(infile.fileno())

        cmdline = ["child", infilename]
    
        with open("stdout.txt", "wb") as stdout:
            retcode = subprocess.call(cmdline, stdout=stdout)
    
        if retcode != len(indata):
            print("Failure, got {} but expected {}".format(retcode, len(indata)))
    finally:
        time.sleep(0.5)
        os.remove(infilename)

def start_thread():
    thread = threading.Thread(target=do_work)
    thread.start()
    return thread

def main():
    num_threads = 10
    print("Spawning {} threads".format(num_threads))
    threads = [ start_thread() for x in range(num_threads) ]
    results = [ thread.join() for thread in threads ]
        

if __name__ == "__main__":
    main()