93 lines
2.7 KiB
Python
93 lines
2.7 KiB
Python
"""Test pygount determinism with serial and parallel execution."""
|
|
|
|
import subprocess
|
|
import json
|
|
from pathlib import Path
|
|
import concurrent.futures
|
|
|
|
test_file = Path("pyucc/core/countings_impl.py")
|
|
print(f"Testing: {test_file}")
|
|
print(f"File size: {test_file.stat().st_size} bytes")
|
|
print()
|
|
|
|
|
|
def run_pygount(run_id):
|
|
"""Run pygount once and return results."""
|
|
proc = subprocess.run(
|
|
["pygount", "--format", "json", str(test_file)],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
data = json.loads(proc.stdout)
|
|
if isinstance(data, list):
|
|
item = data[0]
|
|
else:
|
|
item = data["files"][0]
|
|
|
|
return {
|
|
"id": run_id,
|
|
"lineCount": item.get("lineCount"),
|
|
"sourceCount": item.get("sourceCount"),
|
|
"codeCount": item.get("codeCount"),
|
|
"documentationCount": item.get("documentationCount"),
|
|
"emptyCount": item.get("emptyCount"),
|
|
"language": item.get("language"),
|
|
}
|
|
|
|
|
|
# Test serial
|
|
print("Serial execution (5 runs):")
|
|
serial_results = []
|
|
for i in range(5):
|
|
result = run_pygount(i + 1)
|
|
serial_results.append(result)
|
|
print(
|
|
f' Run {result["id"]}: code={result["codeCount"]}, '
|
|
f'doc={result["documentationCount"]}, empty={result["emptyCount"]}'
|
|
)
|
|
|
|
# Test parallel
|
|
print()
|
|
print("Parallel execution (10 runs):")
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
|
|
futures = [executor.submit(run_pygount, i + 1) for i in range(10)]
|
|
parallel_results = [f.result() for f in futures]
|
|
|
|
for result in parallel_results:
|
|
print(
|
|
f' Run {result["id"]}: code={result["codeCount"]}, '
|
|
f'doc={result["documentationCount"]}, empty={result["emptyCount"]}'
|
|
)
|
|
|
|
# Check consistency
|
|
print()
|
|
base = {k: v for k, v in serial_results[0].items() if k != "id"}
|
|
all_serial = all(
|
|
{k: v for k, v in r.items() if k != "id"} == base for r in serial_results
|
|
)
|
|
all_parallel = all(
|
|
{k: v for k, v in r.items() if k != "id"} == base for r in parallel_results
|
|
)
|
|
|
|
print(f"Serial results consistent: {all_serial}")
|
|
print(f"Parallel results consistent: {all_parallel}")
|
|
print(f"Serial == Parallel: {all_serial and all_parallel}")
|
|
|
|
if not (all_serial and all_parallel):
|
|
print()
|
|
print("FOUND INCONSISTENCIES!")
|
|
print("First result:", base)
|
|
if not all_serial:
|
|
print("Serial differences:")
|
|
for r in serial_results:
|
|
r_clean = {k: v for k, v in r.items() if k != "id"}
|
|
if r_clean != base:
|
|
print(f' Run {r["id"]}: {r_clean}')
|
|
if not all_parallel:
|
|
print("Parallel differences:")
|
|
for r in parallel_results:
|
|
r_clean = {k: v for k, v in r.items() if k != "id"}
|
|
if r_clean != base:
|
|
print(f' Run {r["id"]}: {r_clean}')
|