Hands-on lab — Day 64: Reading and Writing Files

"""Driver for the Safe File I/O Toolkit — the thin shell around the toolkit.

Run it from the lab directory:

    python3 examples/demo.py              # workspace defaults to ./workspace
    python3 examples/demo.py my-scratch   # or name your own workspace directory

Everything it creates lives inside the workspace directory, so cleanup is a
single `rm -rf workspace`. It makes no network calls and needs no
privileges.
"""

import sys
from pathlib import Path

import fileio_toolkit as toolkit

LINE_COUNT = 200_000
LONG_LINE_AT = 100_000


def mib(byte_count):
    """Bytes as a human-readable MiB string (1 MiB = 1024 * 1024 bytes)."""
    return f"{byte_count / (1024 * 1024):.2f} MiB"


def part_1_round_trip(workspace):
    print("[1] text round-trip with an explicit encoding")
    notes = workspace / "notes.txt"
    text = "Day 64: reading and writing files\ncafé, naïve, 日本語\nthird line\n"
    size = toolkit.write_text_file(notes, text)
    back = toolkit.read_text_file(notes)
    print(f"    wrote {notes.name}: {size} bytes on disk, {len(text)} characters in memory")
    print(f"    lines read back: {len(back.splitlines())}")
    print(f"    round-trip identical: {back == text}")
    print(f"    note: {size} bytes > {len(text)} characters — non-ASCII characters")
    print("          take more than one byte each in UTF-8")
    print()


def part_2_memory(workspace):
    print("[2] whole-file read vs line-by-line streaming")
    big = workspace / "big.log"
    size = toolkit.make_big_file(big, LINE_COUNT, long_line_at=LONG_LINE_AT)
    print(f"    generated {big.name}: {LINE_COUNT} lines, {size} bytes ({mib(size)})")

    whole_len, whole_peak = toolkit.longest_line_whole_file(big)
    stream_len, stream_peak = toolkit.longest_line_streaming(big)

    print(f"    read() + splitlines()  -> longest line {whole_len} chars, "
          f"peak memory {mib(whole_peak)}")
    print(f"    for line in handle     -> longest line {stream_len} chars, "
          f"peak memory {mib(stream_peak)}")
    print(f"    same answer: {whole_len == stream_len}")
    print(f"    streaming used {whole_peak / max(stream_peak, 1):.0f}x less memory")
    print()


def part_3_decoding(workspace):
    print("[3] decoding trouble and how to survive it")
    broken = workspace / "broken.log"
    # Written as RAW BYTES so we control exactly what lands on disk. 0xE9 is
    # "é" in latin-1, but on its own it is not valid UTF-8.
    broken.write_bytes(b"caf\xe9 was logged by a latin-1 program\n")
    print(f"    wrote {broken.name} as raw bytes: {broken.read_bytes()!r}")

    try:
        toolkit.read_text_file(broken)
        print("    strict utf-8 read: no error (unexpected)")
    except UnicodeDecodeError as err:
        print(f"    strict utf-8 read raised UnicodeDecodeError: {err}")

    text, recovered = toolkit.read_text_surviving_bad_bytes(broken)
    print(f"    errors='replace' read: {text.strip()!r}")
    print(f"    recovered: {recovered}; U+FFFD present: {chr(0xFFFD) in text}")
    print()


def part_4_atomic(workspace):
    print("[4] atomic write — proved by looking at the directory mid-operation")
    box = workspace / "config"
    box.mkdir(exist_ok=True)
    config = box / "config.txt"
    toolkit.write_text_file(config, "version = 1\n")
    print(f"    before: {sorted(p.name for p in box.iterdir())} "
          f"content={toolkit.read_text_file(config).strip()!r}")

    def peek(temp_path):
        during = sorted(p.name for p in box.iterdir())
        still_old = toolkit.read_text_file(config).strip()
        print(f"    during: {during} content={still_old!r}")
        print(f"            the new bytes are all in {temp_path.name}; readers of "
              f"{config.name} still get the OLD file, whole")

    toolkit.atomic_write_text(config, "version = 2\n", after_temp=peek)
    print(f"    after : {sorted(p.name for p in box.iterdir())} "
          f"content={toolkit.read_text_file(config).strip()!r}")

    # The contrast: the naive write, caught in the act.
    unsafe = box / "unsafe.txt"
    toolkit.write_text_file(unsafe, "version = 1\n")

    seen = {}

    def peek_unsafe(path):
        seen["mid"] = toolkit.read_text_file(path)

    toolkit.unsafe_write_text(unsafe, "version = 2 with more content\n", after_open=peek_unsafe)
    print(f"    contrast: mode 'w' straight onto the real file was caught mid-write "
          f"holding {seen['mid']!r}")
    print("              — a crash there destroys the old file and leaves that fragment")
    print()


def part_5_pathlib(workspace):
    print("[5] walking a directory tree with pathlib")
    tree = workspace / "tree"
    (tree / "sub" / "deeper").mkdir(parents=True, exist_ok=True)
    toolkit.write_text_file(tree / "a.txt", "a\n")
    toolkit.write_text_file(tree / "sub" / "b.txt", "b\n")
    toolkit.write_text_file(tree / "sub" / "deeper" / "c.txt", "c\n")

    found = toolkit.list_tree(tree)
    for relative in found:
        print(f"    {relative}")
    print(f"    {len(found)} file(s) found by Path.rglob")
    print(f"    tree exists: {tree.exists()}; is a directory: {tree.is_dir()}")
    print()


def main(argv):
    workspace = Path(argv[1] if len(argv) > 1 else "workspace")
    workspace.mkdir(parents=True, exist_ok=True)
    print(f"Safe File I/O Toolkit — workspace: {workspace}")
    print()
    part_1_round_trip(workspace)
    part_2_memory(workspace)
    part_3_decoding(workspace)
    part_4_atomic(workspace)
    part_5_pathlib(workspace)
    print(f"Done. Remove everything with: rm -rf {workspace}")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))

"""Safe file I/O toolkit — the reference implementation.

Nine small functions covering the five habits of reading and writing files
well:

  1. round-trip text with an EXPLICIT encoding             write_text_file / read_text_file
  2. compare whole-file reading with line-by-line reading  longest_line_whole_file /
                                                           longest_line_streaming
  3. survive undecodable bytes                             read_text_surviving_bad_bytes
  4. write without ever leaving a half-written file        atomic_write_text
  5. walk a directory tree with pathlib                    list_tree

Standard library only: os, tracemalloc, pathlib. Nothing here touches the
network, and every file this module opens is closed by a `with` block even
if an error is raised part-way through.
"""

import os
import tracemalloc
from pathlib import Path

# One padded record, so every generated line has the same length.
FILLER = "log line padding-padding-padding"


def write_text_file(path, text, encoding="utf-8"):
    """Write `text` to `path`, replacing anything already there.

    Mode "w" truncates the file to zero bytes before the first write, so
    this is a replace, not an append. The encoding is explicit on purpose:
    without it Python picks a locale-dependent default and the same code
    produces different bytes on different machines. `newline="\\n"` pins
    the line ending so the output is byte-identical everywhere.

    Returns the size of the finished file in bytes.
    """
    with open(path, "w", encoding=encoding, newline="\n") as handle:
        handle.write(text)
    return os.path.getsize(path)


def read_text_file(path, encoding="utf-8"):
    """Read the whole of `path` back as one string, decoding with `encoding`."""
    with open(path, "r", encoding=encoding) as handle:
        return handle.read()


def make_big_file(path, line_count, long_line_at=None):
    """Generate a large-ish log file: one numbered record per line.

    Every line has the same length except the one numbered `long_line_at`,
    which gets extra padding so the "longest line" question has a single
    correct answer. Returns the size of the file in bytes.
    """
    with open(path, "w", encoding="utf-8", newline="\n") as handle:
        for number in range(1, line_count + 1):
            extra = "!" * 20 if number == long_line_at else ""
            handle.write(f"{number:07d} {FILLER}{extra}\n")
    return os.path.getsize(path)


def longest_line_whole_file(path):
    """Length of the longest line, read the EXPENSIVE way.

    `handle.read()` pulls the entire file into one string, and
    `.splitlines()` then builds a list holding every line at once. Peak
    memory therefore scales with the size of the file: a 2 GB log needs
    more than 2 GB of RAM before a single line is examined.

    Returns (longest_length, peak_bytes); the peak is measured by
    `tracemalloc`, Python's built-in allocation tracker.
    """
    tracemalloc.start()
    with open(path, "r", encoding="utf-8") as handle:
        text = handle.read()
    lines = text.splitlines()
    longest = max((len(line) for line in lines), default=0)
    peak = tracemalloc.get_traced_memory()[1]
    tracemalloc.stop()
    return longest, peak


def longest_line_streaming(path):
    """Length of the longest line, read the CHEAP way.

    Iterating the file object hands you one line at a time and lets the
    previous line be reclaimed, so peak memory is the size of the longest
    single line plus a read buffer — flat, no matter how big the file is.

    Returns (longest_length, peak_bytes), the same answer as
    `longest_line_whole_file` for far less memory.
    """
    tracemalloc.start()
    longest = 0
    with open(path, "r", encoding="utf-8") as handle:
        for line in handle:
            length = len(line.rstrip("\n"))
            if length > longest:
                longest = length
    peak = tracemalloc.get_traced_memory()[1]
    tracemalloc.stop()
    return longest, peak


def read_text_surviving_bad_bytes(path, encoding="utf-8"):
    """Read `path` as text, surviving bytes that are not valid `encoding`.

    The first attempt decodes strictly, which is what you want by default:
    a `UnicodeDecodeError` tells you the file is not what you assumed, and
    silence would be worse. If it fails, the second attempt uses
    `errors="replace"`, which substitutes the replacement character U+FFFD
    (shown as an inverted question mark) for each undecodable byte instead
    of raising.

    Returns (text, recovered) where `recovered` is True when the fallback
    was needed.
    """
    try:
        with open(path, "r", encoding=encoding) as handle:
            return handle.read(), False
    except UnicodeDecodeError:
        with open(path, "r", encoding=encoding, errors="replace") as handle:
            return handle.read(), True


def atomic_write_text(path, text, encoding="utf-8", after_temp=None):
    """Replace `path` with `text` so no reader ever sees a half-written file.

    The pattern has four steps:

      1. write the new content to a temporary file IN THE SAME DIRECTORY
         (same directory so the final rename stays inside one filesystem),
      2. `flush()` so Python's buffer reaches the operating system,
      3. `os.fsync()` so the operating system writes it to the disk itself,
      4. `os.replace()` — an atomic rename: every reader sees either the
         complete old file or the complete new one, never a mixture.

    A crash before step 4 leaves the old file untouched and a stray `.tmp`
    beside it; a crash after step 4 leaves the complete new file.

    `after_temp`, if given, is called with the temporary path between steps
    3 and 4 — that is the moment the lab inspects, to show both files
    existing side by side.

    Returns the path that was written.
    """
    path = Path(path)
    temp = path.with_name(path.name + ".tmp")
    with open(temp, "w", encoding=encoding, newline="\n") as handle:
        handle.write(text)
        handle.flush()
        os.fsync(handle.fileno())
    if after_temp is not None:
        after_temp(temp)
    os.replace(temp, path)
    return path


def unsafe_write_text(path, text, encoding="utf-8", after_open=None):
    """The naive alternative, kept for contrast — do NOT copy this.

    Opening the real file in mode "w" truncates it immediately, so from
    that instant until the write completes the file on disk is empty or
    partial. A crash in that window destroys the old content and leaves
    nothing usable behind. `after_open` is called while the file is open
    and still incomplete, so the lab can look at the damage.
    """
    with open(path, "w", encoding=encoding, newline="\n") as handle:
        handle.write(text[: len(text) // 2])
        handle.flush()
        if after_open is not None:
            after_open(path)
        handle.write(text[len(text) // 2 :])
    return path


def list_tree(root):
    """Every file under `root`, as sorted relative paths, using pathlib.

    `Path.rglob("*")` walks the whole tree recursively and yields `Path`
    objects; `is_file()` filters out the directories; `relative_to(root)`
    turns each absolute path back into a short readable one. The same job
    with `os.path` string juggling needs `os.walk` plus `os.path.join` plus
    `os.path.relpath` and a lot of care about separators.
    """
    root = Path(root)
    return sorted(str(p.relative_to(root)) for p in root.rglob("*") if p.is_file())

lesson_id: D064
day: 64
kind: python-program
languages: [python]
setup_commands:
  - cd labs/sections/programming-with-python/day-064-reading-and-writing-files
  - python3 --version
run_commands:
  - python3 examples/demo.py
  - head -2 workspace/big.log
  - wc -l < workspace/big.log
  - python3 starter/demo.py
  - rm -rf workspace
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - rm -rf workspace
  - 'git checkout -- starter/fileio_toolkit.py  # optional: reset your work'
requires_network: false
requires_api_key: false
estimated_minutes: 30
last_executed: '2026-07-19'
executed_on: 'macOS (Apple Silicon), Python 3.14.0, bash tests/run_tests.sh -> 28 checks, 0 failure(s), exit 0'

# Dependencies — Day 064 lab

**Python 3 only. No third-party packages, no network, no privileges.**

- `python3` (3.8 or newer; tested on 3.14.0). Preinstalled on most Linux
  distributions and installable on macOS; you set this up on Day 43.
- `bash` for the test runner (preinstalled on macOS and Linux).
- Only the Python standard library is used: `os`, `pathlib`, `tracemalloc`,
  and `sys`. There is deliberately no `requirements.txt` — reading and
  writing files is a built-in capability, and the point of this lab is that
  the tools you need for it ship with Python.

Check your Python is present and new enough:

```bash
python3 --version
```

If that prints `Python 3.8` or higher, you are ready.

## Disk space

The lab generates a `workspace/` directory containing an 8.2 MB log file
(200,000 lines) so the memory comparison has something real to measure. Allow
about 10 MB of free space. Everything the lab creates lives inside
`workspace/`, and `rm -rf workspace` removes all of it.

## Platform notes

- **macOS and Linux:** every command runs as written.
- **Windows:** run the commands inside WSL. The Python code is portable —
  `pathlib` handles separators and the code passes `newline="\n"` explicitly
  where line endings matter — but the test runner is a bash script.
- `os.fsync` is available on macOS, Linux, and Windows. On network
  filesystems some devices acknowledge a sync before the bytes are truly on
  physical media; that is a property of the hardware, not of Python, and it
  does not affect anything this lab measures.

"""Driver for the Safe File I/O Toolkit — provided complete; do not edit it.

It imports `fileio_toolkit` from the directory next to it, so this copy
drives YOUR toolkit in `starter/fileio_toolkit.py`. Run it from the lab
directory:

    python3 starter/demo.py              # workspace defaults to ./workspace
    python3 starter/demo.py my-scratch   # or name your own workspace directory

Until you finish an exercise, the part that needs it stops with
`NotImplementedError` — that is the expected behaviour, not a bug.

Everything it creates lives inside the workspace directory, so cleanup is a
single `rm -rf workspace`. It makes no network calls and needs no
privileges.
"""

import sys
from pathlib import Path

import fileio_toolkit as toolkit

LINE_COUNT = 200_000
LONG_LINE_AT = 100_000


def mib(byte_count):
    """Bytes as a human-readable MiB string (1 MiB = 1024 * 1024 bytes)."""
    return f"{byte_count / (1024 * 1024):.2f} MiB"


def part_1_round_trip(workspace):
    print("[1] text round-trip with an explicit encoding")
    notes = workspace / "notes.txt"
    text = "Day 64: reading and writing files\ncafé, naïve, 日本語\nthird line\n"
    size = toolkit.write_text_file(notes, text)
    back = toolkit.read_text_file(notes)
    print(f"    wrote {notes.name}: {size} bytes on disk, {len(text)} characters in memory")
    print(f"    lines read back: {len(back.splitlines())}")
    print(f"    round-trip identical: {back == text}")
    print(f"    note: {size} bytes > {len(text)} characters — non-ASCII characters")
    print("          take more than one byte each in UTF-8")
    print()


def part_2_memory(workspace):
    print("[2] whole-file read vs line-by-line streaming")
    big = workspace / "big.log"
    size = toolkit.make_big_file(big, LINE_COUNT, long_line_at=LONG_LINE_AT)
    print(f"    generated {big.name}: {LINE_COUNT} lines, {size} bytes ({mib(size)})")

    whole_len, whole_peak = toolkit.longest_line_whole_file(big)
    stream_len, stream_peak = toolkit.longest_line_streaming(big)

    print(f"    read() + splitlines()  -> longest line {whole_len} chars, "
          f"peak memory {mib(whole_peak)}")
    print(f"    for line in handle     -> longest line {stream_len} chars, "
          f"peak memory {mib(stream_peak)}")
    print(f"    same answer: {whole_len == stream_len}")
    print(f"    streaming used {whole_peak / max(stream_peak, 1):.0f}x less memory")
    print()


def part_3_decoding(workspace):
    print("[3] decoding trouble and how to survive it")
    broken = workspace / "broken.log"
    # Written as RAW BYTES so we control exactly what lands on disk. 0xE9 is
    # "é" in latin-1, but on its own it is not valid UTF-8.
    broken.write_bytes(b"caf\xe9 was logged by a latin-1 program\n")
    print(f"    wrote {broken.name} as raw bytes: {broken.read_bytes()!r}")

    try:
        toolkit.read_text_file(broken)
        print("    strict utf-8 read: no error (unexpected)")
    except UnicodeDecodeError as err:
        print(f"    strict utf-8 read raised UnicodeDecodeError: {err}")

    text, recovered = toolkit.read_text_surviving_bad_bytes(broken)
    print(f"    errors='replace' read: {text.strip()!r}")
    print(f"    recovered: {recovered}; U+FFFD present: {chr(0xFFFD) in text}")
    print()


def part_4_atomic(workspace):
    print("[4] atomic write — proved by looking at the directory mid-operation")
    box = workspace / "config"
    box.mkdir(exist_ok=True)
    config = box / "config.txt"
    toolkit.write_text_file(config, "version = 1\n")
    print(f"    before: {sorted(p.name for p in box.iterdir())} "
          f"content={toolkit.read_text_file(config).strip()!r}")

    def peek(temp_path):
        during = sorted(p.name for p in box.iterdir())
        still_old = toolkit.read_text_file(config).strip()
        print(f"    during: {during} content={still_old!r}")
        print(f"            the new bytes are all in {temp_path.name}; readers of "
              f"{config.name} still get the OLD file, whole")

    toolkit.atomic_write_text(config, "version = 2\n", after_temp=peek)
    print(f"    after : {sorted(p.name for p in box.iterdir())} "
          f"content={toolkit.read_text_file(config).strip()!r}")

    # The contrast: the naive write, caught in the act.
    unsafe = box / "unsafe.txt"
    toolkit.write_text_file(unsafe, "version = 1\n")

    seen = {}

    def peek_unsafe(path):
        seen["mid"] = toolkit.read_text_file(path)

    toolkit.unsafe_write_text(unsafe, "version = 2 with more content\n", after_open=peek_unsafe)
    print(f"    contrast: mode 'w' straight onto the real file was caught mid-write "
          f"holding {seen['mid']!r}")
    print("              — a crash there destroys the old file and leaves that fragment")
    print()


def part_5_pathlib(workspace):
    print("[5] walking a directory tree with pathlib")
    tree = workspace / "tree"
    (tree / "sub" / "deeper").mkdir(parents=True, exist_ok=True)
    toolkit.write_text_file(tree / "a.txt", "a\n")
    toolkit.write_text_file(tree / "sub" / "b.txt", "b\n")
    toolkit.write_text_file(tree / "sub" / "deeper" / "c.txt", "c\n")

    found = toolkit.list_tree(tree)
    for relative in found:
        print(f"    {relative}")
    print(f"    {len(found)} file(s) found by Path.rglob")
    print(f"    tree exists: {tree.exists()}; is a directory: {tree.is_dir()}")
    print()


def main(argv):
    workspace = Path(argv[1] if len(argv) > 1 else "workspace")
    workspace.mkdir(parents=True, exist_ok=True)
    print(f"Safe File I/O Toolkit — workspace: {workspace}")
    print()
    part_1_round_trip(workspace)
    part_2_memory(workspace)
    part_3_decoding(workspace)
    part_4_atomic(workspace)
    part_5_pathlib(workspace)
    print(f"Done. Remove everything with: rm -rf {workspace}")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))

"""Safe file I/O toolkit — YOUR working file.

Five numbered exercises. Each unfinished function raises
`NotImplementedError` on purpose, so an empty function can never be
mistaken for a working one. Replace each `raise NotImplementedError(...)`
line with the real body described above it, keeping the docstring's promise
exactly — the docstring is the contract the tests check.

Work in order; run the driver after each exercise to see your progress:

    python3 starter/demo.py

and check your work at any time with:

    bash tests/run_tests.sh

The reference solution is in `examples/fileio_toolkit.py`. Use it when you
are stuck, not before — the point is the typing.
"""

import os
import tracemalloc
from pathlib import Path

# One padded record, so every generated line has the same length.
FILLER = "log line padding-padding-padding"


# --- Exercise 1 of 5: round-trip text with an EXPLICIT encoding -------------
#
# Fill in both functions.
#
# write_text_file: open `path` in mode "w" with `encoding=encoding` and
#   `newline="\n"`, inside a `with ... as handle:` block, and `handle.write`
#   the text. Then return `os.path.getsize(path)`.
#   Reminder: `write()` adds NO newline of its own — whatever you pass is
#   exactly what lands on disk.
#
# read_text_file: open `path` in mode "r" with `encoding=encoding`, inside a
#   `with` block, and return `handle.read()`.
#
# Both must name the encoding explicitly. Never rely on the platform default.

def write_text_file(path, text, encoding="utf-8"):
    """Write `text` to `path`, replacing anything already there.

    Returns the size of the finished file in bytes.
    """
    raise NotImplementedError("Exercise 1a: open in mode 'w' with an explicit encoding and write")


def read_text_file(path, encoding="utf-8"):
    """Read the whole of `path` back as one string, decoding with `encoding`."""
    raise NotImplementedError("Exercise 1b: open in mode 'r' with an explicit encoding and read")


# --- Provided complete: the file generator and the expensive reader ---------
# You do not edit these two. They exist so Exercise 2 has something big to
# read and something to be compared against.

def make_big_file(path, line_count, long_line_at=None):
    """Generate a large-ish log file: one numbered record per line.

    Every line has the same length except the one numbered `long_line_at`,
    which gets extra padding. Returns the size of the file in bytes.
    """
    with open(path, "w", encoding="utf-8", newline="\n") as handle:
        for number in range(1, line_count + 1):
            extra = "!" * 20 if number == long_line_at else ""
            handle.write(f"{number:07d} {FILLER}{extra}\n")
    return os.path.getsize(path)


def longest_line_whole_file(path):
    """Length of the longest line, read the EXPENSIVE way.

    Returns (longest_length, peak_bytes) — peak memory measured with
    `tracemalloc`, Python's built-in allocation tracker.
    """
    tracemalloc.start()
    with open(path, "r", encoding="utf-8") as handle:
        text = handle.read()
    lines = text.splitlines()
    longest = max((len(line) for line in lines), default=0)
    peak = tracemalloc.get_traced_memory()[1]
    tracemalloc.stop()
    return longest, peak


# --- Exercise 2 of 5: the same answer, one line at a time -------------------
#
# Give the SAME answer as longest_line_whole_file for a fraction of the
# memory. Between `tracemalloc.start()` and the peak reading:
#
#   1. set `longest = 0`
#   2. open `path` in mode "r" with `encoding="utf-8"` inside a `with` block
#   3. `for line in handle:` — iterating the file object hands you one line
#      at a time and lets the previous one be reclaimed
#   4. measure `len(line.rstrip("\n"))` and keep the largest
#
# Do NOT call `handle.read()` or `handle.readlines()` anywhere in here —
# both pull the whole file into memory and defeat the exercise.

def longest_line_streaming(path):
    """Length of the longest line, read the CHEAP way.

    Returns (longest_length, peak_bytes), the same answer as
    `longest_line_whole_file` for far less memory.
    """
    tracemalloc.start()
    raise NotImplementedError("Exercise 2: iterate the file object one line at a time")


# --- Exercise 3 of 5: survive bytes that are not valid UTF-8 ----------------
#
# 1. `try:` reading `path` strictly (a plain `with open(..., encoding=...)`
#    and `handle.read()`), and return `(text, False)` when it works.
# 2. `except UnicodeDecodeError:` open it again, this time adding
#    `errors="replace"`, read it, and return `(text, True)`.
#
# `errors="replace"` swaps each undecodable byte for the replacement
# character U+FFFD instead of raising. Strict-first is the right default:
# an error tells you the file is not what you assumed, and silence is worse.

def read_text_surviving_bad_bytes(path, encoding="utf-8"):
    """Read `path` as text, surviving bytes that are not valid `encoding`.

    Returns (text, recovered) where `recovered` is True when the
    `errors="replace"` fallback was needed.
    """
    raise NotImplementedError("Exercise 3: strict read first, then errors='replace' on failure")


# --- Exercise 4 of 5: the atomic write --------------------------------------
#
# Replace `path` with `text` so that no reader ever sees a half-written
# file. Four steps, in this order:
#
#   1. `path = Path(path)` and `temp = path.with_name(path.name + ".tmp")`
#      — the temp file must sit in the SAME directory, so the final rename
#      stays inside one filesystem and therefore stays atomic.
#   2. `with open(temp, "w", encoding=encoding, newline="\n") as handle:`
#      write the text, then `handle.flush()` (Python buffer -> operating
#      system) and `os.fsync(handle.fileno())` (operating system -> disk).
#   3. if `after_temp is not None:` call `after_temp(temp)` — that is the
#      moment the driver inspects the directory and finds both files.
#   4. `os.replace(temp, path)` — the atomic rename. Return `path`.

def atomic_write_text(path, text, encoding="utf-8", after_temp=None):
    """Replace `path` with `text` so no reader ever sees a half-written file.

    Returns the path that was written.
    """
    raise NotImplementedError("Exercise 4: temp file -> flush -> fsync -> os.replace")


def unsafe_write_text(path, text, encoding="utf-8", after_open=None):
    """The naive alternative, provided complete for contrast — do NOT copy it.

    Mode "w" truncates the real file immediately, so until the write
    finishes the file on disk is empty or partial. `after_open` is called
    while it is still incomplete, so the driver can look at the damage.
    """
    with open(path, "w", encoding=encoding, newline="\n") as handle:
        handle.write(text[: len(text) // 2])
        handle.flush()
        if after_open is not None:
            after_open(path)
        handle.write(text[len(text) // 2 :])
    return path


# --- Exercise 5 of 5: walk a tree with pathlib ------------------------------
#
# Return every FILE under `root` as sorted relative path strings.
#
#   1. `root = Path(root)`
#   2. `root.rglob("*")` yields every entry in the tree, recursively
#   3. keep only those where `p.is_file()` is true
#   4. turn each into `str(p.relative_to(root))`
#   5. wrap the whole thing in `sorted(...)`
#
# One generator expression inside `sorted(...)` does all of it.

def list_tree(root):
    """Every file under `root`, as sorted relative paths, using pathlib."""
    raise NotImplementedError("Exercise 5: sorted(str(p.relative_to(root)) for p in ...)")

#!/usr/bin/env bash
# Tests for the Day 064 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# These checks exercise real file behaviour, not file existence: text is
# written and read back byte for byte, a real large file is generated and
# read both ways with tracemalloc measuring both peaks, a genuinely
# undecodable byte is written and recovered from, the atomic writer is
# interrupted mid-operation and the directory inspected, and a real
# directory tree is walked with pathlib. Every check runs in a throwaway
# temporary directory that is removed afterwards. No network, no
# privileges, non-interactive. Exits 0 only if every check passes.
set -u

export PYTHONDONTWRITEBYTECODE=1

lab_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
failures=0
checks=0

scratch="$(mktemp -d -t day064-tests.XXXXXX)"
cleanup() { rm -rf "${scratch}"; }
trap cleanup EXIT

check() {
  local label="$1" ok="$2"
  checks=$((checks + 1))
  if [ "${ok}" = "yes" ]; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    failures=$((failures + 1))
  fi
}

# check_toolkit <label> <toolkit_dir> <python-body>
# Runs an assertion body against fileio_toolkit imported from toolkit_dir.
# A clean exit (every assert passed) is a pass.
check_toolkit() {
  local label="$1" toolkit_dir="$2" body="$3"
  local out
  if out="$(PYTHONPATH="${toolkit_dir}" SCRATCH="${scratch}" python3 -c "
import os
from pathlib import Path
import fileio_toolkit as t
scratch = Path(os.environ['SCRATCH'])
${body}" 2>&1)"; then
    check "${label}" "yes"
  else
    check "${label}" "no"
    echo "    (${out##*$'\n'})"
  fi
}

run_toolkit_checks() {
  local dir="$1"
  echo "Testing the toolkit in ${dir} (real files in a temporary directory) ..."

  check_toolkit "1. text round-trips through an explicit utf-8 encoding" "${dir}" "
p = scratch / 'round.txt'
text = 'line one\ncafé\n'
size = t.write_text_file(p, text)
assert t.read_text_file(p) == text, 'round-trip changed the text'
assert size == len(text.encode('utf-8')), f'size {size} is not the utf-8 byte count'
assert size > len(text), 'café must take more bytes than characters in utf-8'
"

  check_toolkit "1b. mode 'w' replaces rather than appends" "${dir}" "
p = scratch / 'replace.txt'
t.write_text_file(p, 'first\n')
t.write_text_file(p, 'second\n')
assert t.read_text_file(p) == 'second\n', 'mode w must truncate first'
"

  check_toolkit "1c. write() adds no newline of its own" "${dir}" "
p = scratch / 'nonewline.txt'
size = t.write_text_file(p, 'abc')
assert size == 3, f'expected 3 bytes, got {size}'
assert t.read_text_file(p) == 'abc'
"

  check_toolkit "2. streaming and whole-file reads agree on the longest line" "${dir}" "
p = scratch / 'big.log'
t.make_big_file(p, 20000, long_line_at=10000)
whole_len, whole_peak = t.longest_line_whole_file(p)
stream_len, stream_peak = t.longest_line_streaming(p)
assert whole_len == stream_len == 60, f'{whole_len} != {stream_len} != 60'
assert stream_peak > 0, 'tracemalloc peak should be measured, not zero'
"

  check_toolkit "2b. streaming peak memory is far below the whole-file peak" "${dir}" "
p = scratch / 'big.log'
t.make_big_file(p, 20000, long_line_at=10000)
_, whole_peak = t.longest_line_whole_file(p)
_, stream_peak = t.longest_line_streaming(p)
assert stream_peak * 4 < whole_peak, f'streaming {stream_peak} vs whole {whole_peak}'
"

  check_toolkit "3. a strict utf-8 read of a latin-1 byte raises UnicodeDecodeError" "${dir}" "
p = scratch / 'broken.log'
p.write_bytes(b'caf\xe9 logged\n')
try:
    t.read_text_file(p)
    raise SystemExit('strict read should have raised')
except UnicodeDecodeError as err:
    assert '0xe9' in str(err), str(err)
"

  check_toolkit "3b. errors='replace' recovers the file with U+FFFD" "${dir}" "
p = scratch / 'broken.log'
p.write_bytes(b'caf\xe9 logged\n')
text, recovered = t.read_text_surviving_bad_bytes(p)
assert recovered is True, 'the fallback should have been needed'
assert text == 'caf� logged\n', repr(text)
"

  check_toolkit "3c. a clean utf-8 file needs no fallback" "${dir}" "
p = scratch / 'clean.txt'
t.write_text_file(p, 'café\n')
text, recovered = t.read_text_surviving_bad_bytes(p)
assert recovered is False and text == 'café\n', (recovered, text)
"

  check_toolkit "4. atomic write leaves the OLD file intact until os.replace" "${dir}" "
box = scratch / 'atomic'
box.mkdir(exist_ok=True)
target = box / 'config.txt'
t.write_text_file(target, 'version = 1\n')
seen = {}
def peek(temp):
    seen['names'] = sorted(p.name for p in box.iterdir())
    seen['content'] = t.read_text_file(target)
    seen['temp'] = t.read_text_file(temp)
t.atomic_write_text(target, 'version = 2\n', after_temp=peek)
assert seen['names'] == ['config.txt', 'config.txt.tmp'], seen['names']
assert seen['content'] == 'version = 1\n', seen['content']
assert seen['temp'] == 'version = 2\n', seen['temp']
"

  check_toolkit "4b. after os.replace the new content is there and no .tmp remains" "${dir}" "
box = scratch / 'atomic2'
box.mkdir(exist_ok=True)
target = box / 'config.txt'
t.write_text_file(target, 'version = 1\n')
t.atomic_write_text(target, 'version = 2\n')
assert t.read_text_file(target) == 'version = 2\n'
assert sorted(p.name for p in box.iterdir()) == ['config.txt'], 'a .tmp file was left behind'
"

  check_toolkit "4c. the temp file sits in the SAME directory as the target" "${dir}" "
box = scratch / 'atomic3'
box.mkdir(exist_ok=True)
target = box / 'config.txt'
t.write_text_file(target, 'old\n')
seen = {}
t.atomic_write_text(target, 'new\n', after_temp=lambda temp: seen.update(parent=temp.parent))
assert seen['parent'] == box, f\"temp was in {seen['parent']}, not {box}\"
"

  check_toolkit "4d. the naive write is caught holding a half-written file" "${dir}" "
p = scratch / 'unsafe.txt'
t.write_text_file(p, 'version = 1\n')
seen = {}
t.unsafe_write_text(p, 'version = 2 with more content\n',
                    after_open=lambda path: seen.update(mid=t.read_text_file(path)))
assert seen['mid'] != 'version = 1\n', 'the old content should already be gone'
assert seen['mid'] != 'version = 2 with more content\n', 'it should still be incomplete'
"

  check_toolkit "5. list_tree finds every file as a sorted relative path" "${dir}" "
tree = scratch / 'tree'
(tree / 'sub' / 'deeper').mkdir(parents=True, exist_ok=True)
t.write_text_file(tree / 'a.txt', 'a\n')
t.write_text_file(tree / 'sub' / 'b.txt', 'b\n')
t.write_text_file(tree / 'sub' / 'deeper' / 'c.txt', 'c\n')
found = t.list_tree(tree)
assert found == ['a.txt', os.path.join('sub', 'b.txt'), os.path.join('sub', 'deeper', 'c.txt')], found
"

  check_toolkit "5b. list_tree returns files only, never directories" "${dir}" "
tree = scratch / 'tree'
found = t.list_tree(tree)
assert all(not (tree / f).is_dir() for f in found), found
assert 'sub' not in found, 'directories must be filtered out'
"
}

run_demo_checks() {
  local script="$1" label_prefix="$2"
  echo "Testing ${script} end to end ..."
  local out code demo_ws
  demo_ws="${scratch}/demo-$(basename "$(dirname "${script}")")"
  out="$(python3 "${script}" "${demo_ws}" 2>&1)"
  code=$?
  if [ "${code}" -eq 0 ]; then
    check "${label_prefix}: driver exits 0" "yes"
  else
    check "${label_prefix}: driver exits 0" "no"
    echo "    (exit ${code}; last line: ${out##*$'\n'})"
  fi
  for needle in \
    'round-trip identical: True' \
    'same answer: True' \
    'UnicodeDecodeError' \
    'recovered: True' \
    "config.txt', 'config.txt.tmp'" \
    '3 file(s) found by Path.rglob'
  do
    if printf '%s' "${out}" | grep -qF "${needle}"; then
      check "${label_prefix}: driver output contains \"${needle}\"" "yes"
    else
      check "${label_prefix}: driver output contains \"${needle}\"" "no"
    fi
  done
  rm -rf "${demo_ws}"
}

# --- Reference: always tested strictly ---
run_toolkit_checks "${lab_dir}/examples"
run_demo_checks "${lab_dir}/examples/demo.py" "examples"

# --- Learner starter ---
echo "Testing starter/fileio_toolkit.py ..."
starter_toolkit="${lab_dir}/starter/fileio_toolkit.py"
if python3 -c "compile(open('${starter_toolkit}').read(), '${starter_toolkit}', 'exec')" 2>/dev/null; then
  check "starter toolkit is valid Python" "yes"
else
  check "starter toolkit is valid Python" "no"
fi

if grep -q 'NotImplementedError' "${starter_toolkit}"; then
  echo "Note: starter/fileio_toolkit.py still has unfinished exercises — testing structure only."
  for name in write_text_file read_text_file longest_line_streaming \
              read_text_surviving_bad_bytes atomic_write_text list_tree; do
    if grep -q "def ${name}" "${starter_toolkit}"; then
      check "starter defines ${name}" "yes"
    else
      check "starter defines ${name}" "no"
    fi
  done
else
  run_toolkit_checks "${lab_dir}/starter"
  run_demo_checks "${lab_dir}/starter/demo.py" "starter"
  # A streaming reader that secretly slurps the file is not streaming.
  if grep -A 20 'def longest_line_streaming' "${starter_toolkit}" | grep -qE '\.read\(\)|\.readlines\(\)'; then
    check "starter longest_line_streaming avoids read()/readlines()" "no"
  else
    check "starter longest_line_streaming avoids read()/readlines()" "yes"
  fi
fi

echo
echo "${checks} checks, ${failures} failure(s)."
[ "${failures}" -eq 0 ]