MySQL Protocol Reference

Interface declaration:

db = service("mysql",
    interface("main", "mysql", 3306),
    image = "mysql:8",
    env = {"MYSQL_ROOT_PASSWORD": "test", "MYSQL_DATABASE": "testdb"},
    healthcheck = ready(timeout = "120s"),
)

Credentials come from env= — MYSQL_USER/MYSQL_PASSWORD when the spec creates a named user, otherwise root with MYSQL_ROOT_PASSWORD — and MYSQL_DATABASE selects the default schema. Steps, the healthcheck and ready() all use them. An explicit user=/password=/database= on a step wins; dsn= bypasses the lot.

Fixed in v0.16.1. Before this, the generated DSN was a bare root@tcp(host:port)/ — no password, no database. Against a stock mysql:8 every step failed with Access denied for user ‘root’ (using password: NO), which reached the spec as the far less legible invalid connection; against a passwordless server, statements failed with Error 1046: No database selected. If your MySQL steps have never worked, this is why.

Methods

query(sql="", user="", password="", database="", dsn="")

Execute a SQL query that returns rows.

resp = db.main.query(sql="SELECT * FROM users WHERE id=1")
# resp.data = [{"id": 1, "name": "alice"}]

resp = db.main.query(sql="SELECT count(*) as n FROM orders")
# resp.data = [{"n": 42}]

exec(sql="", user="", password="", database="", dsn="")

Execute a SQL statement that doesn’t return rows.

resp = db.main.exec(sql="INSERT INTO users (name) VALUES ('bob')")
# resp.data = {"rows_affected": 1}

resp = db.main.exec(sql="CREATE TABLE IF NOT EXISTS orders (id INT AUTO_INCREMENT PRIMARY KEY, item VARCHAR(255))")

Response Object

Same as Postgres — query() returns list of dicts, exec() returns {"rows_affected": N}.

Fault Rules

error(query=, message=)

Reject matching queries with a MySQL error.

insert_fail = fault_assumption("insert_fail",
    target = db.main,
    rules = [error(query="INSERT*", message="disk full")],
)

read_only = fault_assumption("read_only",
    target = db.main,
    rules = [error(query="INSERT*", message="read only"),
             error(query="UPDATE*", message="read only"),
             error(query="DELETE*", message="read only")],
)

delay(query=, delay=)

slow_reads = fault_assumption("slow_reads",
    target = db.main,
    rules = [delay(query="SELECT*", delay="2s")],
)

drop(query=)

drop_writes = fault_assumption("drop_writes",
    target = db.main,
    rules = [drop(query="INSERT*")],
)

Seed / Reset Patterns

Seed and reset run before your assertions do, so a silent failure there becomes a confusing failure later — a test that reports a missing row rather than a database that never got seeded. Check them.

def must(sql):
    r = db.main.exec(sql=sql)
    assert_true(r.ok, "seed step failed (%s): %s" % (sql[:40], r.error))

def seed_mysql():
    must("CREATE TABLE IF NOT EXISTS users (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255))")
    must("CREATE TABLE IF NOT EXISTS orders (id INT AUTO_INCREMENT PRIMARY KEY, user_id INT, item VARCHAR(255))")
    must("INSERT INTO users (name) VALUES ('alice'), ('bob')")

def reset_mysql():
    must("SET FOREIGN_KEY_CHECKS=0")
    must("TRUNCATE TABLE orders")
    must("TRUNCATE TABLE users")
    must("SET FOREIGN_KEY_CHECKS=1")
    must("INSERT INTO users (name) VALUES ('alice'), ('bob')")

db = service("mysql",
    interface("main", "mysql", 3306),
    image = "mysql:8",
    env = {"MYSQL_ROOT_PASSWORD": "test", "MYSQL_DATABASE": "testdb"},
    healthcheck = ready(timeout = "120s"),
    reuse = True,
    seed = seed_mysql,
    reset = reset_mysql,
)

Tip: MySQL requires SET FOREIGN_KEY_CHECKS=0 before truncating tables with foreign key constraints.

Event Sources

No native event source. Use observe.stdout to capture MySQL logs:

db = service("mysql", ...,
    observe = [observe.stdout(decoder=decoder("logfmt"))],
)