Phase 3: run statements with COM_QUERY

lib/mysqlex/client.ex

@@ -9,7 +9,7 @@ defmodule Mysqlex.Client do
 
   alias Mysqlex.Auth.NativePassword
   alias Mysqlex.Packet
-  alias Mysqlex.Protocol.{Codec, Err, Handshake, HandshakeResponse}
+  alias Mysqlex.Protocol.{Codec, Err, Handshake, HandshakeResponse, Ok}
 
   defstruct [:transport, :socket, :capabilities, :connection_id, :server_version, buffer: ""]
 
@@ -70,6 +70,43 @@ defmodule Mysqlex.Client do
     end
   end
 
+  @doc """
+  Runs a statement and reads the server's answer.
+
+  Every command starts a fresh sequence, so this writes as sequence id 0 and
+  expects the reply as 1.
+  """
+  def query(%__MODULE__{} = client, statement, timeout \\ @default_timeout) do
+    with :ok <- send_message(client.transport, client.socket, [0x03, statement], 0),
+         {:ok, _sequence_id, payload, buffer} <-
+           recv_message(client.transport, client.socket, client.buffer, timeout) do
+      query_response(payload, %{client | buffer: buffer})
+    end
+  end
+
+  # The first byte of a COM_QUERY response decides everything: 0x00 is an OK for
+  # a statement that returned no rows, 0xFF an error, 0xFB the server asking the
+  # client to upload a file, and anything else is a length-encoded column count
+  # introducing a resultset.
+  defp query_response(<<0x00, _rest::binary>> = payload, client) do
+    {:ok, ok} = Ok.decode(payload, client.capabilities)
+
+    {:ok, ok, client}
+  end
+
+  defp query_response(<<0xFF, _rest::binary>> = payload, client) do
+    {:ok, error} = Err.decode(payload, client.capabilities)
+
+    {:error, error, client}
+  end
+
+  # CLIENT_LOCAL_FILES is not claimed, so a server asking for a file is either
+  # misbehaving or hostile. Refuse rather than read from the filesystem.
+  defp query_response(<<0xFB, _rest::binary>>, client),
+    do: {:error, :local_infile_not_supported, client}
+
+  defp query_response(_payload, client), do: {:error, :resultset_not_implemented, client}
+
   @doc """
   Reads one message, reading again for as long as the buffer holds less than a
   whole message.

test/integration/query_test.exs

@@ -0,0 +1,69 @@
+defmodule Mysqlex.QueryIntegrationTest do
+  use ExUnit.Case, async: false
+
+  @moduletag :integration
+
+  alias Mysqlex.{Client, Error, Tcp}
+
+  # see test/integration/login_test.exs for how to start the server
+  @opts [
+    hostname: "127.0.0.1",
+    port: 3307,
+    username: "mysqlex",
+    password: "secret",
+    database: "mysqlex_test"
+  ]
+
+  setup do
+    {:ok, client} = Client.connect(@opts)
+    on_exit(fn -> Tcp.close(client.socket) end)
+
+    {:ok, _ok, client} = Client.query(client, "DROP TABLE IF EXISTS widgets")
+
+    {:ok, _ok, client} =
+      Client.query(
+        client,
+        "CREATE TABLE widgets (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(20))"
+      )
+
+    %{client: client}
+  end
+
+  test "reports affected rows and the generated key for an insert", %{client: client} do
+    assert {:ok, first, client} =
+             Client.query(client, "INSERT INTO widgets (name) VALUES ('first')")
+
+    assert first.affected_rows == 1
+    assert first.last_insert_id == 1
+
+    assert {:ok, second, _client} =
+             Client.query(client, "INSERT INTO widgets (name) VALUES ('second')")
+
+    assert second.affected_rows == 1
+    assert second.last_insert_id == 2
+  end
+
+  test "reports rows matched for an update", %{client: client} do
+    {:ok, _ok, client} = Client.query(client, "INSERT INTO widgets (name) VALUES ('first')")
+
+    assert {:ok, update, _client} =
+             Client.query(client, "UPDATE widgets SET name = 'renamed' WHERE id = 1")
+
+    assert update.affected_rows == 1
+    assert update.info =~ "Rows matched: 1"
+  end
+
+  test "returns the server's error for a missing table", %{client: client} do
+    assert {:error, %Error{} = error, _client} =
+             Client.query(client, "SELECT * FROM definitely_missing")
+
+    assert error.code == 1146
+    assert error.sql_state == "42S02"
+    assert error.message =~ "doesn't exist"
+  end
+
+  test "refuses a resultset until it can parse one", %{client: client} do
+    assert {:error, :resultset_not_implemented, _client} =
+             Client.query(client, "SELECT * FROM widgets")
+  end
+end

test/mysqlex/client_query_test.exs

@@ -0,0 +1,56 @@
+defmodule Mysqlex.ClientQueryTest do
+  use ExUnit.Case, async: true
+
+  alias Mysqlex.{Client, Error, FakeTransport, Packet}
+  alias Mysqlex.Protocol.{HandshakeResponse, Ok}
+
+  defp framed(payload, sequence_id), do: IO.iodata_to_binary(Packet.encode(payload, sequence_id))
+
+  defp client(chunks) do
+    socket = FakeTransport.script(chunks)
+
+    %Client{
+      transport: FakeTransport,
+      socket: socket,
+      capabilities: HandshakeResponse.capabilities("mysqlex_test")
+    }
+  end
+
+  test "sends the statement as command 0x03, starting a new sequence at 0" do
+    ok = <<0x00, 1, 42, 0x02, 0x00, 0x00, 0x00>>
+    client = client([framed(ok, 1)])
+
+    assert {:ok, %Ok{}, _client} = Client.query(client, "INSERT INTO t VALUES (1)")
+
+    assert FakeTransport.sent(client.socket) ==
+             framed(<<0x03, "INSERT INTO t VALUES (1)">>, 0)
+  end
+
+  test "returns what the server reports about a write" do
+    ok = <<0x00, 1, 42, 0x02, 0x00, 0x00, 0x00>>
+
+    assert {:ok, result, _client} =
+             Client.query(client([framed(ok, 1)]), "INSERT INTO t VALUES (1)")
+
+    assert result.affected_rows == 1
+    assert result.last_insert_id == 42
+  end
+
+  test "returns the server's error for a bad statement" do
+    err = <<0xFF, 0x7A, 0x04, ?#, "42S02", "Table 'mysqlex_test.nope' doesn't exist">>
+
+    assert {:error, %Error{} = error, _client} =
+             Client.query(client([framed(err, 1)]), "SELECT 1")
+
+    assert error.code == 1146
+    assert error.sql_state == "42S02"
+  end
+
+  test "does not pretend to understand a resultset yet" do
+    # a payload whose first byte is neither 0x00 nor 0xFF is a column count
+    column_count = <<0x01>>
+
+    assert {:error, :resultset_not_implemented, _client} =
+             Client.query(client([framed(column_count, 1)]), "SELECT 1")
+  end
+end