Testing for the method stream_dir
inside of ansible-runner.
Run:
python size_test.py
You have to hard-code in the location to examine.
import os | |
import io | |
import time | |
import stat | |
from ansible_runner.utils.streaming import stream_dir | |
# source_directory = '/home/alancoding/repos/awx' | |
source_directory = '.' | |
s = time.time() | |
for dirpath, dirs, files in os.walk(source_directory): | |
for fname in files + dirs: | |
pass | |
print(f'Took {time.time() - s} to crawl all of {source_directory}') | |
s = time.time() | |
for dirpath, dirs, files in os.walk(source_directory): | |
for fname in files + dirs: | |
full_path = os.path.join(dirpath, fname) | |
os.path.islink(full_path) | |
print(f'Took {time.time() - s} to crawl all of {source_directory} while checking for symlinks') | |
ct = 0 | |
for dirpath, dirs, files in os.walk(source_directory): | |
for fname in files + dirs: | |
full_path = os.path.join(dirpath, fname) | |
if os.path.islink(full_path): | |
continue | |
ct += 1 | |
print(f'That includes {ct} objects') | |
s = time.time() | |
for dirpath, dirs, files in os.walk(source_directory): | |
for fname in files + dirs: | |
full_path = os.path.join(dirpath, fname) | |
if os.path.islink(full_path): | |
continue | |
stat.S_ISFIFO(os.stat(full_path).st_mode) | |
print(f'Took {time.time() - s} to crawl all of {source_directory} with stat') | |
s = time.time() | |
outgoing_buffer = io.BytesIO() | |
outgoing_buffer.name = 'not_stdout' | |
stream_dir(source_directory, outgoing_buffer) | |
print(f'Took {time.time() - s} to actually stream to in-memory buffer') |