如何Mock支持Unix Socket的MariaDB服务以满足测试需求？

针对你的需求——无需创建数据库/表、支持Unix Socket连接、接受所有INSERT/UPDATE查询的Mock服务，以下是几个可直接运行的方案：

1. Go语言轻量Mock服务（基于go-mysql-server） #

这是一个灵活的自定义Mock方案，完全符合你的需求：

• 安装依赖：

  go get github.com/dolthub/go-mysql-server
  

• 编写服务代码（保存为mock_mariadb.go）：

  package main
  
  import (
    "github.com/dolthub/go-mysql-server/auth"
    "github.com/dolthub/go-mysql-server/server"
    "github.com/dolthub/go-mysql-server/sql"
    "os"
  )
  
  type mockEngine struct{}
  
  func (e *mockEngine) Database(name string) (sql.Database, error) {
    return &mockDB{name: name}, nil
  }
  
  func (e *mockEngine) AllDatabases() []sql.Database {
    return []sql.Database{}
  }
  
  type mockDB struct{ name string }
  
  func (d *mockDB) Name() string { return d.name }
  func (d *mockDB) Tables() ([]sql.Table, error) { return []sql.Table{}, nil }
  func (d *mockDB) GetTable(ctx *sql.Context, name string) (sql.Table, bool, error) {
    return &mockTable{name: name}, true, nil
  }
  
  type mockTable struct{ name string }
  
  func (t *mockTable) Name() string { return t.name }
  func (t *mockTable) Schema() sql.Schema { return []sql.Column{} }
  func (t *mockTable) Partitions(ctx *sql.Context) (sql.PartitionIter, error) { return nil, nil }
  func (t *mockTable) PartitionRows(ctx *sql.Context, partition sql.Partition) (sql.RowIter, error) { return nil, nil }
  func (t *mockTable) Insert(ctx *sql.Context, rows sql.RowIter) error { return nil }
  func (t *mockTable) Update(ctx *sql.Context, updater sql.Updater) error { return nil }
  func (t *mockTable) Delete(ctx *sql.Context, d sql.Deleter) error { return nil }
  
  func main() {
    socketPath := "/tmp/mariadb_mock.sock"
    os.Remove(socketPath)
  
    engine := &mockEngine{}
    config := server.Config{
      Protocol: "unix",
      Address:  socketPath,
      Auth:     auth.NewNativeSingle("test_user", "test_pass", auth.AllPermissions),
    }
  
    s, err := server.NewServer(config, engine)
    if err != nil {
      panic(err)
    }
  
    if err := s.Start(); err != nil {
      panic(err)
    }
  }
  

• 启动服务：

  go run mock_mariadb.go
  

该服务监听/tmp/mariadb_mock.sock，接受配置的用户名密码（可自行修改），所有写操作均返回成功，无需提前创建任何库表。

2. Node.js工具：mysql-mock-server #

这是一个开箱即用的Mock服务，无需编写代码：

• 安装工具：

  npm install -g mysql-mock-server
  

• 启动Unix Socket模式的Mock服务：

  mysql-mock-server --socket /tmp/mariadb_mock.sock --allowAnyQuery
  

--allowAnyQuery参数会让服务自动接受所有查询并返回成功响应，完全满足你的测试需求。

3. Python自定义Mock服务（基于mysql-protocol） #

适合熟悉Python的开发者快速搭建：

• 安装依赖：

  pip install mysql-protocol
  

• 编写服务代码（保存为mock_mariadb.py）：

  import socket
  import os
  from mysql_protocol import MysqlProtocol
  
  def handle_client(conn):
    proto = MysqlProtocol(conn)
    # 完成握手和认证流程
    proto.handshake()
    proto.authenticate()
    
    while True:
      try:
        # 读取客户端查询包
        packet = proto.read_packet()
        if not packet:
          break
        # 对所有查询返回OK响应
        proto.write_ok()
      except Exception:
        break
    conn.close()
  
  def main():
    socket_path = "/tmp/mariadb_mock.sock"
    # 清理旧的Socket文件
    try:
      os.unlink(socket_path)
    except OSError:
      if os.path.exists(socket_path):
        raise
  
    server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    server_sock.bind(socket_path)
    server_sock.listen(5)
  
    print(f"Mock MariaDB listening on Unix Socket: {socket_path}")
    while True:
      conn, _ = server_sock.accept()
      handle_client(conn)
  
  if __name__ == "__main__":
    main()
  

• 启动服务：

  python mock_mariadb.py
  

这个服务会自动处理连接认证，对所有客户端发送的INSERT/UPDATE等查询返回成功结果。

内容的提问来源于stack exchange，提问作者Mathis Germa