loadtest_symlink.py

#!/usr/local/bin/python3

import os
import random
import string
import tempfile
import time
from multiprocessing import Pool
from multiprocessing import cpu_count


def get_random_string(length):
    letters = string.ascii_lowercase
    return ''.join(random.choice(letters) for i in range(length))


def create_symlink():
    file_name = '/tmp/' + get_random_string(8)
    file_symlink = file_name + '_symlink_'
    with open(file_name, 'w') as f:
        f.write('Hello World')
    os.symlink(file_name, file_symlink)
    if not os.path.islink(file_symlink):
        raise Exception(f'Expected {file_symlink} to be a symlink')
    else:
        print(f'Invocation: {count}')
    os.remove(file_name)
    os.remove(file_symlink)


def f(x):
    global count
    count = 0
    set_time = os.environ['STRESS_MINS']
    timeout = time.time() + 60*float(set_time)  # X minutes from now
    while True:
        if time.time() > timeout:
            break
        create_symlink()
        count = count + 1

if __name__ == '__main__':
    processes = cpu_count()
    print ('utilizing %d cores\n' % processes)
    pool = Pool(processes)
    pool.map(f, range(processes))