filesystem-read-write-organize

code_execution · verified · null · json · download .py

Read, write, move, and organize files on the local filesystem

import sys
import shutil
import tempfile
from pathlib import Path

# ----------------------------------------
# PRE_EXECUTION
# no registry fetch needed — stdlib only
# set up isolated temp workspace
# clean up on any failure
# ----------------------------------------

workspace = Path(tempfile.mkdtemp(prefix="checklist_fs_"))
print(f"PRE_EXECUTION: workspace created at {workspace} ✓")

try:

    # ----------------------------------------
    # EXECUTION
    # FM-2.6: use pathlib — not os.path string concat
    # FM-1.3: exist_ok=True — safe on retry
    # FM-1.1: atomic write via temp file + rename
    # ----------------------------------------

    # 1. create directory structure
    docs_dir = workspace / "documents"
    archive_dir = workspace / "archive"
    docs_dir.mkdir(exist_ok=True)    # FM-1.3: idempotent
    archive_dir.mkdir(exist_ok=True)

    print("EXECUTION: directories created ✓")

    # 2. write file — atomic pattern (write to temp, rename)
    # FM-1.1: prevents partial writes if process dies mid-write
    target = docs_dir / "hello.txt"
    tmp = target.with_suffix(".tmp")
    tmp.write_text("hello world", encoding="utf-8")
    tmp.rename(target)

    print("EXECUTION: file written atomically ✓")

    # 3. read file back
    content = target.read_text(encoding="utf-8")

    # FM-3.3: content check, not just existence
    assert content == "hello world", \
        f"FAIL: expected 'hello world', got '{content}'"

    print("EXECUTION: file read verified ✓")

    # 4. write a second file
    second = docs_dir / "notes.txt"
    second.write_text("agent notes", encoding="utf-8")

    # 5. list directory — verify both files present
    files = list(docs_dir.iterdir())
    assert len(files) == 2, \
        f"FAIL: expected 2 files in docs_dir, got {len(files)}"

    print(f"EXECUTION: directory contains {len(files)} files ✓")

    # 6. move file to archive
    archived = archive_dir / target.name
    shutil.move(str(target), str(archived))

    # FM-3.2: verify move — check source gone AND dest exists
    assert not target.exists(), \
        "FAIL: source file still exists after move"
    assert archived.exists(), \
        "FAIL: destination file does not exist after move"

    print("EXECUTION: file moved to archive ✓")

    # 7. rename file
    renamed = archive_dir / "hello_renamed.txt"
    archived.rename(renamed)

    assert renamed.exists(), \
        "FAIL: renamed file does not exist"
    assert not archived.exists(), \
        "FAIL: original name still exists after rename"

    print("EXECUTION: file renamed ✓")

    # ----------------------------------------
    # POST_EXECUTION
    # FM-3.2: verify final state completely
    # FM-3.3: content round-trip on renamed file
    # ----------------------------------------

    # verify final workspace state
    assert (workspace / "documents").exists(), \
        "FAIL: documents dir missing"
    assert (workspace / "archive").exists(), \
        "FAIL: archive dir missing"

    # one file in docs (notes.txt), one in archive (hello_renamed.txt)
    doc_files = list(docs_dir.iterdir())
    arc_files = list(archive_dir.iterdir())

    assert len(doc_files) == 1, \
        f"FAIL: expected 1 file in documents, got {len(doc_files)}"
    assert len(arc_files) == 1, \
        f"FAIL: expected 1 file in archive, got {len(arc_files)}"

    # FM-3.3: content round-trip check
    final_content = renamed.read_text(encoding="utf-8")
    assert final_content == "hello world", \
        f"FAIL: content corrupted after move/rename, got '{final_content}'"

    print()
    print("POST_EXECUTION: directory structure verified ✓")
    print("POST_EXECUTION: content round-trip verified ✓")

    result = {
        "status": "pass",
        "atomic_write_verified": True,
        "content_round_trip_verified": True,
        "move_verified": True,
        "rename_verified": True,
        "directory_structure_verified": True,
    }
    print(result)
    print("PASS")

finally:
    # always clean up workspace
    shutil.rmtree(workspace, ignore_errors=True)
    print(f"POST_EXECUTION: workspace cleaned up ✓")