Private
Public Access
60 lines
2.5 KiB
Python
60 lines
2.5 KiB
Python
import concurrent.futures
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
|
|
helper = Path('scripts/test_native_build.py').resolve()
|
|
with tempfile.TemporaryDirectory(dir='tmp', prefix='native-cache-probe.') as work:
|
|
root = Path(work).resolve()
|
|
header = root / 'nested header.h'
|
|
header.write_text('#define VALUE 0\n')
|
|
source = root / 'a.cpp'
|
|
source.write_text('#include "nested header.h"\nint main(){return VALUE;}\n')
|
|
other = root / 'b.cpp'
|
|
other.write_text('int other(){return 0;}\n')
|
|
output = root / 'test'
|
|
cmd = ['python3', str(helper), str(output), '--', 'g++', str(source), str(other)]
|
|
def run(command=cmd, success=True):
|
|
result = subprocess.run(command, timeout=30)
|
|
assert (result.returncode == 0) == success
|
|
return output.stat().st_mtime_ns
|
|
first = run()
|
|
assert run() == first
|
|
state = json.loads(Path(str(output)+'.json').read_text())
|
|
assert str(header) in state['dependencies'] and str(source) in state['dependencies']
|
|
times = header.stat()
|
|
header.write_text('#define VALUE 1\n')
|
|
os.utime(header, ns=(times.st_atime_ns, times.st_mtime_ns))
|
|
changed = run()
|
|
assert changed != first and subprocess.run([output], timeout=10).returncode == 1
|
|
header.write_text('#define VALUE 0\n')
|
|
os.utime(header, ns=(times.st_atime_ns, times.st_mtime_ns))
|
|
reverted = run()
|
|
assert reverted != changed and subprocess.run([output], timeout=10).returncode == 0
|
|
flagged = run(cmd+['-DOTHER=1'])
|
|
assert flagged != reverted and run(cmd+['-DOTHER=1']) == flagged
|
|
run(cmd+['-invalid-flag'], False)
|
|
assert not Path(str(output)+'.json').exists()
|
|
run()
|
|
source.write_text('broken source\n')
|
|
run(success=False)
|
|
assert not Path(str(output)+'.json').exists()
|
|
source.write_text('#include "nested header.h"\nint main(){return VALUE;}\n')
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
|
|
stamps = list(pool.map(lambda _: run(), range(4)))
|
|
assert len(set(stamps)) == 1
|
|
output.write_bytes(b'corrupt')
|
|
run()
|
|
assert subprocess.run([output], timeout=10).returncode == 0
|
|
wrapper = root / 'compiler'
|
|
wrapper.write_text('#!/bin/sh\nexec g++ "$@"\n')
|
|
wrapper.chmod(0o755)
|
|
custom = cmd.copy()
|
|
custom[4] = str(wrapper)
|
|
before = run(custom)
|
|
wrapper.write_text('#!/bin/sh\n# changed compiler\nexec g++ "$@"\n')
|
|
assert run(custom) != before
|
|
print('PASS: first/hit, transitive content/timestamp/revert, flags, failed scan/build, concurrency, binary corruption, compiler identity')
|