troupe-standup/wizard-work/spark/test_line_counter.py

#!/usr/bin/env python3
"""The cell for wizard-work/spark/line_counter.py (Q125).

WHAT THIS PINS
  Every cell runs the shipped tool as a subprocess and compares its whole
  stdout, so a number that moves for any reason is a red cell, not a rounding
  difference in a review.  The two cells the queue item names by hand are
  `blank_lines_are_lines` (a file with blank lines must read 4, not 2) and
  `final_unterminated_fragment_is_a_record` (a file whose last row has no
  newline must read lines 3 / records 4, not line count == record count).

  The last two cells are not tests of the tool, they are tests of *this file*:
  each copies the module to a scratch directory, applies one named mutation
  with an assertion that the edit landed, and re-runs this suite against the
  copy expecting rc 1 and the expected cell among the failures.  A suite that
  survives those mutants is a suite that cannot fail, which is the defect the
  queue item was opened over -- so the mutants are cells, not a one-off reading
  in a report nobody re-runs.

USAGE
  python3 wizard-work/spark/test_line_counter.py            run everything
  python3 wizard-work/spark/test_line_counter.py --no-mutants   inner run
  LINE_COUNTER=/path/to/line_counter.py python3 test_line_counter.py
        grade a different copy of the tool (how the mutation cells do it)

EXIT CODES
  0  every cell passed
  1  at least one cell failed
  2  the suite could not grade: the module or a fixture it needs is missing
"""

import json
import os
import subprocess
import sys
import tempfile

HERE = os.path.dirname(os.path.abspath(__file__))
TOOL = os.environ.get("LINE_COUNTER") or os.path.join(HERE, "line_counter.py")
SELF = os.path.abspath(__file__)

CELLS = []


def cell(fn):
    CELLS.append(fn)
    return fn


def expect(condition, message):
    if not condition:
        raise AssertionError(message)


def run(*args):
    return subprocess.run(
        [sys.executable, TOOL, *args], capture_output=True, text=True
    )


def exact(*args, expected):
    result = run(*args)
    expect(
        result.returncode == 0,
        f"exit {result.returncode}, stderr {result.stderr.strip()!r}",
    )
    expect(
        result.stdout == expected,
        f"stdout {result.stdout!r} != expected {expected!r}",
    )


@cell
def blank_lines_are_lines():
    """`printf 'a\\n\\n\\nb\\n'` reads 4 lines and 4 records, not 2."""
    with tempfile.TemporaryDirectory() as tmp:
        path = os.path.join(tmp, "blank.py")
        with open(path, "wb") as handle:
            handle.write(b"a\n\n\nb\n")
        exact(
            path,
            expected=(
                "files: 1\nlines: 4\nrecords: 4\n"
                "blank: 2\ncomment: 0\ncode: 2\n"
            ),
        )


@cell
def final_unterminated_fragment_is_a_record():
    """Without a trailing newline, lines and records must differ by one."""
    with tempfile.TemporaryDirectory() as tmp:
        path = os.path.join(tmp, "noterm.py")
        with open(path, "wb") as handle:
            handle.write(b"a\n\n\nb")
        exact(
            path,
            expected=(
                "files: 1\nlines: 3\nrecords: 4\n"
                "blank: 2\ncomment: 0\ncode: 2\n"
            ),
        )


@cell
def committed_log_fixture():
    """test_log.txt: 10 rows of text, 9 newlines -- lines 9, records 10."""
    path = os.path.join(HERE, "test_log.txt")
    expect(os.path.isfile(path), f"fixture missing: {path}")
    exact(
        path,
        expected=(
            "files: 1\nlines: 9\nrecords: 10\n"
            "blank: -\ncomment: -\ncode: -\n"
        ),
    )


@cell
def committed_csv_fixture_fields():
    """test_data.csv holds five comma-separated items."""
    path = os.path.join(HERE, "test_data.csv")
    expect(os.path.isfile(path), f"fixture missing: {path}")
    exact("--fields", path, expected="fields: 5\n")


@cell
def quoted_separator_is_data():
    """`"a,b",c` is 2 fields; str.split(',') reads 3."""
    with tempfile.TemporaryDirectory() as tmp:
        path = os.path.join(tmp, "quoted.csv")
        with open(path, "w", encoding="utf-8") as handle:
            handle.write('"a,b",c\n')
        exact("--fields", path, expected="fields: 2\n")
        expect(
            len('"a,b",c\n'.split("\n")[0].split(",")) == 3,
            "the contrast this cell rests on no longer holds",
        )


@cell
def directory_walk_sums_and_breaks_down():
    """Three files, two of them Python: sums and the lexical breakdown."""
    with tempfile.TemporaryDirectory() as tmp:
        nested = os.path.join(tmp, "sub")
        os.makedirs(nested)
        with open(os.path.join(tmp, "a.py"), "wb") as handle:
            handle.write(b"x = 1\n# c\n\n")
        with open(os.path.join(nested, "b.txt"), "wb") as handle:
            handle.write(b"hello\n")
        with open(os.path.join(nested, "c.py"), "wb") as handle:
            handle.write(b"y = 2")
        exact(
            tmp,
            expected=(
                "files: 3\nlines: 4\nrecords: 5\n"
                "blank: 1\ncomment: 1\ncode: 2\n"
            ),
        )


@cell
def non_python_target_prints_dashes():
    """A target with no .py file must print '-' rather than an uncounted 0."""
    with tempfile.TemporaryDirectory() as tmp:
        with open(os.path.join(tmp, "only.txt"), "wb") as handle:
            handle.write(b"one\ntwo\n")
        exact(
            tmp,
            expected=(
                "files: 1\nlines: 2\nrecords: 2\n"
                "blank: -\ncomment: -\ncode: -\n"
            ),
        )


@cell
def json_matches_the_text_report():
    """--json carries the same figures, nulls where the text prints '-'."""
    with tempfile.TemporaryDirectory() as tmp:
        path = os.path.join(tmp, "blank.py")
        with open(path, "wb") as handle:
            handle.write(b"a\n\n\nb\n")
        result = run("--json", path)
        expect(result.returncode == 0, f"--json exited {result.returncode}")
        data = json.loads(result.stdout)
        expect(
            data
            == {
                "path": path,
                "files": 1,
                "lines": 4,
                "records": 4,
                "blank": 2,
                "comment": 0,
                "code": 2,
            },
            f"--json payload differs: {data}",
        )
        text = run(path)
        expect(
            text.stdout.splitlines()
            == [
                "files: 1",
                "lines: 4",
                "records: 4",
                "blank: 2",
                "comment: 0",
                "code: 2",
            ],
            f"text report differs: {text.stdout!r}",
        )


@cell
def missing_target_exits_two():
    """An unreadable target is rc 2 with the path named, never a partial total."""
    with tempfile.TemporaryDirectory() as tmp:
        missing = os.path.join(tmp, "nope.txt")
        result = run(missing)
        expect(result.returncode == 2, f"exit {result.returncode}, want 2")
        expect(result.stdout == "", f"stdout {result.stdout!r}, want empty")
        expect(missing in result.stderr, f"path not named: {result.stderr!r}")


@cell
def usage_error_exits_one():
    """No target at all is a usage error, not an empty report."""
    result = run()
    expect(result.returncode == 1, f"exit {result.returncode}, want 1")
    expect("usage:" in result.stderr, f"no usage line: {result.stderr!r}")


@cell
def word_and_char_counts():
    """--words: three whitespace tokens with a repeat, and six characters."""
    with tempfile.TemporaryDirectory() as tmp:
        path = os.path.join(tmp, "words.txt")
        with open(path, "w", encoding="utf-8") as handle:
            handle.write("a b a\n")
        exact("--words", path, expected="files: 1\nwords: 3\nchars: 6\n")


@cell
def fields_on_a_directory_exits_two():
    """--fields needs one record, so a directory is a refusal."""
    with tempfile.TemporaryDirectory() as tmp:
        result = run("--fields", tmp)
        expect(result.returncode == 2, f"exit {result.returncode}, want 2")
        expect("--fields" in result.stderr, f"no reason: {result.stderr!r}")


MUTANTS = [
    (
        "count_returns_ten",
        "    lines = data.count(b\"\\n\")\n"
        "    records = lines + (1 if data and not data.endswith(b\"\\n\") else 0)\n"
        "    return lines, records",
        "    return 10, 10",
        "blank_lines_are_lines",
    ),
    (
        "count_non_blank_lines_only",
        "    lines = data.count(b\"\\n\")\n"
        "    records = lines + (1 if data and not data.endswith(b\"\\n\") else 0)\n"
        "    return lines, records",
        "    nonblank = sum(1 for _r in data.split(b\"\\n\") if _r.strip())\n"
        "    return nonblank, nonblank",
        "blank_lines_are_lines",
    ),
    (
        "records_equal_lines",
        "    records = lines + (1 if data and not data.endswith(b\"\\n\") else 0)",
        "    records = lines",
        "final_unterminated_fragment_is_a_record",
    ),
    (
        "fields_by_str_split",
        "    return len(next(csv.reader(io.StringIO(first))))",
        "    return len(first.split(\",\"))",
        "quoted_separator_is_data",
    ),
    (
        "words_by_unique",
        "        words += len(text.split())",
        "        words += len(set(text.split()))",
        "word_and_char_counts",
    ),
]


@cell
def every_mutant_is_killed():
    """Each named mutation must turn this suite red on the cell that names it."""
    if "--no-mutants" in sys.argv:
        return
    with open(TOOL, "r", encoding="utf-8") as handle:
        original = handle.read()
    for name, old, new, expected_cell in MUTANTS:
        expect(old in original, f"mutant {name}: anchor not found in {TOOL}")
        with tempfile.TemporaryDirectory() as tmp:
            mutant = os.path.join(tmp, "line_counter.py")
            with open(mutant, "w", encoding="utf-8") as handle:
                handle.write(original.replace(old, new, 1))
            with open(mutant, "r", encoding="utf-8") as handle:
                expect(
                    old not in handle.read(),
                    f"mutant {name}: the edit did not land",
                )
            env = dict(os.environ, LINE_COUNTER=mutant)
            inner = subprocess.run(
                [sys.executable, SELF, "--no-mutants"],
                capture_output=True,
                text=True,
                env=env,
            )
            expect(
                inner.returncode == 1,
                f"mutant {name} survived: rc {inner.returncode}\n{inner.stdout}",
            )
            expect(
                f"FAIL: {expected_cell}" in inner.stdout,
                f"mutant {name} was caught, but not by {expected_cell}:\n"
                f"{inner.stdout}",
            )
    print(
        f"       ({len(MUTANTS)} mutants applied to scratch copies, "
        f"{len(MUTANTS)} killed)"
    )


def main(argv):
    if not os.path.isfile(TOOL):
        print(f"CANNOT GRADE: tool missing: {TOOL}", file=sys.stderr)
        return 2
    failures = 0
    for fn in CELLS:
        name = fn.__name__
        try:
            fn()
        except AssertionError as exc:
            failures += 1
            print(f"FAIL: {name} -- {exc}")
        else:
            print(f"PASS: {name}")
    print(f"\n{len(CELLS) - failures}/{len(CELLS)} cells passed")
    print("ALL CELLS PASSED" if not failures else f"{failures} CELL(S) FAILED")
    return 0 if not failures else 1


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