// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

// MySQL client TRANSACTION integration tests, run through brpc's PUBLIC API
// against a REAL mysqld.
//
// Each TEST_F drives a transaction scenario end to end:
//   * brpc::Channel(protocol="mysql", connection_type="pooled",
//     auth=MysqlAuthenticator) -> a live connection to the server;
//   * brpc::NewMysqlTransaction(channel, opts) -> a connection-pinned
//     transaction handle (START TRANSACTION on a dedicated socket);
//   * MysqlRequest(tx).Query(...) + channel.CallMethod(...) -> statements
//     INSIDE the transaction (same pinned socket);
//   * MysqlRequest().Query(...) on the SAME channel -> a SECOND connection
//     from the pool, used as an independent observer to prove isolation
//     (uncommitted rows are invisible until commit());
//   * tx->commit() / tx->rollback() -> terminate the transaction.
//
// Because transactions, simple SELECTs and DML all flow through the same
// COM_QUERY text protocol, these tests also cover simple-query execution
// and text-result parsing (column metadata + row field decoding).
//
// Harness (server spawn / skip convention, -mysql_use_running_server and
// -mysql_host/-port/-user/-password gflags) follows
// test/mysql/brpc_mysql_auth_handshake_unittest.cpp and, transitively,
// test/brpc_redis_unittest.cpp's which-then-spawn pattern.  When mysqld is
// absent every test GTEST_SKIP()s, so the file is CI-safe with no server.

#include <gflags/gflags.h>
#include <gtest/gtest.h>

#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>

#include <cstdio>
#include <cstdlib>
#include <pthread.h>
#include <string>

#include "brpc/channel.h"
#include "brpc/policy/mysql/mysql.h"
#include "brpc/policy/mysql/mysql_transaction.h"
#include "brpc/policy/mysql/mysql_authenticator.h"
#include "butil/logging.h"

// These gflags are intentionally re-declared here (not shared with the auth
// unittest): the CMake glob at test/CMakeLists.txt builds each
// brpc_mysql_*_unittest.cpp into its OWN executable, so there is no symbol
// collision across test binaries.
DEFINE_bool(mysql_use_running_server, false,
            "Use an already-running MySQL server instead of spawning a "
            "throwaway one; the running server is neither started nor stopped "
            "by the test.");
DEFINE_string(mysql_host, "127.0.0.1",
              "Host of the running MySQL server "
              "(only with -mysql_use_running_server).");
DEFINE_int32(mysql_port, 13306,
             "TCP port of the MySQL server (used for both the running server "
             "and the spawned throwaway server).");
DEFINE_string(mysql_user, "root", "Login user for the transaction tests.");
DEFINE_string(mysql_password, "",
              "Password for -mysql_user (empty for the spawned server).");
DEFINE_string(mysql_schema, "brpc_txn_test",
              "Schema (database) the transaction tests create and use.");

namespace {

// Throwaway-server harness (mirrors brpc_mysql_auth_handshake_unittest.cpp,
// which mirrors brpc_redis_unittest.cpp).  >0: forked pid; -2: external
// running server reachable; -1: no server -> tests skip.
#define MYSQLD_BIN "mysqld"

static pthread_once_t s_start_once = PTHREAD_ONCE_INIT;
static pid_t s_mysqld_pid = -1;
static std::string s_host = "127.0.0.1";
static int s_port = 13306;
static std::string s_user = "root";
static std::string s_password;

static std::string TestDataDir() {
    char cwd[1024];
    if (getcwd(cwd, sizeof(cwd)) == NULL) {
        return std::string("/tmp/mysql_txn_data_for_test");
    }
    return std::string(cwd) + "/mysql_txn_data_for_test";
}

static void RemoveMysqlServer() {
    if (s_mysqld_pid > 0) {
        puts("[Stopping mysqld]");
        char cmd[1280];
        snprintf(cmd, sizeof(cmd), "kill %d", s_mysqld_pid);
        CHECK(0 == system(cmd));
        usleep(500000);
        snprintf(cmd, sizeof(cmd), "rm -rf '%s'", TestDataDir().c_str());
        CHECK(0 == system(cmd));
    }
}

// Raw TCP probe to detect server readiness; returns fd (caller closes) or -1.
static int ProbeMysql() {
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0) {
        return -1;
    }
    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(static_cast<uint16_t>(s_port));
    addr.sin_addr.s_addr = inet_addr(s_host.c_str());
    if (connect(fd, (struct sockaddr*)&addr, sizeof(addr)) != 0) {
        close(fd);
        return -1;
    }
    return fd;
}

static void RunMysqlServer() {
    if (FLAGS_mysql_use_running_server) {
        s_host = FLAGS_mysql_host;
        s_port = FLAGS_mysql_port;
        s_user = FLAGS_mysql_user;
        s_password = FLAGS_mysql_password;
        printf("[Using running mysqld at %s:%d as user '%s']\n",
               s_host.c_str(), s_port, s_user.c_str());
        int fd = ProbeMysql();
        if (fd >= 0) {
            close(fd);
            s_mysqld_pid = -2;
        } else {
            printf("Cannot reach running mysqld at %s:%d, tests will skip\n",
                   s_host.c_str(), s_port);
        }
        return;
    }

    if (system("which " MYSQLD_BIN) != 0) {
        puts("Fail to find " MYSQLD_BIN ", transaction tests will be skipped");
        return;
    }
    s_host = "127.0.0.1";
    s_port = FLAGS_mysql_port;
    s_user = "root";
    s_password.clear();
    const std::string datadir = TestDataDir();
    char cmd[2048];
    snprintf(cmd, sizeof(cmd), "rm -rf '%s' && mkdir -p '%s'",
             datadir.c_str(), datadir.c_str());
    if (system(cmd) != 0) {
        puts("Fail to create datadir, transaction tests will be skipped");
        return;
    }
    snprintf(cmd, sizeof(cmd),
             MYSQLD_BIN " --initialize-insecure --datadir='%s'"
             " --log-error='%s/init.err'",
             datadir.c_str(), datadir.c_str());
    if (system(cmd) != 0) {
        puts("Fail to initialize mysqld datadir, tests will be skipped");
        snprintf(cmd, sizeof(cmd), "rm -rf '%s'", datadir.c_str());
        CHECK(0 == system(cmd));
        return;
    }
    atexit(RemoveMysqlServer);

    s_mysqld_pid = fork();
    if (s_mysqld_pid < 0) {
        puts("Fail to fork");
        exit(1);
    } else if (s_mysqld_pid == 0) {
        puts("[Starting mysqld]");
        char port_arg[32];
        snprintf(port_arg, sizeof(port_arg), "--port=%d", FLAGS_mysql_port);
        const std::string datadir_arg = "--datadir=" + datadir;
        const std::string socket_arg = "--socket=" + datadir + "/mysqld.sock";
        const std::string pidfile_arg = "--pid-file=" + datadir + "/mysqld.pid";
        const std::string logerr_arg = "--log-error=" + datadir + "/mysqld.err";
        char* const argv[] = {
            (char*)MYSQLD_BIN,
            (char*)datadir_arg.c_str(),
            (char*)port_arg,
            (char*)socket_arg.c_str(),
            (char*)pidfile_arg.c_str(),
            (char*)logerr_arg.c_str(),
            (char*)"--mysqlx=OFF",
            (char*)"--bind-address=127.0.0.1",
            NULL};
        if (execvp(MYSQLD_BIN, argv) < 0) {
            puts("Fail to run " MYSQLD_BIN);
            exit(1);
        }
    }
    // Wait for TCP readiness (fresh tablespace recovery), then create a
    // password account so the caching_sha2 client can authenticate over TCP
    // exactly like the running-server mode.  root keeps its empty password on
    // the unix socket; we exercise the spawned server as empty-password root.
    for (int i = 0; i < 300; ++i) {
        int fd = ProbeMysql();
        if (fd >= 0) {
            close(fd);
            return;
        }
        usleep(100000);
    }
    puts("mysqld did not become ready, transaction tests will be skipped");
    s_mysqld_pid = -1;
}

// Small helpers over the brpc MySQL public API.

// Runs |sql| outside any transaction on |channel| (a fresh pooled
// connection).  Returns false on transport failure.
static bool RunPlain(brpc::Channel& channel, const std::string& sql,
                     brpc::MysqlResponse* resp) {
    brpc::MysqlRequest req;
    if (!req.Query(sql)) {
        return false;
    }
    brpc::Controller cntl;
    channel.CallMethod(NULL, &cntl, &req, resp, NULL);
    return !cntl.Failed();
}

// Runs |sql| INSIDE transaction |tx| (its pinned connection).  Returns false
// on transport failure.
static bool RunInTx(brpc::Channel& channel, const brpc::MysqlTransaction* tx,
                    const std::string& sql, brpc::MysqlResponse* resp) {
    brpc::MysqlRequest req(tx);
    if (!req.Query(sql)) {
        return false;
    }
    brpc::Controller cntl;
    channel.CallMethod(NULL, &cntl, &req, resp, NULL);
    return !cntl.Failed();
}

// Convenience: expects a single OK reply for a DML/DDL statement.
static ::testing::AssertionResult ExpectOk(const brpc::MysqlResponse& resp) {
    if (resp.reply_size() < 1) {
        return ::testing::AssertionFailure() << "no reply";
    }
    const brpc::MysqlReply& r = resp.reply(0);
    if (r.is_error()) {
        return ::testing::AssertionFailure()
               << "ERR " << r.error().errcode() << ": "
               << r.error().msg().as_string();
    }
    if (!r.is_ok()) {
        return ::testing::AssertionFailure() << "reply is not OK, type=" << r.type();
    }
    return ::testing::AssertionSuccess();
}

// Returns the row count of the FIRST reply, asserting it is a result set.
// On any non-resultset reply returns -1 (so callers can fail clearly).
static int64_t ResultRowCount(const brpc::MysqlResponse& resp) {
    if (resp.reply_size() < 1) {
        return -1;
    }
    const brpc::MysqlReply& r = resp.reply(0);
    if (!r.is_resultset()) {
        return -1;
    }
    return static_cast<int64_t>(r.row_count());
}

// Fixture: one channel + a scratch table per test (built in SetUp, dropped in
// TearDown).  InnoDB so DML is transactional.
class MysqlTxnIntegrationTest : public testing::Test {
protected:
    static bool NoServer() { return s_mysqld_pid == -1; }

    void SetUp() override {
        pthread_once(&s_start_once, RunMysqlServer);
        if (NoServer()) {
            GTEST_SKIP() << "no mysqld available; skipping transaction tests";
        }
        // Authenticator carries user/password and the working schema.  An
        // empty schema is created first over a schema-less channel.
        ASSERT_TRUE(InitChannel(&_setup_channel, /*schema=*/""));
        brpc::MysqlResponse resp;
        ASSERT_TRUE(RunPlain(_setup_channel, "CREATE DATABASE IF NOT EXISTS " +
                                                 FLAGS_mysql_schema, &resp));

        ASSERT_TRUE(InitChannel(&_channel, FLAGS_mysql_schema));
        ASSERT_TRUE(RunPlain(_channel, "DROP TABLE IF EXISTS " + Table(), &resp));
        ASSERT_TRUE(ExpectOk(resp)) << "drop pre-existing scratch table";
        ASSERT_TRUE(RunPlain(_channel,
                             "CREATE TABLE " + Table() +
                                 " (id INT PRIMARY KEY, name VARCHAR(32)) "
                                 "ENGINE=InnoDB",
                             &resp));
        ASSERT_TRUE(ExpectOk(resp)) << "create scratch table";
    }

    void TearDown() override {
        if (NoServer()) {
            return;
        }
        brpc::MysqlResponse resp;
        RunPlain(_channel, "DROP TABLE IF EXISTS " + Table(), &resp);
    }

    // Pooled channel is required so a transaction can pin its own dedicated
    // connection while the test issues independent observer queries on others.
    bool InitChannel(brpc::Channel* channel, const std::string& schema) {
        _auth.reset(new brpc::policy::MysqlAuthenticator(s_user, s_password,
                                                         schema));
        brpc::ChannelOptions options;
        options.protocol = "mysql";
        options.connection_type = "pooled";
        options.auth = _auth.get();
        options.timeout_ms = 5000;
        options.max_retry = 0;
        char addr[128];
        snprintf(addr, sizeof(addr), "%s:%d", s_host.c_str(), s_port);
        return channel->Init(addr, &options) == 0;
    }

    std::string Table() const { return "txn_scratch"; }

    brpc::Channel _setup_channel;
    brpc::Channel _channel;
    // Authenticator must outlive the channels that point at it.
    std::unique_ptr<brpc::policy::MysqlAuthenticator> _auth;
};

// Test cases.  Each fat test chains several transactional behaviors so a single
// TEST_F validates a whole group of related transaction guarantees together.

// Transaction lifecycle: commit publishes a row to other connections, and a
// rolled-back insert as well as a rolled-back delete leave the table exactly as
// it was before the transaction started.
TEST_F(MysqlTxnIntegrationTest, CommitPublishesRollbackRestores) {
    brpc::MysqlResponse resp;

    // 1) committed INSERT must be visible on a fresh connection afterwards.
    {
        brpc::MysqlTransactionUniquePtr tx =
            brpc::NewMysqlTransaction(_channel, brpc::MysqlTransactionOptions());
        ASSERT_TRUE(tx != NULL) << "failed to start transaction";
        ASSERT_TRUE(RunInTx(_channel, tx.get(),
                            "INSERT INTO " + Table() + " VALUES (3107, 'quill')",
                            &resp));
        EXPECT_TRUE(ExpectOk(resp));
        EXPECT_EQ(resp.reply(0).ok().affect_row(), 1u);
        ASSERT_TRUE(tx->commit());
    }
    ASSERT_TRUE(RunPlain(_channel, "SELECT id, name FROM " + Table(), &resp));
    EXPECT_EQ(ResultRowCount(resp), 1);
    {
        const brpc::MysqlReply& r = resp.reply(0);
        ASSERT_TRUE(r.is_resultset());
        ASSERT_EQ(r.row_count(), 1u);
        const brpc::MysqlReply::Row& row = r.next();
        ASSERT_EQ(row.field_count(), 2u);
        EXPECT_EQ(row.field(0).sinteger(), 3107);
        EXPECT_EQ(row.field(1).string().as_string(), "quill");
    }

    // 2) a rolled-back INSERT must leave no trace (still exactly one row).
    {
        brpc::MysqlTransactionUniquePtr tx =
            brpc::NewMysqlTransaction(_channel, brpc::MysqlTransactionOptions());
        ASSERT_TRUE(tx != NULL);
        ASSERT_TRUE(RunInTx(_channel, tx.get(),
                            "INSERT INTO " + Table() + " VALUES (5288, 'brindle')",
                            &resp));
        EXPECT_TRUE(ExpectOk(resp));
        ASSERT_TRUE(tx->rollback());
    }
    ASSERT_TRUE(RunPlain(_channel, "SELECT id FROM " + Table(), &resp));
    EXPECT_EQ(ResultRowCount(resp), 1) << "rolled-back insert must vanish";

    // 3) a rolled-back DELETE of the committed row must restore it.
    {
        brpc::MysqlTransactionUniquePtr tx =
            brpc::NewMysqlTransaction(_channel, brpc::MysqlTransactionOptions());
        ASSERT_TRUE(tx != NULL);
        ASSERT_TRUE(RunInTx(_channel, tx.get(),
                            "DELETE FROM " + Table() + " WHERE id = 3107", &resp));
        EXPECT_TRUE(ExpectOk(resp));
        EXPECT_EQ(resp.reply(0).ok().affect_row(), 1u);
        ASSERT_TRUE(tx->rollback());
    }
    ASSERT_TRUE(RunPlain(_channel, "SELECT id FROM " + Table(), &resp));
    EXPECT_EQ(ResultRowCount(resp), 1) << "rolled-back delete must restore row";
    {
        const brpc::MysqlReply& r = resp.reply(0);
        ASSERT_TRUE(r.is_resultset());
        ASSERT_EQ(r.row_count(), 1u);
        EXPECT_EQ(r.next().field(0).sinteger(), 3107);
    }
}

// Isolation in both directions on the same open transaction: the transaction
// reads its own not-yet-committed write on its pinned connection, while an
// independent pooled connection sees nothing until the rollback.
TEST_F(MysqlTxnIntegrationTest, OwnWriteVisibleOthersIsolated) {
    brpc::MysqlTransactionUniquePtr tx =
        brpc::NewMysqlTransaction(_channel, brpc::MysqlTransactionOptions());
    ASSERT_TRUE(tx != NULL);

    brpc::MysqlResponse resp;
    ASSERT_TRUE(RunInTx(_channel, tx.get(),
                        "INSERT INTO " + Table() + " VALUES (6741, 'tangle')",
                        &resp));
    EXPECT_TRUE(ExpectOk(resp));

    // Same pinned connection: must read its own uncommitted row.
    ASSERT_TRUE(RunInTx(_channel, tx.get(),
                        "SELECT name FROM " + Table() + " WHERE id = 6741", &resp));
    EXPECT_EQ(ResultRowCount(resp), 1);
    {
        const brpc::MysqlReply& r = resp.reply(0);
        ASSERT_TRUE(r.is_resultset());
        ASSERT_EQ(r.row_count(), 1u);
        EXPECT_EQ(r.next().field(0).string().as_string(), "tangle");
    }

    // A different pooled connection must NOT see the uncommitted write.
    ASSERT_TRUE(RunPlain(_channel, "SELECT id FROM " + Table(), &resp));
    EXPECT_EQ(ResultRowCount(resp), 0)
        << "uncommitted write leaked to another connection";

    ASSERT_TRUE(tx->rollback());

    // After rollback nothing remains anywhere.
    ASSERT_TRUE(RunPlain(_channel, "SELECT id FROM " + Table(), &resp));
    EXPECT_EQ(ResultRowCount(resp), 0);
}

// Autocommit behavior, both states in one test: with autocommit on (the
// default) a bare INSERT is immediately durable on a new connection; toggling
// autocommit off on a pinned connection turns a later INSERT into pending work
// that ROLLBACK discards.
TEST_F(MysqlTxnIntegrationTest, AutocommitOnDurableOffRollbackable) {
    brpc::MysqlResponse resp;

    // autocommit ON: immediate durability.
    ASSERT_TRUE(RunPlain(_channel,
                         "INSERT INTO " + Table() + " VALUES (4419, 'amber')",
                         &resp));
    EXPECT_TRUE(ExpectOk(resp));
    ASSERT_TRUE(RunPlain(_channel, "SELECT id FROM " + Table(), &resp));
    EXPECT_EQ(ResultRowCount(resp), 1);

    // autocommit OFF on a pinned connection: a new INSERT is pending and a
    // ROLLBACK drops only it, leaving the earlier durable row in place.
    {
        brpc::MysqlTransactionUniquePtr tx =
            brpc::NewMysqlTransaction(_channel, brpc::MysqlTransactionOptions());
        ASSERT_TRUE(tx != NULL);
        ASSERT_TRUE(RunInTx(_channel, tx.get(), "SET autocommit = 0", &resp));
        EXPECT_TRUE(ExpectOk(resp));
        ASSERT_TRUE(RunInTx(_channel, tx.get(),
                            "INSERT INTO " + Table() + " VALUES (8053, 'frost')",
                            &resp));
        EXPECT_TRUE(ExpectOk(resp));
        ASSERT_TRUE(tx->rollback());
    }
    ASSERT_TRUE(RunPlain(_channel, "SELECT id FROM " + Table(), &resp));
    EXPECT_EQ(ResultRowCount(resp), 1)
        << "autocommit=0 + rollback should drop only the pending insert";
    EXPECT_EQ(resp.reply(0).next().field(0).sinteger(), 4419);
}

// Multi-statement transactional grouping plus partial undo: two inserts grouped
// under one transaction become visible together only after commit, and within a
// second transaction a SAVEPOINT lets a later insert be peeled back while the
// pre-savepoint work survives the final commit.
TEST_F(MysqlTxnIntegrationTest, GroupedInsertsThenSavepointPartialUndo) {
    brpc::MysqlResponse resp;

    // Two inserts under one transaction: invisible until commit, then both show.
    {
        brpc::MysqlTransactionUniquePtr tx =
            brpc::NewMysqlTransaction(_channel, brpc::MysqlTransactionOptions());
        ASSERT_TRUE(tx != NULL);
        ASSERT_TRUE(RunInTx(_channel, tx.get(),
                            "INSERT INTO " + Table() + " VALUES (211, 'one')",
                            &resp));
        EXPECT_TRUE(ExpectOk(resp));
        ASSERT_TRUE(RunInTx(_channel, tx.get(),
                            "INSERT INTO " + Table() + " VALUES (733, 'two')",
                            &resp));
        EXPECT_TRUE(ExpectOk(resp));

        // Not yet visible to a separate connection.
        ASSERT_TRUE(RunPlain(_channel, "SELECT id FROM " + Table(), &resp));
        EXPECT_EQ(ResultRowCount(resp), 0);

        ASSERT_TRUE(tx->commit());
    }
    ASSERT_TRUE(RunPlain(_channel, "SELECT id FROM " + Table(), &resp));
    EXPECT_EQ(ResultRowCount(resp), 2) << "both grouped inserts visible";

    // Start fresh for the savepoint half.
    ASSERT_TRUE(RunPlain(_channel, "DELETE FROM " + Table(), &resp));
    ASSERT_TRUE(ExpectOk(resp));

    // SAVEPOINT then ROLLBACK TO it: pre-savepoint row kept, post dropped.
    {
        brpc::MysqlTransactionUniquePtr tx =
            brpc::NewMysqlTransaction(_channel, brpc::MysqlTransactionOptions());
        ASSERT_TRUE(tx != NULL);
        ASSERT_TRUE(RunInTx(_channel, tx.get(),
                            "INSERT INTO " + Table() + " VALUES (901, 'kept')",
                            &resp));
        EXPECT_TRUE(ExpectOk(resp));
        ASSERT_TRUE(RunInTx(_channel, tx.get(), "SAVEPOINT mark1", &resp));
        EXPECT_TRUE(ExpectOk(resp));
        ASSERT_TRUE(RunInTx(_channel, tx.get(),
                            "INSERT INTO " + Table() + " VALUES (902, 'gone')",
                            &resp));
        EXPECT_TRUE(ExpectOk(resp));
        ASSERT_TRUE(RunInTx(_channel, tx.get(), "ROLLBACK TO SAVEPOINT mark1",
                            &resp));
        EXPECT_TRUE(ExpectOk(resp));

        // Inside the txn only the pre-savepoint row remains.
        ASSERT_TRUE(RunInTx(_channel, tx.get(), "SELECT id FROM " + Table(),
                            &resp));
        EXPECT_EQ(ResultRowCount(resp), 1);

        ASSERT_TRUE(tx->commit());
    }
    ASSERT_TRUE(RunPlain(_channel, "SELECT id FROM " + Table(), &resp));
    EXPECT_EQ(ResultRowCount(resp), 1) << "only the kept row should persist";
    {
        const brpc::MysqlReply& r = resp.reply(0);
        ASSERT_TRUE(r.is_resultset());
        ASSERT_EQ(r.row_count(), 1u);
        EXPECT_EQ(r.next().field(0).sinteger(), 901);
    }
}

// Error surfaces from within a transaction: a duplicate-primary-key insert
// returns an ERR reply (and the transaction still rolls back cleanly), and a
// write attempted in a read-only transaction is likewise rejected with ERR.
TEST_F(MysqlTxnIntegrationTest, DuplicateKeyAndReadOnlyWriteReportErr) {
    brpc::MysqlResponse resp;

    // Seed a committed row so the in-txn insert collides on the primary key.
    ASSERT_TRUE(RunPlain(_channel,
                         "INSERT INTO " + Table() + " VALUES (1505, 'seed')",
                         &resp));
    ASSERT_TRUE(ExpectOk(resp));

    {
        brpc::MysqlTransactionUniquePtr tx =
            brpc::NewMysqlTransaction(_channel, brpc::MysqlTransactionOptions());
        ASSERT_TRUE(tx != NULL);
        // Duplicate-key insert -> ERR packet (errno 1062, ER_DUP_ENTRY).
        ASSERT_TRUE(RunInTx(_channel, tx.get(),
                            "INSERT INTO " + Table() + " VALUES (1505, 'clash')",
                            &resp));
        ASSERT_GE(resp.reply_size(), 1u);
        const brpc::MysqlReply& r = resp.reply(0);
        EXPECT_TRUE(r.is_error()) << "duplicate key should yield an ERR reply";
        if (r.is_error()) {
            EXPECT_EQ(r.error().errcode(), 1062) << "expected ER_DUP_ENTRY (1062)";
        }
        ASSERT_TRUE(tx->rollback());
    }

    // A read-only transaction must reject a write.
    {
        brpc::MysqlTransactionOptions opts;
        opts.readonly = true;
        brpc::MysqlTransactionUniquePtr tx =
            brpc::NewMysqlTransaction(_channel, opts);
        ASSERT_TRUE(tx != NULL) << "failed to start read-only transaction";
        ASSERT_TRUE(RunInTx(_channel, tx.get(),
                            "INSERT INTO " + Table() + " VALUES (1777, 'nope')",
                            &resp));
        ASSERT_GE(resp.reply_size(), 1u);
        const brpc::MysqlReply& r = resp.reply(0);
        EXPECT_TRUE(r.is_error())
            << "write in a read-only transaction should be rejected with ERR";
        if (r.is_error()) {
            // ER_CANT_EXECUTE_IN_READ_ONLY_TRANSACTION == 1792.
            EXPECT_EQ(r.error().errcode(), 1792)
                << "expected read-only-transaction error (1792)";
        }
        ASSERT_TRUE(tx->rollback());
    }
}

}  // namespace