Semana 14 — Internet y una mirada a Python

Objetivos

Seguir paso a paso lo que ocurre al abrir una página web.
Explicar qué protege el cifrado de HTTPS.
Leer un programa corto en Python y explicar en qué se diferencia del mismo programa en C.

Teoría — Internet y la web · Una mirada a Python

Temas:

Internet es una red de redes. Paquetes, routers y direcciones IP.
DNS, y HTTP: petición, respuesta y códigos de estado.
Qué protege HTTPS y qué no, sin matemática.
Lenguaje compilado frente a interpretado. Por qué existe más de un lenguaje.
El mismo programa en pseudocódigo, en C y en Python.
Idea clave: Python es cómodo porque alguien resolvió por debajo lo que ustedes ya vieron en C.

Laboratorio — Ver HTTP por dentro y comparar C con Python

Esta semana la sesión de laboratorio es para la Práctica 4. Las actividades de abajo son trabajo autónomo y no se entregan. Sus temas entran en el examen final.

Levantar el servidor de code/week-14/ y consultarlo con curl -v. Identificar la petición, los encabezados y el código de estado.
Pedir una ruta que no existe y observar el 404.
Correr el programa de primos de code/week-14/ en C y en Python. Comparar el código y el tiempo. El tiempo depende de la máquina.
Completar lab/http_status.c y comprobarlo con make test-lab.

Entregas

Práctica 4: struct, archivos, búsqueda y eficiencia. Ver practicas/README.md.

Preguntas de repaso

Ordenen lo que pasa al abrir https://www.uni.edu.pe: TLS, DNS, HTTP y TCP.
¿Qué significan los códigos 200, 404 y 500?
Con HTTPS, ¿qué puede ver quien administra la red del laboratorio, y qué no?
¿Qué error da C y qué error da Python al usar un índice fuera de rango? ¿Qué implica para quien programa?
¿Por qué un ciclo en Python suele ser más lento que en C?

Código de la semana

Compilar y probar dentro del contenedor del curso con make y make test.

Ejemplos (17)Errores (3)Pruebas (3)Laboratorio (1)Compilación (1)

Ejemplos

Los programas que usan los apuntes y la clase.

division.cDescargar

// La división en C depende del tipo de los operandos.
#include <stdio.h>

int main(void) {
    printf("%d\n", 7 / 2);
    printf("%.1f\n", 7.0 / 2);
    return 0;
}

division.pyDescargar

# En Python, / siempre da un número con decimales, y // es la división entera.
print(7 / 2)
print(7 // 2)

factorial.cDescargar

// Factorial con int. El número viene de la entrada para que UBSan lo vigile al correr.
#include <stdio.h>

int main(void) {
    int n;
    if (scanf("%d", &n) != 1) {
        printf("Entrada inválida\n");
        return 1;
    }
    int result = 1;
    for (int i = 2; i <= n; i++) {
        result = result * i;
    }
    printf("%d! = %d\n", n, result);
    return 0;
}

factorial.pyDescargar

# El mismo factorial. Aquí el entero crece todo lo que haga falta.
n = int(input())
result = 1
for i in range(2, n + 1):
    result = result * i
print(f"{n}! = {result}")

int_size.pyDescargar

# Cuántos bytes ocupa un entero de Python. Depende de la versión y de la máquina.
import sys

print(sys.getsizeof(1))
print(sys.getsizeof(2 ** 100))

limit.txtDescargar

primes.cDescargar

// Cuenta los primos hasta un límite que lee de la entrada.
#include <stdio.h>

int is_prime(int n) {
    if (n < 2) {
        return 0;
    }
    int d = 2;
    while (d <= n / d) {
        if (n % d == 0) {
            return 0;
        }
        d = d + 1;
    }
    return 1;
}

int main(void) {
    int limit;
    if (scanf("%d", &limit) != 1) {
        printf("Entrada inválida\n");
        return 1;
    }
    int count = 0;
    for (int n = 2; n <= limit; n++) {
        if (is_prime(n)) {
            count = count + 1;
        }
    }
    printf("Hay %d primos hasta %d\n", count, limit);
    return 0;
}

primes.pyDescargar

# Cuenta los primos hasta un límite que lee de la entrada.


def is_prime(n):
    if n < 2:
        return False
    d = 2
    while d <= n // d:
        if n % d == 0:
            return False
        d = d + 1
    return True


limit = int(input())
count = 0
for n in range(2, limit + 1):
    if is_prime(n):
        count = count + 1
print(f"Hay {count} primos hasta {limit}")

running_server.pyDescargar

"""Ayuda de las pruebas: levanta server.py o spy.py en un puerto libre y lo detiene al salir."""
import contextlib
import pathlib
import socket
import subprocess
import sys

HERE = pathlib.Path(__file__).parent


def free_port():
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        return probe.getsockname()[1]


@contextlib.contextmanager
def tool(*args):
    """Corre una herramienta hasta que anuncia que ya escucha; devuelve el proceso."""
    process = subprocess.Popen(
        [sys.executable, "-u", *args], cwd=HERE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
        text=True,
    )
    try:
        process.banner = process.stdout.readline()
        yield process
    finally:
        process.terminate()
        process.wait(timeout=10)
        process.stdout.close()


@contextlib.contextmanager
def web_server(*options):
    """Levanta server.py; devuelve el puerto."""
    port = free_port()
    with tool("server.py", *options, str(port)):
        yield port

selection_sort.cDescargar

// Ordena con selección, como en la semana 13. Crece como N al cuadrado.
#include <stdio.h>

#define MAX 100000

int main(void) {
    int values[MAX];
    int n;
    if (scanf("%d", &n) != 1 || n < 1 || n > MAX) {
        printf("Se espera un número entre 1 y %d\n", MAX);
        return 1;
    }
    for (int i = 0; i < n; i++) {
        values[i] = (i * 7919) % 100003;
    }
    for (int i = 0; i < n - 1; i++) {
        int smallest = i;
        for (int j = i + 1; j < n; j++) {
            if (values[j] < values[smallest]) {
                smallest = j;
            }
        }
        int saved = values[i];
        values[i] = values[smallest];
        values[smallest] = saved;
    }
    printf("menor: %d, mayor: %d\n", values[0], values[n - 1]);
    return 0;
}

server.pyDescargar

"""Servidor web de la semana 14. Atiende solo a esta computadora (127.0.0.1).

    python3 server.py           HTTP en el puerto 8014
    python3 server.py 9000      HTTP en otro puerto
    python3 server.py --tls     HTTPS en el puerto 8443; antes hay que correr make cert

Se detiene con Ctrl+C. No hace falta entender todo este archivo: es la herramienta del
laboratorio. Lo que importa es lo que responde.
"""
import http.server
import ssl
import sys

PAGES = {
    "/": ("text/html; charset=utf-8",
          "<!DOCTYPE html>\n<html>\n<body>\n<h1>Hola desde el servidor</h1>\n</body>\n</html>\n"),
    "/saludo": ("text/plain; charset=utf-8", "Buenos días, Perú\n"),
}
NOT_FOUND = "No existe esa página\n"
SERVER_ERROR = "Algo falló dentro del servidor\n"


class Handler(http.server.BaseHTTPRequestHandler):
    # Sin esta línea, la biblioteca responde con HTTP/1.0.
    protocol_version = "HTTP/1.1"

    def version_string(self):
        return "ServidorDelCurso/1.0"

    def do_GET(self):
        path = self.path.split("?")[0]
        if path in PAGES:
            content_type, text = PAGES[path]
            self.respond(200, content_type, text)
        elif path == "/error":
            # Un 500 a propósito, para verlo con curl.
            self.respond(500, "text/plain; charset=utf-8", SERVER_ERROR)
        else:
            self.respond(404, "text/plain; charset=utf-8", NOT_FOUND)

    def respond(self, code, content_type, text):
        body = text.encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", content_type)
        # Se cuentan bytes, no letras.
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


def main():
    args = sys.argv[1:]
    use_tls = "--tls" in args
    numbers = [arg for arg in args if arg != "--tls"]
    port = int(numbers[0]) if numbers else (8443 if use_tls else 8014)
    try:
        server = http.server.ThreadingHTTPServer(("127.0.0.1", port), Handler)
    except OSError as error:
        print(f"No se pudo abrir el puerto {port}: {error}")
        return 1
    scheme = "http"
    if use_tls:
        scheme = "https"
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.load_cert_chain("build/cert.pem", "build/key.pem")
        server.socket = context.wrap_socket(server.socket, server_side=True)
    host = "localhost" if use_tls else "127.0.0.1"
    print(f"Servidor en {scheme}://{host}:{port}/ (Ctrl+C para detener)", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nServidor detenido")
    return 0


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

show_bytecode.pyDescargar

# Muestra las instrucciones de la máquina virtual de Python para una función.
import dis


def add_one(x):
    return x + 1


dis.dis(add_one)

sort_library.pyDescargar

# Los mismos datos, con sorted: un algoritmo mejor, escrito en C dentro de Python.
n = int(input())
values = [(i * 7919) % 100003 for i in range(n)]
values = sorted(values)
print(f"menor: {values[0]}, mayor: {values[n - 1]}")

sort_size.txtDescargar

spy.pyDescargar

"""Un espía de la semana 14: se pone en medio de una conexión e imprime lo que pasa por ahí.

    python3 spy.py 8015 8014    escucha en 8015 y reenvía todo al puerto 8014

Hace en pequeño lo que puede hacer cualquier equipo por donde pasan sus paquetes. Solo sirve
para esta computadora. Lo que no se puede imprimir se muestra como un punto. Se detiene con Ctrl+C.
"""
import socket
import sys
import threading


def printable(data):
    return "".join(chr(b) if 32 <= b < 127 or b == 10 else "." for b in data)


def pump(source, target, label):
    """Copia bytes de una punta a la otra y los muestra."""
    while True:
        data = source.recv(4096)
        if not data:
            break
        print(f"--- {label}, {len(data)} bytes\n{printable(data)}", flush=True)
        target.sendall(data)
    target.shutdown(socket.SHUT_WR)


def attend(client, target_port):
    with client, socket.create_connection(("127.0.0.1", target_port)) as server:
        answer = threading.Thread(target=pump, args=(server, client, "servidor a cliente"))
        answer.start()
        pump(client, server, "cliente a servidor")
        answer.join()


def main():
    listen_port, target_port = int(sys.argv[1]), int(sys.argv[2])
    with socket.create_server(("127.0.0.1", listen_port)) as listener:
        print(f"Espía en el puerto {listen_port}, reenvía al {target_port}", flush=True)
        try:
            while True:
                client, _ = listener.accept()
                attend(client, target_port)
        except KeyboardInterrupt:
            print("\nEspía detenido")


if __name__ == "__main__":
    main()

text_length.cDescargar

// strlen cuenta bytes.
#include <stdio.h>
#include <string.h>

int main(void) {
    printf("%zu\n", strlen("año"));
    return 0;
}

text_length.pyDescargar

# len de un texto cuenta caracteres; encode da los bytes en UTF-8.
print(len("año"))
print(len("año".encode("utf-8")))

Errores

Ejemplos con errores a propósito: no compilan, fallan o dan un resultado incorrecto. Los apuntes explican cada uno.

errors/index_out.cDescargar

// Lee un índice y lo usa sin revisarlo. Con un índice fuera de rango, el comportamiento es indefinido.
#include <stdio.h>

int main(void) {
    int values[5] = {14, 9, 17, 11, 20};
    int index;
    if (scanf("%d", &index) != 1) {
        printf("Entrada inválida\n");
        return 1;
    }
    printf("%d\n", values[index]);
    return 0;
}

errors/index_out.pyDescargar

# El mismo programa. Python revisa el índice en cada acceso.
values = [14, 9, 17, 11, 20]
index = int(input())
print(values[index])

errors/late_error.pyDescargar

# La última línea tiene un error de tipos. Python solo lo descubre si llega a correrla.
n = int(input())
if n > 0:
    print("positivo")
else:
    print(n + "x")

Pruebas

Comprueban lo que afirman los apuntes; se ejecutan con make test.

test_examples.pyDescargar

"""Comprueba lo que los apuntes afirman sobre los programas de la semana 14."""
import pathlib
import socket
import subprocess
import sys
import tempfile
import unittest

from running_server import HERE, free_port, tool, web_server

BUILD = HERE / "build"
FLAGS = ["-Wall", "-Wextra", "-std=c17"]
SANITIZE = ["-g", "-fsanitize=address,undefined"]


def execute(command, stdin=""):
    done = subprocess.run(
        command, input=stdin, cwd=HERE, encoding="utf-8", errors="replace",
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=120,
    )
    return done.stdout, done.returncode


def run_c(program, stdin=""):
    return execute([str(BUILD / program)], stdin)


def run_python(script, stdin=""):
    return execute([sys.executable, script], stdin)


def build(source, *extra):
    out = pathlib.Path(tempfile.mkdtemp()) / pathlib.Path(source).stem
    messages, code = execute(["gcc", *FLAGS, *extra, source, "-o", str(out)])
    return messages, code, out


class LateError(unittest.TestCase):
    def test_python_only_finds_it_when_the_line_runs(self):
        self.assertEqual(run_python("errors/late_error.py", "5\n"), ("positivo\n", 0))
        output, code = run_python("errors/late_error.py", "-1\n")
        self.assertEqual(code, 1)
        self.assertIn("TypeError", output)


class CleanCode(unittest.TestCase):
    def test_c_has_no_warnings(self):
        for source in sorted(HERE.glob("*.c")) + sorted((HERE / "errors").glob("*.c")):
            with self.subTest(source=source.name):
                messages, code, _ = build(str(source.relative_to(HERE)))
                self.assertEqual((messages, code), ("", 0))


class PrimesInBothLanguages(unittest.TestCase):
    KNOWN = {-5: 0, 0: 0, 1: 0, 2: 1, 3: 2, 4: 2, 10: 4, 48: 15, 49: 15, 100: 25, 1000: 168,
             10000: 1229}

    def test_both_count_the_known_number_of_primes(self):
        for limit, count in self.KNOWN.items():
            expected = (f"Hay {count} primos hasta {limit}\n", 0)
            with self.subTest(limit=limit):
                self.assertEqual(run_c("primes", f"{limit}\n"), expected)
                self.assertEqual(run_python("primes.py", f"{limit}\n"), expected)

    def test_the_timing_input_gives_the_count_in_the_notes(self):
        self.assertEqual(run_c("primes_fast", (HERE / "limit.txt").read_text()),
                         ("Hay 78498 primos hasta 1000000\n", 0))

    def test_invalid_input(self):
        self.assertEqual(run_c("primes", "abc\n"), ("Entrada inválida\n", 1))
        output, code = run_python("primes.py", "abc\n")
        self.assertEqual(code, 1)
        self.assertIn("ValueError", output)


class Differences(unittest.TestCase):
    def test_division(self):
        self.assertEqual(run_c("division"), ("3\n3.5\n", 0))
        self.assertEqual(run_python("division.py"), ("3.5\n3\n", 0))

    def test_text_length(self):
        self.assertEqual(run_c("text_length"), ("4\n", 0))
        self.assertEqual(run_python("text_length.py"), ("3\n4\n", 0))

    def test_factorial_in_c_overflows_at_13(self):
        self.assertEqual(run_c("factorial", "12\n"), ("12! = 479001600\n", 0))
        output, _ = run_c("factorial", "13\n")
        self.assertIn("signed integer overflow", output)

    def test_factorial_in_python_is_exact(self):
        self.assertEqual(run_python("factorial.py", "13\n"), ("13! = 6227020800\n", 0))
        self.assertEqual(run_python("factorial.py", "25\n"),
                         ("25! = 15511210043330985984000000\n", 0))

    def test_python_limits_printing_very_long_integers(self):
        output, code = run_python("factorial.py", "2000\n")
        self.assertEqual(code, 1)
        self.assertIn("Exceeds the limit (4300 digits)", output)

    def test_index_out_of_range(self):
        _, _, checked = build("errors/index_out.c", *SANITIZE)
        output, _ = execute([str(checked)], "7\n")
        self.assertIn("index 7 out of bounds", output)
        self.assertIn("stack-buffer-overflow", output)
        _, _, unchecked = build("errors/index_out.c")
        self.assertEqual(execute([str(unchecked)], "2\n"), ("17\n", 0))
        output, code = run_python("errors/index_out.py", "7\n")
        self.assertEqual(code, 1)
        self.assertIn("IndexError: list index out of range", output)

    def test_negative_index_is_valid_in_python_only(self):
        self.assertEqual(run_python("errors/index_out.py", "-1\n"), ("20\n", 0))
        _, _, checked = build("errors/index_out.c", *SANITIZE)
        output, _ = execute([str(checked)], "-1\n")
        self.assertIn("index -1 out of bounds", output)

    def test_sort_programs_agree(self):
        for n in [1, 2, 1000, 5000]:
            with self.subTest(n=n):
                values = sorted((i * 7919) % 100003 for i in range(n))
                expected = (f"menor: {values[0]}, mayor: {values[-1]}\n", 0)
                self.assertEqual(run_c("selection_sort", f"{n}\n"), expected)
                self.assertEqual(run_python("sort_library.py", f"{n}\n"), expected)

    def test_sort_size_gives_the_result_in_the_notes(self):
        self.assertEqual(run_python("sort_library.py", (HERE / "sort_size.txt").read_text()),
                         ("menor: 0, mayor: 100001\n", 0))

    def test_bytecode_tool_runs(self):
        output, code = run_python("show_bytecode.py")
        self.assertEqual(code, 0)
        self.assertIn("BINARY_OP", output)

    def test_int_size_tool_runs(self):
        output, code = run_python("int_size.py")
        self.assertEqual(code, 0)
        small, big = [int(line) for line in output.split()]
        self.assertGreater(big, small)


def raw_request(port, text):
    """Manda bytes tal cual por una conexión y devuelve todo lo que responde el servidor."""
    with socket.create_connection(("127.0.0.1", port), timeout=10) as connection:
        connection.sendall(text.encode("ascii"))
        answer = b""
        while chunk := connection.recv(4096):
            answer += chunk
    return answer


def get(port, path):
    return raw_request(port, f"GET {path} HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n")


class Server(unittest.TestCase):
    def check(self, answer, status, body):
        head, _, actual_body = answer.partition(b"\r\n\r\n")
        lines = head.decode("ascii").split("\r\n")
        self.assertEqual(lines[0], status)
        self.assertEqual(actual_body.decode("utf-8"), body)
        self.assertIn(f"Content-Length: {len(body.encode('utf-8'))}", lines)

    def test_routes(self):
        with web_server() as port:
            self.check(get(port, "/saludo"), "HTTP/1.1 200 OK", "Buenos días, Perú\n")
            self.check(get(port, "/saludo?x=1"), "HTTP/1.1 200 OK", "Buenos días, Perú\n")
            self.check(get(port, "/nada"), "HTTP/1.1 404 Not Found", "No existe esa página\n")
            self.check(get(port, "/error"), "HTTP/1.1 500 Internal Server Error",
                       "Algo falló dentro del servidor\n")
            self.assertIn(b"<h1>Hola desde el servidor</h1>", get(port, "/"))

    def test_a_request_is_only_text(self):
        with web_server() as port:
            answer = raw_request(port, "GET /saludo HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n")
        self.assertTrue(answer.startswith(b"HTTP/1.1 200 OK\r\n"))

    def test_messages_at_start(self):
        port = free_port()
        with tool("server.py", str(port)) as server:
            self.assertEqual(server.banner,
                             f"Servidor en http://127.0.0.1:{port}/ (Ctrl+C para detener)\n")
        with web_server() as target:
            spy_port = free_port()
            with tool("spy.py", str(spy_port), str(target)) as spy:
                self.assertEqual(spy.banner, f"Espía en el puerto {spy_port}, reenvía al {target}\n")

    def test_busy_port_gives_a_message(self):
        with web_server() as port:
            output, code = execute([sys.executable, "server.py", str(port)])
        self.assertEqual(code, 1)
        self.assertIn(f"No se pudo abrir el puerto {port}", output)


class Eavesdropping(unittest.TestCase):
    """El espía ve la cookie en HTTP. En HTTPS ve el nombre, pero no la ruta ni la cookie."""

    def spy_on(self, url_scheme, target_port, host, *curl_options):
        spy_port = free_port()
        with tool("spy.py", str(spy_port), str(target_port)) as spy:
            answer, code = execute(
                ["curl", "-s", "-H", "Cookie: sesion=abc123", *curl_options,
                 f"{url_scheme}://{host}:{spy_port}/saludo"])
            self.assertEqual((answer, code), ("Buenos días, Perú\n", 0))
            spy.terminate()
            return spy.stdout.read()

    def test_http_shows_everything(self):
        with web_server() as port:
            seen = self.spy_on("http", port, "127.0.0.1")
        self.assertIn("GET /saludo HTTP/1.1", seen)
        self.assertIn("Cookie: sesion=abc123", seen)
        self.assertIn("HTTP/1.1 200 OK", seen)

    def test_https_hides_path_and_cookie(self):
        done = subprocess.run(["make", "--no-print-directory", "cert"], cwd=HERE,
                              capture_output=True, text=True)
        self.assertEqual(done.returncode, 0, done.stderr)
        with web_server("--tls") as port:
            seen = self.spy_on("https", port, "localhost", "--cacert", "build/cert.pem")
        self.assertIn("localhost", seen)
        for secret in ["/saludo", "sesion", "abc123", "Buenos", "HTTP/1.1 200"]:
            with self.subTest(secret=secret):
                self.assertNotIn(secret, seen)

    def test_without_trusting_the_certificate_curl_refuses(self):
        done = subprocess.run(["make", "--no-print-directory", "cert"], cwd=HERE, capture_output=True)
        self.assertEqual(done.returncode, 0)
        with web_server("--tls") as port:
            output, code = execute(["curl", "-s", "-S", f"https://localhost:{port}/saludo"])
        self.assertEqual(code, 60)
        self.assertIn("self-signed certificate", output)



if __name__ == "__main__":
    unittest.main()

test_lab.pyDescargar

"""Pruebas del laboratorio. LAB_DIR dice dónde está el código: lab o solution."""
import os
import shutil
import subprocess
import sys
import tempfile
import unittest

from running_server import HERE, web_server

LAB_DIR = os.environ.get("LAB_DIR", "solution")


def run(stdin):
    done = subprocess.run(
        [str(HERE / "build" / LAB_DIR / "http_status")], input=stdin.encode("utf-8"),
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=10,
    )
    return done.stdout.decode("utf-8", "replace"), done.returncode


def summary(code, headers, body_bytes):
    return (f"código: {code}\nclase: {code // 100}xx\nencabezados: {headers}\n"
            f"cuerpo: {body_bytes} bytes\n", 0)


NOT_HTTP = ("No es una respuesta HTTP\n", 1)


@unittest.skipUnless((HERE / LAB_DIR).is_dir(), f"no existe {LAB_DIR}/")
class StatusLine(unittest.TestCase):
    def test_no_warnings(self):
        done = subprocess.run(
            ["gcc", "-Wall", "-Wextra", "-std=c17", "-Werror", "-fsyntax-only",
             str(HERE / LAB_DIR / "http_status.c")],
            text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
        )
        self.assertEqual(done.returncode, 0, done.stdout)

    def test_valid_codes(self):
        cases = {"HTTP/1.1 200 OK": 200, "HTTP/1.1 404 Not Found": 404,
                 "HTTP/1.0 500 Internal Server Error": 500, "HTTP/2 301 ": 301,
                 "HTTP/3 200": 200, "HTTP/1.1 100 Continue": 100, "HTTP/1.1 599 x": 599}
        for line, code in cases.items():
            with self.subTest(line=line):
                self.assertEqual(run(line + "\r\n\r\n"), summary(code, 0, 0))

    def test_codes_out_of_range(self):
        for line in ["HTTP/1.1 99 x", "HTTP/1.1 600 x", "HTTP/1.1 0 x", "HTTP/1.1 -200 x"]:
            with self.subTest(line=line):
                self.assertEqual(run(line + "\r\n\r\n"), NOT_HTTP)

    def test_not_a_status_line(self):
        for line in ["hola", "hola 200", "GET / HTTP/1.1", "HTTP 200 OK", "http/1.1 200 OK",
                     "HTTP/1.1 abc", "HTTP/1.1", "", "200"]:
            with self.subTest(line=line):
                self.assertEqual(run(line + "\r\n\r\n"), NOT_HTTP)

    def test_long_version_does_not_overflow(self):
        self.assertEqual(run("HTTP/" + "1" * 200 + "x 200 OK\r\n\r\n"), NOT_HTTP)

    @unittest.skipUnless(shutil.which("valgrind"), "no hay valgrind")
    def test_sscanf_result_is_checked(self):
        # Sin revisar lo que devuelve sscanf, se usa un code que nadie asignó. Su valor puede caer
        # fuera de rango por casualidad, así que se le pregunta a valgrind y no a la salida.
        with tempfile.TemporaryDirectory() as folder:
            program = os.path.join(folder, "http_status")
            subprocess.run(["gcc", "-std=c17", "-g", "-O0", str(HERE / LAB_DIR / "http_status.c"),
                            "-o", program], check=True, capture_output=True)
            for line in ["HTTP/1.1 abc\r\n\r\n", "HTTP/1.1\r\n\r\n", "hola\r\n\r\n"]:
                with self.subTest(line=line):
                    done = subprocess.run(
                        ["valgrind", "-q", "--error-exitcode=99", program], input=line.encode(),
                        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=60,
                    )
                    self.assertNotEqual(done.returncode, 99, done.stdout.decode("utf-8", "replace"))

    def test_empty_input(self):
        self.assertEqual(run(""), ("La entrada está vacía\n", 1))


@unittest.skipUnless((HERE / LAB_DIR).is_dir(), f"no existe {LAB_DIR}/")
class Parts(unittest.TestCase):
    def test_headers_and_body(self):
        self.assertEqual(run("HTTP/1.1 200 OK\r\nA: 1\r\nB: 2\r\n\r\nhola\r\nmundo\n"),
                         summary(200, 2, 12))

    def test_example_in_the_notes(self):
        self.assertEqual(run("HTTP/1.1 404 Not Found\r\nA: 1\r\n\r\nhola\n"), summary(404, 1, 5))

    def test_no_body(self):
        self.assertEqual(run("HTTP/1.1 404 Not Found\r\nA: 1\r\n\r\n"), summary(404, 1, 0))

    def test_no_headers(self):
        self.assertEqual(run("HTTP/1.1 200 OK\r\n\r\nabc"), summary(200, 0, 3))

    def test_blank_line_inside_the_body_is_body(self):
        self.assertEqual(run("HTTP/1.1 200 OK\r\nA: 1\r\n\r\nx\r\n\r\ny\n"), summary(200, 1, 7))

    def test_header_like_line_in_the_body_is_body(self):
        self.assertEqual(run("HTTP/1.1 200 OK\r\nA: 1\r\n\r\nB: 2\r\n"), summary(200, 1, 6))

    def test_lines_ending_in_newline_only(self):
        self.assertEqual(run("HTTP/1.1 500 x\nA: 1\nB: 2\nC: 3\n\nz"), summary(500, 3, 1))

    def test_no_blank_line_at_all(self):
        # Sin línea vacía, todo lo que hay son encabezados y el cuerpo queda en 0.
        self.assertEqual(run("HTTP/1.1 200 OK\r\nA: 1\r\nB: 2\r\n"), summary(200, 2, 0))

    def test_body_longer_than_the_buffer(self):
        for size in [255, 256, 1000, 5000]:
            with self.subTest(size=size):
                body = "x" * (size - 1) + "\n"
                self.assertEqual(run("HTTP/1.1 200 OK\r\nA: 1\r\n\r\n" + body),
                                 summary(200, 1, size))

    def test_multibyte_body_counts_bytes(self):
        self.assertEqual(run("HTTP/1.1 200 OK\r\n\r\nBuenos días, Perú\n"), summary(200, 0, 20))


@unittest.skipUnless((HERE / LAB_DIR).is_dir(), f"no existe {LAB_DIR}/")
class WithTheServer(unittest.TestCase):
    def fetch(self, port, path):
        done = subprocess.run(["curl", "-si", f"http://127.0.0.1:{port}{path}"],
                              stdout=subprocess.PIPE, timeout=10)
        return done.stdout.decode("utf-8")

    def test_real_responses(self):
        # El servidor manda cuatro encabezados: Server, Date, Content-Type y Content-Length.
        with web_server() as port:
            self.assertEqual(run(self.fetch(port, "/saludo")), summary(200, 4, 20))
            self.assertEqual(run(self.fetch(port, "/nada")), summary(404, 4, 22))
            self.assertEqual(run(self.fetch(port, "/error")), summary(500, 4, 32))


class GivenPrograms(unittest.TestCase):
    """Python no se evalúa: solo se comprueba que lo que les damos corre."""

    def test_primes_in_c_and_python_agree(self):
        c_program = subprocess.run([str(HERE / "build" / "primes")], input="1000\n",
                                   capture_output=True, text=True, timeout=30)
        python = subprocess.run([sys.executable, str(HERE / "primes.py")], input="1000\n",
                                capture_output=True, text=True, timeout=30)
        self.assertEqual(c_program.stdout, "Hay 168 primos hasta 1000\n")
        self.assertEqual(python.stdout, c_program.stdout)


if __name__ == "__main__":
    unittest.main()

test_notes.pyDescargar

"""Vuelve a correr los comandos que aparecen en los apuntes y en las diapositivas
y comprueba que el código citado existe de verdad en esta carpeta."""
import pathlib
import re
import subprocess
import unittest

from running_server import HERE, web_server

ROOT = HERE.parent.parent
NOTES = ROOT / "notes" / "week-14.md"
SLIDES = ROOT / "slides" / "week-14.tex"
# Antes de un bloque con fechas, direcciones o rutas. Se corre, pero solo se exigen las líneas estables.
MACHINE_DEPENDENT = "<!-- depende de la corrida -->"
# Antes de un bloque que no se puede repetir aquí: necesita internet, otra terminal o da tiempos.
# Lo que afirman esos bloques se comprueba en test_examples.py y test_lab.py.
NOT_RUN = "<!-- se muestra, no se corre -->"
STABLE = ["AddressSanitizer: stack-buffer-overflow", "index 7 out of bounds", "IndexError:",
          "TypeError:", "signed integer overflow", "BINARY_OP", "positivo", "localhost"]
PORT = "8014"


def normalize(text):
    text = text.replace("‘", "'").replace("’", "'")
    text = re.sub(r"^((?:< )?[Dd]ate:) .*$", r"\1", text, flags=re.M)
    return re.sub(r"0x[0-9a-f]+", "0x", text)


def run_shell(command, port):
    done = subprocess.run(
        ["bash", "-c", command.replace(PORT, str(port))], cwd=HERE, encoding="utf-8",
        errors="replace", stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT, timeout=120,
    )
    return normalize(done.stdout.replace(str(port), PORT))


def split_commands(lines):
    """Parte un bloque de terminal en pares (comando, líneas de salida)."""
    pairs = []
    for line in lines:
        if line.startswith("$ "):
            pairs.append((line[2:], []))
        elif pairs:
            pairs[-1][1].append(normalize(line.rstrip()))
    return pairs


def note_blocks(language):
    """Devuelve (líneas, marca) de cada bloque de los apuntes."""
    blocks, current, mark, previous = [], None, "", ""
    for raw in NOTES.read_text(encoding="utf-8").splitlines():
        line = re.sub(r"^> ?", "", raw)
        if current is None:
            if line.strip() == "```" + language:
                current, mark = [], previous.strip()
            elif line.strip():
                previous = line
        elif line.strip() == "```":
            blocks.append((current, mark))
            current = None
            previous = ""
        else:
            current.append(line)
    return blocks


def slide_blocks(environments):
    text = SLIDES.read_text(encoding="utf-8")
    pattern = r"\\begin\{(?:%s)\}[^\n]*\n(.*?)\\end\{(?:%s)\}" % (environments, environments)
    return [match.group(1).splitlines() for match in re.finditer(pattern, text, re.S)]


def known_code_lines():
    lines = set()
    sources = []
    for pattern in ["*.c", "*.py"]:
        sources += list(HERE.glob(pattern)) + list((HERE / "errors").glob(pattern))
    for source in sources:
        lines.update(line.strip() for line in source.read_text(encoding="utf-8").splitlines())
    return lines


class NotesAreTrue(unittest.TestCase):
    @unittest.skipUnless(NOTES.exists(), "faltan los apuntes")
    def test_terminal_blocks_in_notes(self):
        checked = 0
        with web_server() as port:
            for lines, mark in note_blocks("text"):
                if not lines or not lines[0].startswith("$ ") or mark == NOT_RUN:
                    continue
                for command, expected in split_commands(lines):
                    actual = run_shell(command, port).rstrip("\n")
                    checked += 1
                    with self.subTest(command=command):
                        if mark == MACHINE_DEPENDENT:
                            self.check_stable(expected, actual)
                        else:
                            lines_got = [line.rstrip() for line in actual.split("\n")]
                            self.assertEqual("\n".join(lines_got), "\n".join(expected))
        self.assertGreater(checked, 12)

    def check_stable(self, expected, actual):
        for line in expected:
            is_http = re.match(r"[<>] \S", line) and not re.match(r"[<>] (User-Agent|Date)", line)
            if is_http:
                self.assertIn(line, [got.rstrip() for got in actual.split("\n")])
            for token in STABLE:
                if token in line:
                    self.assertIn(token, actual)

    @unittest.skipUnless(NOTES.exists(), "faltan los apuntes")
    def test_code_blocks_in_notes_exist(self):
        known = known_code_lines()
        for language in ["c", "python"]:
            for lines, _ in note_blocks(language):
                for line in lines:
                    if line.strip() and line.strip() != "...":
                        with self.subTest(line=line):
                            self.assertIn(line.strip(), known)

    @unittest.skipUnless(NOTES.exists(), "faltan los apuntes")
    def test_notes_flag_what_depends_on_the_machine_or_the_network(self):
        text = " ".join(NOTES.read_text(encoding="utf-8").split())
        for phrase in ["dependen de la máquina", "depende de la red", "2 de noviembre de 2026",
                       "en esta corrida", "sin sanitizers"]:
            with self.subTest(phrase=phrase):
                self.assertIn(phrase, text.replace("**", ""))


class SlidesAreTrue(unittest.TestCase):
    @unittest.skipUnless(SLIDES.exists(), "faltan las diapositivas")
    def test_terminal_blocks_in_slides(self):
        with web_server() as port:
            for block in slide_blocks("terminal|terminalchica"):
                for command, expected in split_commands(block):
                    actual = run_shell(command, port).splitlines()
                    position = 0
                    # Una diapositiva puede omitir líneas, pero no cambiarlas ni reordenarlas.
                    for line in expected:
                        line = line.strip()
                        if not line:
                            continue
                        with self.subTest(command=command, line=line):
                            found = [i for i in range(position, len(actual))
                                     if actual[i].strip() == line]
                            self.assertTrue(found, f"no aparece en la salida real: {line}")
                            if found:
                                position = found[0] + 1

    @unittest.skipUnless(SLIDES.exists(), "faltan las diapositivas")
    def test_code_blocks_in_slides_exist(self):
        known = known_code_lines()
        for block in slide_blocks("codigo|codigochico|codigomini"):
            for line in block:
                if line.strip() and line.strip() != "...":
                    with self.subTest(line=line):
                        self.assertIn(line.strip(), known)


if __name__ == "__main__":
    unittest.main()

Laboratorio

Archivos de partida del laboratorio de la semana.

lab/http_status.cDescargar

// Lee una respuesta HTTP de la entrada y resume sus partes.
//   curl -si http://127.0.0.1:8014/saludo | build/lab/http_status
//
// Completen las partes marcadas con TAREA. Los apuntes explican cada una.
#include <stdio.h>
#include <string.h>

#define LINE_SIZE 256

// Devuelve el código de una línea como "HTTP/1.1 404 Not Found", o -1 si no tiene esa forma.
// Tiene esa forma si empieza con HTTP/ y una versión, y le sigue un número entre 100 y 599.
int parse_status(const char line[]) {
    // TAREA: sepárenla con sscanf y revisen lo que devuelve. Pongan límite a la versión.
    (void) line;
    return -1;
}

// Devuelve 1 si la línea está vacía. HTTP termina las líneas con \r\n; se acepta también \n.
int is_blank_line(const char line[]) {
    // TAREA: comparen con strcmp.
    (void) line;
    return 0;
}

int main(void) {
    char line[LINE_SIZE];
    if (fgets(line, LINE_SIZE, stdin) == NULL) {
        printf("La entrada está vacía\n");
        return 1;
    }
    int code = parse_status(line);
    if (code == -1) {
        printf("No es una respuesta HTTP\n");
        return 1;
    }

    // TAREA: cuenten los encabezados. Terminan en la primera línea vacía, que no se cuenta.
    int headers = 0;

    // TAREA: sumen los bytes del cuerpo, que es todo lo que queda.
    int body_bytes = 0;

    printf("código: %d\n", code);
    printf("clase: %dxx\n", code / 100);
    printf("encabezados: %d\n", headers);
    printf("cuerpo: %d bytes\n", body_bytes);
    return 0;
}

Compilación

Cómo se compilan y ejecutan los ejemplos.

MakefileDescargar

# Semana 14: internet y una mirada a Python. Se usa dentro del entorno del curso.
#
#   make           compila los programas de la clase en build/
#   make test      corre las pruebas de los ejemplos y de los apuntes
#   make test-lab  prueba su laboratorio (lab/http_status.c); falla hasta que lo terminen
#   make cert      crea un certificado de prueba para python3 server.py --tls
#   make clean     borra lo generado
#
# build/primes y build/selection_sort llevan sanitizers, y por eso son lentos. Para medir
# tiempos están build/primes_fast y build/selection_sort_fast: -O2 y sin sanitizers.
# Los programas de errors/ se compilan a mano, con los comandos de los apuntes.

CC = gcc
CFLAGS = -Wall -Wextra -std=c17 -g -fsanitize=address,undefined
FAST = -Wall -Wextra -std=c17 -O2
BUILD = build
PROGRAMS = primes division text_length factorial selection_sort

.PHONY: all lab test test-lab cert clean

all: $(PROGRAMS:%=$(BUILD)/%) $(BUILD)/primes_fast $(BUILD)/selection_sort_fast

$(BUILD)/%: %.c | $(BUILD)
	$(CC) $(CFLAGS) $< -o $@

$(BUILD)/%_fast: %.c | $(BUILD)
	$(CC) $(FAST) $< -o $@

$(BUILD) $(BUILD)/lab $(BUILD)/solution:
	mkdir -p $@

$(BUILD)/lab/% : lab/%.c | $(BUILD)/lab
	$(CC) $(CFLAGS) $< -o $@

$(BUILD)/solution/% : solution/%.c | $(BUILD)/solution
	$(CC) $(CFLAGS) $< -o $@

lab: $(BUILD)/lab/http_status

# El certificado solo sirve para las pruebas en esta computadora.
cert: | $(BUILD)
	openssl req -x509 -newkey rsa:2048 -nodes -days 30 -subj "/CN=localhost" \
	  -addext "subjectAltName=DNS:localhost" \
	  -keyout $(BUILD)/key.pem -out $(BUILD)/cert.pem 2> /dev/null

# Si existe solution/, también se comprueba la solución del laboratorio.
test: all
	@if [ -d solution ]; then $(MAKE) --no-print-directory $(BUILD)/solution/http_status; fi
	python3 -m unittest discover -p 'test_*.py'

test-lab: all lab
	LAB_DIR=lab python3 -m unittest -v test_lab

clean:
	rm -rf $(BUILD) __pycache__