tests.py

#!/usr/bin/env python3
"""
tests.py — Automated tests for the Build-your-own-Redis workshop.

Verifies that all 10 steps are properly structured:
  1. Required files exist in each step folder
  2. Python files have valid syntax (AST parse)
  3. README/GUIDE/EXERCISES files are non-empty and have expected sections
  4. Makefile has all required targets
  5. Top-level files exist
  6. Key concepts appear in the right step

Run: make test
  or: uv run python tests.py
"""
from __future__ import annotations

import ast
from pathlib import Path

WORKSHOP_DIR = Path(__file__).parent
PASS = "\033[32m✓\033[0m"
FAIL = "\033[31m✗\033[0m"

results = {"passed": 0, "failed": 0, "errors": []}


def check(description: str, condition: bool, detail: str = ""):
    if condition:
        results["passed"] += 1
        print(f"  {PASS} {description}")
    else:
        results["failed"] += 1
        msg = f"{description}: {detail}" if detail else description
        results["errors"].append(msg)
        print(f"  {FAIL} {description}" + (f" — {detail}" if detail else ""))


STEPS = [
    {"dir": "01-tcp-echo",          "main": "server.py"},
    {"dir": "02-resp-parser",       "main": "server.py"},
    {"dir": "03-get-set",           "main": "server.py"},
    {"dir": "04-expiry",            "main": "server.py"},
    {"dir": "05-aof-persistence",   "main": "server.py"},
    {"dir": "06-rdb-snapshots",     "main": "server.py"},
    {"dir": "07-pubsub",            "main": "server.py"},
    {"dir": "08-replication",       "main": "server.py"},
    {"dir": "09-event-loop",        "main": "server.py"},
    {"dir": "10-capstone",          "main": "bench.py"},
]


def test_top_level_files():
    print("\nTop-Level Files")
    for f in ["README.md", "Makefile", "pyproject.toml", ".gitignore", "tests.py"]:
        p = WORKSHOP_DIR / f
        check(f"{f} exists", p.exists())
        if p.exists():
            check(f"{f} is non-empty", p.stat().st_size > 0)


def test_step_structure():
    print("\nStep Structure")
    for s in STEPS:
        d = WORKSHOP_DIR / s["dir"]
        check(f"{s['dir']}/ exists", d.is_dir())
        if not d.is_dir():
            continue
        for f in [s["main"], "README.md", "GUIDE.md", "EXERCISES.md"]:
            check(f"{s['dir']}/{f} exists", (d / f).exists())


def test_python_syntax():
    print("\nPython Syntax")
    for s in STEPS:
        p = WORKSHOP_DIR / s["dir"] / s["main"]
        if not p.exists():
            continue
        try:
            ast.parse(p.read_text())
            check(f"{s['dir']}/{s['main']} compiles", True)
        except SyntaxError as e:
            check(f"{s['dir']}/{s['main']} compiles", False, str(e))


def test_doc_content():
    print("\nDocumentation Content")
    for s in STEPS:
        d = WORKSHOP_DIR / s["dir"]
        if not d.is_dir():
            continue
        readme = d / "README.md"
        if readme.exists():
            content = readme.read_text()
            check(f"{s['dir']}/README.md has title", content.startswith("#"))
            check(f"{s['dir']}/README.md substantial", len(content) > 200, f"only {len(content)} chars")
        guide = d / "GUIDE.md"
        if guide.exists():
            content = guide.read_text()
            check(f"{s['dir']}/GUIDE.md substantial", len(content) > 500, f"only {len(content)} chars")
        ex = d / "EXERCISES.md"
        if ex.exists():
            content = ex.read_text()
            n = content.count("## Exercise")
            check(f"{s['dir']}/EXERCISES.md has 3+ exercises", n >= 3, f"found {n}")
            cp = content.lower().count("checkpoint")
            check(f"{s['dir']}/EXERCISES.md has checkpoints", cp >= 2, f"found {cp}")


def test_makefile_targets():
    print("\nMakefile Targets")
    m = WORKSHOP_DIR / "Makefile"
    if not m.exists():
        check("Makefile exists", False); return
    content = m.read_text()
    required = ["install"] + [s["dir"] for s in STEPS] + ["test", "clean"]
    for t in required:
        check(f"Makefile has '{t}' target", f"{t}:" in content)


def test_pyproject():
    print("\nDependencies")
    p = WORKSHOP_DIR / "pyproject.toml"
    if not p.exists():
        check("pyproject.toml exists", False); return
    content = p.read_text()
    check("pyproject.toml has project name", "build-your-own-redis-python" in content)
    check("pyproject.toml requires Python 3.10+", ">=3.10" in content)


def test_key_concepts():
    print("\nKey Concepts in Code")
    checks = [
        ("01-tcp-echo",        "server.py", "SOCK_STREAM",     "TCP socket"),
        ("01-tcp-echo",        "server.py", "SO_REUSEADDR",    "Port reuse"),
        ("02-resp-parser",     "server.py", "parse_resp",      "RESP parser"),
        ("02-resp-parser",     "server.py", "CRLF",            "CRLF separator"),
        ("03-get-set",         "server.py", "COMMANDS",        "Dispatch table"),
        ("03-get-set",         "server.py", "STORE",           "In-memory store"),
        ("04-expiry",          "server.py", "EXPIRES",         "Expiry sidecar"),
        ("04-expiry",          "server.py", "TTL",             "TTL command"),
        ("05-aof-persistence", "server.py", "aof_append",      "AOF append"),
        ("05-aof-persistence", "server.py", "aof_replay",      "AOF replay"),
        ("06-rdb-snapshots",   "server.py", "rdb_save",        "RDB save"),
        ("06-rdb-snapshots",   "server.py", "os.fork",         "BGSAVE via fork"),
        ("07-pubsub",          "server.py", "SUBSCRIBERS",     "Subscriber registry"),
        ("07-pubsub",          "server.py", "PUBLISH",         "Publish command"),
        ("08-replication",     "server.py", "REPLICA_SOCKETS", "Replica sockets"),
        ("08-replication",     "server.py", "SYNC",            "SYNC command"),
        ("09-event-loop",      "server.py", "selectors",       "selectors import"),
        ("09-event-loop",      "server.py", "EVENT_READ",      "Event flags"),
        ("10-capstone",        "bench.py",  "run_phase",       "Benchmark phase"),
        ("10-capstone",        "bench.py",  "p99_ms",          "p99 latency"),
    ]
    for d, f, concept, label in checks:
        path = WORKSHOP_DIR / d / f
        if path.exists():
            content = path.read_text()
            check(f"{d} has {label}", concept in content)


if __name__ == "__main__":
    print("=" * 60)
    print("  Build-your-own-Redis Workshop — Automated Tests")
    print("=" * 60)

    test_top_level_files()
    test_step_structure()
    test_python_syntax()
    test_doc_content()
    test_makefile_targets()
    test_pyproject()
    test_key_concepts()

    total = results["passed"] + results["failed"]
    print(f"\n{'=' * 60}")
    print(f"  Results: {results['passed']}/{total} passed")
    if results["errors"]:
        print(f"\n  Failures:")
        for err in results["errors"]:
            print(f"    - {err}")
    print(f"{'=' * 60}")
    exit(0 if results["failed"] == 0 else 1)