sqlite3 and parameterised queries

import sqlite3


def find_by_author(con: sqlite3.Connection, author: str) -> list[str]:
    """Titles by exactly this author, in title order."""
    sql = f"SELECT title FROM book WHERE author = '{author}' ORDER BY title"
    return [row[0] for row in con.execute(sql)]


def search_titles(con: sqlite3.Connection, word: str) -> list[str]:
    """Titles that contain word anywhere, in title order."""
    sql = f"SELECT title FROM book WHERE title LIKE '%{word}%' ORDER BY title"
    return [row[0] for row in con.execute(sql)]

import sqlite3

from main import find_by_author, search_titles


def make_db():
    con = sqlite3.connect(":memory:")
    con.execute("CREATE TABLE book(title TEXT, author TEXT)")
    con.executemany(
        "INSERT INTO book VALUES (?, ?)",
        [
            ("Emma", "Jane Austen"),
            ("Persuasion", "Jane Austen"),
            ("Dune", "Frank Herbert"),
            ("At Swim-Two-Birds", "Flann O'Brien"),
        ],
    )
    return con


def test_author():
    """find_by_author finds both Austen titles"""
    got = find_by_author(make_db(), "Jane Austen")
    assert got == ["Emma", "Persuasion"], f"find_by_author gave {got!r}"


def test_apostrophe():
    """A name with an apostrophe works"""
    got = find_by_author(make_db(), "Flann O'Brien")
    assert got == ["At Swim-Two-Birds"], f"find_by_author gave {got!r}"


def test_author_injection():
    """An injection attempt is just a name that matches nothing"""
    got = find_by_author(make_db(), "x' OR '1'='1")
    assert got == [], f"find_by_author returned {got!r}: bind the value with ?"


def test_search():
    """search_titles finds titles containing the word"""
    got = search_titles(make_db(), "u")
    assert got == ["Dune", "Persuasion"], f"search_titles gave {got!r}"


def test_search_injection():
    """Quotes in the search word cannot change the query"""
    got = search_titles(make_db(), "' OR 1=1 OR title = '")
    assert got == [], f"search_titles returned {got!r}: put the % signs into the bound value"

import sqlite3


def transfer(con: sqlite3.Connection, src: str, dst: str, amount: int) -> None:
    """Move amount from src to dst: both updates happen, or neither does."""
    # Nothing is committed, and nothing stops an overdraft or an unknown account.
    con.execute("UPDATE account SET balance = balance - ? WHERE name = ?", (amount, src))
    con.execute("UPDATE account SET balance = balance + ? WHERE name = ?", (amount, dst))

import sqlite3

from main import transfer


def make_db():
    con = sqlite3.connect(":memory:")
    con.execute("CREATE TABLE account(name TEXT PRIMARY KEY, balance INTEGER)")
    con.executemany("INSERT INTO account VALUES (?, ?)", [("ada", 100), ("bob", 50)])
    con.commit()
    return con


def balances(con):
    return dict(con.execute("SELECT name, balance FROM account"))


def raises_value_error(con, *args):
    try:
        transfer(con, *args)
    except ValueError:
        return True
    return False


def test_moves_money():
    """30 moves from ada to bob"""
    con = make_db()
    transfer(con, "ada", "bob", 30)
    got = balances(con)
    assert got == {"ada": 70, "bob": 80}, f"balances are {got!r}"


def test_committed():
    """The transfer is committed"""
    con = make_db()
    transfer(con, "ada", "bob", 30)
    assert not con.in_transaction, "a transaction is still open: use with con: so it is committed"


def test_overdraft():
    """Too little money raises ValueError and changes nothing"""
    con = make_db()
    raised = raises_value_error(con, "ada", "bob", 500)
    got = balances(con)
    assert raised and got == {"ada": 100, "bob": 50}, f"raised ValueError: {raised}, balances {got!r}"


def test_unknown_account():
    """An unknown receiver raises ValueError and ada keeps her money"""
    con = make_db()
    raised = raises_value_error(con, "ada", "eve", 10)
    got = balances(con)
    assert raised and got == {"ada": 100, "bob": 50}, f"raised ValueError: {raised}, balances {got!r}: roll back the first update"