spark: fs-report.py - list files sorted by size as JSON
This commit is contained in:
1 parent
a05acda49a
commit
b435f23dab
3 files changed
+151
No files matched your search
@@ -0,0 +1,34 @@
|
||||
#!/usr/bin/env python3
|
||||
"""List files in a directory sorted by size, output as JSON."""
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
def generate_report(directory: str) -> dict:
|
||||
"""Generate a file size report for a directory."""
|
||||
path = Path(directory)
|
||||
if not path.exists():
|
||||
return {"error": f"Directory does not exist: {directory}", "files": []}
|
||||
if not path.is_dir():
|
||||
return {"error": f"Path is not a directory: {directory}", "files": []}
|
||||
|
||||
files = []
|
||||
for item in path.iterdir():
|
||||
if item.is_file():
|
||||
files.append({
|
||||
"name": item.name,
|
||||
"size": item.stat().st_size,
|
||||
"path": str(item)
|
||||
})
|
||||
|
||||
files.sort(key=lambda x: x["size"], reverse=True)
|
||||
return {"files": files, "total": len(files), "directory": directory}
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) != 2:
|
||||
print(json.dumps({"error": "Usage: fs-report.py <directory>"}))
|
||||
sys.exit(1)
|
||||
|
||||
report = generate_report(sys.argv[1])
|
||||
print(json.dumps(report, indent=2))
|
||||
@@ -0,0 +1,69 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Test suite for fs-report.py - verifies the script produces valid sorted JSON output."""
|
||||
import subprocess
|
||||
import json
|
||||
import sys
|
||||
|
||||
def test_fs_report():
|
||||
# Run the script
|
||||
result = subprocess.run(
|
||||
['python3', '/root/.hermes/wizard-outputs/spark/fs-report.py', '/tmp'],
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
print(f'FAIL: Script returned non-zero exit code: {result.returncode}')
|
||||
print(f'stderr: {result.stderr}')
|
||||
return False
|
||||
|
||||
# Parse JSON output
|
||||
try:
|
||||
report = json.loads(result.stdout)
|
||||
except json.JSONDecodeError as e:
|
||||
print(f'FAIL: Invalid JSON output: {e}')
|
||||
return False
|
||||
|
||||
# Verify structure
|
||||
if 'files' not in report:
|
||||
print('FAIL: Missing "files" key in output')
|
||||
return False
|
||||
if 'total' not in report:
|
||||
print('FAIL: Missing "total" key in output')
|
||||
return False
|
||||
if 'directory' not in report:
|
||||
print('FAIL: Missing "directory" key in output')
|
||||
return False
|
||||
|
||||
files = report['files']
|
||||
if len(files) < 2:
|
||||
print('FAIL: Need at least 2 files')
|
||||
return False
|
||||
|
||||
# Check descending order by size
|
||||
for i in range(len(files) - 1):
|
||||
if files[i]['size'] < files[i + 1]['size']:
|
||||
print(f'FAIL: Not sorted by size at position {i}: {files[i]["size"]} < {files[i+1]["size"]}')
|
||||
return False
|
||||
|
||||
# Verify all files have required keys
|
||||
for f in files:
|
||||
for key in ['name', 'size', 'path']:
|
||||
if key not in f:
|
||||
print(f'FAIL: Missing key "{key}" in file entry')
|
||||
return False
|
||||
|
||||
# Verify total matches
|
||||
if report['total'] != len(files):
|
||||
print(f'FAIL: total ({report["total"]}) != len(files) ({len(files)})')
|
||||
return False
|
||||
|
||||
print(f'PASS: {len(files)} files correctly sorted and structured')
|
||||
print(f'PASS: All required keys present')
|
||||
print(f'PASS: total count matches: {report["total"]}')
|
||||
print('All tests passed')
|
||||
return True
|
||||
|
||||
if __name__ == '__main__':
|
||||
success = test_fs_report()
|
||||
sys.exit(0 if success else 1)
|
||||
@@ -0,0 +1,48 @@
|
||||
# Spark Lane Verification Script
|
||||
# Used for snippet-level micro-tasks in the spark lane
|
||||
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timezone
|
||||
|
||||
def verify_word_counter():
|
||||
"""Verify word_counter.py functions correctly."""
|
||||
sys.path.insert(0, '/root/troupe-standup/wizard-work/spark')
|
||||
from word_counter import count_words
|
||||
|
||||
tests = [
|
||||
("Hello world test", 3),
|
||||
("", 0),
|
||||
(" ", 0),
|
||||
("one", 1),
|
||||
("one two three four five", 5),
|
||||
]
|
||||
|
||||
passed = 0
|
||||
for text, expected in tests:
|
||||
result = count_words(text)
|
||||
if result == expected:
|
||||
passed += 1
|
||||
|
||||
return passed == len(tests), passed, len(tests)
|
||||
|
||||
def main():
|
||||
print("=== SPARK LANE VERIFICATION ===")
|
||||
print(f"Timestamp: {datetime.now(timezone.utc).isoformat()}")
|
||||
print()
|
||||
|
||||
# Run verification
|
||||
success, passed, total = verify_word_counter()
|
||||
|
||||
print(f"word_counter.py: {passed}/{total} tests passed")
|
||||
|
||||
if success:
|
||||
print("\nVERDICT: All snippets verified successfully")
|
||||
sys.exit(0)
|
||||
else:
|
||||
print("\nVERDICT: Some snippets failed verification")
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user