380 lines
13 KiB
Go
380 lines
13 KiB
Go
package db
|
||
|
||
// agent_e2e_test.go —— JDBC agent 桥端到端验证（真实 JVM + 真实 JDBC 驱动）。
|
||
// 前置：本机 JDK、agent jar（OPSXDB_AGENT_JAR 或默认 ~/.opsxdb/agents/jdbc/）。
|
||
// 触发：OPSXDB_E2E_JDBC=1 才运行（CI 无 JVM 时自动跳过）。
|
||
// 覆盖：StartAgent 全链路（gson 校验/JVM 启动/握手）→ open → DDL/DML → query → 元数据。
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
func TestE2EAgentSQLiteBridge(t *testing.T) {
|
||
if os.Getenv("OPSXDB_E2E_JDBC") != "1" {
|
||
t.Skip("置 OPSXDB_E2E_JDBC=1 并备好 JDK 与 agent jar 后运行")
|
||
}
|
||
home, _ := os.UserHomeDir()
|
||
jars := filepath.Join(home, ".opsxdb", "agents", "jdbc", "drivers", "sqlite-e2e")
|
||
if _, err := os.Stat(filepath.Join(jars, "sqlite-jdbc-3.46.1.3.jar")); err != nil {
|
||
t.Skipf("缺 sqlite-jdbc jar: %s", jars)
|
||
}
|
||
t.Setenv("OPSXDB_E2E_JARS", jars)
|
||
|
||
spec := AgentSpec{
|
||
DriverKey: "e2e-sqlite",
|
||
DriverClass: "org.sqlite.JDBC",
|
||
URLTemplate: "jdbc:sqlite:{{.Database}}",
|
||
JarsDirEnv: "OPSXDB_E2E_JARS",
|
||
}
|
||
agentDriverSpecs[spec.DriverKey] = spec // 注册进表，AgentConfigured 才能按 JarsDirEnv 解析
|
||
t.Cleanup(func() { delete(agentDriverSpecs, spec.DriverKey) })
|
||
cli, err := StartAgent(spec)
|
||
if err != nil {
|
||
t.Fatalf("StartAgent: %v", err)
|
||
}
|
||
t.Cleanup(func() { cli.Shutdown(context.Background()) })
|
||
|
||
dbPath := filepath.Join(t.TempDir(), "e2e.db")
|
||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||
defer cancel()
|
||
|
||
handle := ""
|
||
{
|
||
var res struct {
|
||
Handle string `json:"handle"`
|
||
}
|
||
if err := cli.call(ctx, "open", map[string]any{"url": "jdbc:sqlite:" + dbPath, "user": "", "password": ""}, &res); err != nil {
|
||
t.Fatalf("open: %v (stderr=%s)", err, cli.stderrTail.String())
|
||
}
|
||
handle = res.Handle
|
||
}
|
||
mustExec := func(sql string) {
|
||
t.Helper()
|
||
var r struct {
|
||
AffectedRows int64 `json:"affectedRows"`
|
||
}
|
||
if err := cli.call(ctx, "exec", map[string]any{"handle": handle, "sql": sql, "timeoutSec": 30}, &r); err != nil {
|
||
t.Fatalf("exec %q: %v (stderr=%s)", sql, err, cli.stderrTail.String())
|
||
}
|
||
}
|
||
mustExec("CREATE TABLE demo(id INTEGER PRIMARY KEY, name TEXT)")
|
||
mustExec("INSERT INTO demo(name) VALUES ('opsx-e2e')")
|
||
|
||
{
|
||
var res struct {
|
||
Columns []string `json:"columns"`
|
||
Rows []map[string]interface{} `json:"rows"`
|
||
}
|
||
if err := cli.call(ctx, "query", map[string]any{"handle": handle, "sql": "SELECT id, name FROM demo", "maxRows": 100, "timeoutSec": 30}, &res); err != nil {
|
||
t.Fatalf("query: %v", err)
|
||
}
|
||
if len(res.Rows) != 1 || res.Rows[0]["name"] != "opsx-e2e" {
|
||
t.Fatalf("rows = %+v", res.Rows)
|
||
}
|
||
}
|
||
{
|
||
var res struct {
|
||
Names []string `json:"names"`
|
||
}
|
||
if err := cli.call(ctx, "metaTables", map[string]any{"handle": handle, "schema": ""}, &res); err != nil {
|
||
t.Fatalf("metaTables: %v", err)
|
||
}
|
||
found := false
|
||
for _, n := range res.Names {
|
||
if n == "demo" {
|
||
found = true
|
||
}
|
||
}
|
||
if !found {
|
||
t.Fatalf("metaTables 未包含 demo: %v", res.Names)
|
||
}
|
||
}
|
||
{
|
||
var res struct {
|
||
Columns []ColumnInfo `json:"columns"`
|
||
}
|
||
if err := cli.call(ctx, "metaColumns", map[string]any{"handle": handle, "schema": "", "table": "demo"}, &res); err != nil {
|
||
t.Fatalf("metaColumns: %v", err)
|
||
}
|
||
if len(res.Columns) != 2 {
|
||
t.Fatalf("columns = %+v", res.Columns)
|
||
}
|
||
}
|
||
}
|
||
|
||
// TestE2EElasticsearch —— 真实 ES 8（docker，安全关闭）+ 官方 x-pack-sql-jdbc 走 agent 桥全链路。
|
||
// 触发：OPSXDB_E2E_ES=1；前置 ES :9200 可连、~/.opsxdb/agents/jdbc/drivers/elasticsearch/ 有官方 JDBC jar。
|
||
func TestE2EElasticsearch(t *testing.T) {
|
||
if os.Getenv("OPSXDB_E2E_ES") != "1" {
|
||
t.Skip("置 OPSXDB_E2E_ES=1 并启动 ES :9200 后运行")
|
||
}
|
||
// 种子文档（让 SHOW TABLES / SELECT 有真实对象）
|
||
req, _ := http.NewRequest(http.MethodPut, "http://127.0.0.1:9200/e2e_smoke/_doc/1?refresh=true", strings.NewReader(`{"name":"opsx","n":42}`))
|
||
req.Header.Set("Content-Type", "application/json")
|
||
resp, err := http.DefaultClient.Do(req)
|
||
if err != nil {
|
||
t.Skipf("ES :9200 不可达: %v", err)
|
||
}
|
||
_, _ = io.Copy(io.Discard, resp.Body)
|
||
resp.Body.Close()
|
||
if resp.StatusCode >= 400 {
|
||
t.Fatalf("种子文档写入失败: HTTP %d", resp.StatusCode)
|
||
}
|
||
|
||
c := ConnReq{Driver: "elasticsearch", Host: "127.0.0.1", Port: 9200}
|
||
if err := TestConnection(c); err != nil {
|
||
t.Fatalf("TestConnection: %v", err)
|
||
}
|
||
dbs, err := ListDatabases(context.Background(), c)
|
||
if err != nil || len(dbs) != 1 || dbs[0] != "default" {
|
||
t.Fatalf("ListDatabases = %v, %v", dbs, err)
|
||
}
|
||
cc := c
|
||
cc.Database = "default"
|
||
tables, err := ListTables(context.Background(), cc)
|
||
if err != nil {
|
||
t.Fatalf("ListTables: %v", err)
|
||
}
|
||
found := false
|
||
for _, n := range tables {
|
||
if n == "e2e_smoke" {
|
||
found = true
|
||
}
|
||
}
|
||
if !found {
|
||
t.Fatalf("tables 未含 e2e_smoke: %v", tables)
|
||
}
|
||
r := ExecSQL(context.Background(), ConnReq{Driver: "elasticsearch", Host: "127.0.0.1", Port: 9200, Database: "default", SQL: "SELECT name, n FROM e2e_smoke", Limit: 100})
|
||
if r.Error != "" {
|
||
t.Fatalf("ExecSQL: %s (stderr 见测试日志)", r.Error)
|
||
}
|
||
if r.RowCount < 1 {
|
||
t.Fatalf("rows = %+v", r.Rows)
|
||
}
|
||
}
|
||
|
||
// TestE2EZooKeeper —— 真实 ZK（docker）+ OpsxZkAgent 全链路：连接/浏览/读写/删除。
|
||
// 触发：OPSXDB_E2E_ZK=1；前置 ZK :2181、~/.opsxdb/agents/jdbc/drivers/zookeeper/ 有官方 jar。
|
||
func TestE2EZooKeeper(t *testing.T) {
|
||
if os.Getenv("OPSXDB_E2E_ZK") != "1" {
|
||
t.Skip("置 OPSXDB_E2E_ZK=1 并启动 ZK :2181 后运行")
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
||
defer cancel()
|
||
|
||
c := ConnReq{Driver: "zookeeper", Host: "127.0.0.1", Port: 2181}
|
||
if err := TestConnection(c); err != nil {
|
||
t.Skipf("ZK :2181 不可达: %v", err)
|
||
}
|
||
dbs, err := ListDatabases(ctx, c)
|
||
if err != nil || len(dbs) == 0 {
|
||
t.Fatalf("ListDatabases = %v, %v", dbs, err)
|
||
}
|
||
hasZk := false
|
||
for _, n := range dbs {
|
||
if n == "zookeeper" {
|
||
hasZk = true
|
||
}
|
||
}
|
||
if !hasZk {
|
||
t.Fatalf("根下应含 zookeeper: %v", dbs)
|
||
}
|
||
cc := c
|
||
cc.Database = "zookeeper"
|
||
tables, err := ListTables(ctx, cc)
|
||
if err != nil {
|
||
t.Fatalf("ListTables: %v", err)
|
||
}
|
||
hasConfig := false
|
||
for _, n := range tables {
|
||
if n == "config" {
|
||
hasConfig = true
|
||
}
|
||
}
|
||
if !hasConfig {
|
||
t.Fatalf("/zookeeper 下应含 config: %v", tables)
|
||
}
|
||
// 读：GET
|
||
r := ExecSQL(ctx, ConnReq{Driver: "zookeeper", Host: "127.0.0.1", Port: 2181, Database: "zookeeper", SQL: "GET /zookeeper/config", Limit: 100})
|
||
if r.Error != "" {
|
||
t.Fatalf("GET: %s", r.Error)
|
||
}
|
||
if r.RowCount != 1 || r.Rows[0]["path"] != "/zookeeper/config" {
|
||
t.Fatalf("GET rows = %+v", r.Rows)
|
||
}
|
||
// 写：CREATE → GET → SET → DELETE（DELETE 无 WHERE 属危险语句，走确认通道）
|
||
w := func(sql string) {
|
||
t.Helper()
|
||
res := ExecSQL(ctx, ConnReq{Driver: "zookeeper", Host: "127.0.0.1", Port: 2181, SQL: sql, ConfirmDanger: true})
|
||
if res.Error != "" {
|
||
t.Fatalf("%s → %s", sql, res.Error)
|
||
}
|
||
}
|
||
w("CREATE /opsxe2e hello-opsx")
|
||
r = ExecSQL(ctx, ConnReq{Driver: "zookeeper", Host: "127.0.0.1", Port: 2181, SQL: "GET /opsxe2e", Limit: 100})
|
||
if r.Rows[0]["data"] != "hello-opsx" {
|
||
t.Fatalf("CREATE 后 data = %v", r.Rows[0]["data"])
|
||
}
|
||
w("SET /opsxe2e updated")
|
||
r = ExecSQL(ctx, ConnReq{Driver: "zookeeper", Host: "127.0.0.1", Port: 2181, SQL: "GET /opsxe2e", Limit: 100})
|
||
if r.Rows[0]["data"] != "updated" {
|
||
t.Fatalf("SET 后 data = %v", r.Rows[0]["data"])
|
||
}
|
||
// 危险语句未确认时必须先返回 NeedConfirm
|
||
guard := ExecSQL(ctx, ConnReq{Driver: "zookeeper", Host: "127.0.0.1", Port: 2181, SQL: "DELETE /opsxe2e"})
|
||
if !guard.NeedConfirm {
|
||
t.Fatalf("DELETE 未确认应返回 NeedConfirm，got %+v", guard)
|
||
}
|
||
w("DELETE /opsxe2e")
|
||
}
|
||
|
||
// TestE2EInfluxDB —— 真实 InfluxDB 2.7（docker）原生 HTTP 全链路：bucket/measurement/Flux 查询。
|
||
// 触发：OPSXDB_E2E_INFLUX=1；前置 InfluxDB :8086（org opsx / token opsxdb-demo-token / bucket demo）。
|
||
func TestE2EInfluxDB(t *testing.T) {
|
||
if os.Getenv("OPSXDB_E2E_INFLUX") != "1" {
|
||
t.Skip("置 OPSXDB_E2E_INFLUX=1 并启动 InfluxDB :8086 后运行")
|
||
}
|
||
// 种子点（行协议写入）
|
||
wreq, _ := http.NewRequest(http.MethodPost, "http://127.0.0.1:8086/api/v2/write?org=opsx&bucket=demo&precision=s",
|
||
strings.NewReader("e2e_m,site=local v=42"))
|
||
wreq.Header.Set("Authorization", "Token opsxdb-demo-token")
|
||
wreq.Header.Set("Content-Type", "text/plain; charset=utf-8")
|
||
wresp, err := http.DefaultClient.Do(wreq)
|
||
if err != nil {
|
||
t.Skipf("InfluxDB :8086 不可达: %v", err)
|
||
}
|
||
_, _ = io.Copy(io.Discard, wresp.Body)
|
||
wresp.Body.Close()
|
||
if wresp.StatusCode >= 400 {
|
||
t.Fatalf("种子写入失败: HTTP %d", wresp.StatusCode)
|
||
}
|
||
time.Sleep(500 * time.Millisecond) // 存储引擎落盘
|
||
|
||
c := ConnReq{Driver: "influxdb", Host: "127.0.0.1", Port: 8086, User: "opsx", Password: "opsxdb-demo-token"}
|
||
if err := TestConnection(c); err != nil {
|
||
t.Fatalf("TestConnection: %v", err)
|
||
}
|
||
dbs, err := ListDatabases(context.Background(), c)
|
||
if err != nil {
|
||
t.Fatalf("ListDatabases: %v", err)
|
||
}
|
||
hasDemo := false
|
||
for _, n := range dbs {
|
||
if n == "demo" {
|
||
hasDemo = true
|
||
}
|
||
}
|
||
if !hasDemo {
|
||
t.Fatalf("buckets 未含 demo: %v", dbs)
|
||
}
|
||
cc := c
|
||
cc.Database = "demo"
|
||
tables, err := ListTables(context.Background(), cc)
|
||
if err != nil {
|
||
t.Fatalf("ListTables: %v", err)
|
||
}
|
||
hasM := false
|
||
for _, n := range tables {
|
||
if n == "e2e_m" {
|
||
hasM = true
|
||
}
|
||
}
|
||
if !hasM {
|
||
t.Fatalf("measurements 未含 e2e_m: %v", tables)
|
||
}
|
||
r := ExecSQL(context.Background(), ConnReq{
|
||
Driver: "influxdb", Host: "127.0.0.1", Port: 8086, User: "opsx", Password: "opsxdb-demo-token",
|
||
SQL: `from(bucket: "demo") |> range(start: -1h) |> filter(fn: (r) => r._measurement == "e2e_m")`,
|
||
Limit: 100,
|
||
})
|
||
if r.Error != "" {
|
||
t.Fatalf("Flux 查询: %s", r.Error)
|
||
}
|
||
if r.RowCount < 1 {
|
||
t.Fatalf("rows = %+v", r.Rows)
|
||
}
|
||
}
|
||
|
||
// TestE2EMongoDB —— 真实 MongoDB 7 + OpsxMongoAgent 全链路：库/集合浏览、DSL 查改增删、危险确认。
|
||
// 触发：OPSXDB_E2E_MONGO=1；前置 MongoDB :27017、~/.opsxdb/agents/jdbc/drivers/mongodb/ 有官方 jar。
|
||
func TestE2EMongoDB(t *testing.T) {
|
||
if os.Getenv("OPSXDB_E2E_MONGO") != "1" {
|
||
t.Skip("置 OPSXDB_E2E_MONGO=1 并启动 MongoDB :27017 后运行")
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
||
defer cancel()
|
||
|
||
c := ConnReq{Driver: "mongodb", Host: "127.0.0.1", Port: 27017}
|
||
if err := TestConnection(c); err != nil {
|
||
t.Skipf("MongoDB :27017 不可达: %v", err)
|
||
}
|
||
// 种子：插入文档（确认通道）
|
||
seed := ExecSQL(ctx, ConnReq{Driver: "mongodb", Host: "127.0.0.1", Port: 27017, Database: "opsxe2e", SQL: `db.c1.insertOne({"name":"opsx","n":7})`, ConfirmDanger: false})
|
||
if seed.Error != "" {
|
||
t.Fatalf("insertOne: %s", seed.Error)
|
||
}
|
||
dbs, err := ListDatabases(ctx, c)
|
||
if err != nil || len(dbs) == 0 {
|
||
t.Fatalf("ListDatabases = %v, %v", dbs, err)
|
||
}
|
||
hasDb := false
|
||
for _, n := range dbs {
|
||
if n == "opsxe2e" {
|
||
hasDb = true
|
||
}
|
||
}
|
||
if !hasDb {
|
||
t.Fatalf("库列表未含 opsxe2e: %v", dbs)
|
||
}
|
||
cc := c
|
||
cc.Database = "opsxe2e"
|
||
tables, err := ListTables(ctx, cc)
|
||
if err != nil || len(tables) == 0 {
|
||
t.Fatalf("ListTables = %v, %v", tables, err)
|
||
}
|
||
found := false
|
||
for _, n := range tables {
|
||
if n == "c1" {
|
||
found = true
|
||
}
|
||
}
|
||
if !found {
|
||
t.Fatalf("集合列表未含 c1: %v", tables)
|
||
}
|
||
cols, err := ListColumns(ctx, ConnReq{Driver: "mongodb", Host: "127.0.0.1", Port: 27017, Database: "opsxe2e", Table: "c1"})
|
||
if err != nil || len(cols) == 0 {
|
||
t.Fatalf("ListColumns = %v, %v", cols, err)
|
||
}
|
||
// 查询
|
||
q := ExecSQL(ctx, ConnReq{Driver: "mongodb", Host: "127.0.0.1", Port: 27017, Database: "opsxe2e", SQL: `db.c1.find({"name":"opsx"})`, Limit: 100})
|
||
if q.Error != "" || q.RowCount < 1 {
|
||
t.Fatalf("find: err=%s rows=%+v", q.Error, q.Rows)
|
||
}
|
||
// 危险确认：deleteMany 未确认必须先返回 NeedConfirm
|
||
guard := ExecSQL(ctx, ConnReq{Driver: "mongodb", Host: "127.0.0.1", Port: 27017, Database: "opsxe2e", SQL: `db.c1.deleteMany({"name":"opsx"})`})
|
||
if !guard.NeedConfirm {
|
||
t.Fatalf("deleteMany 未确认应 NeedConfirm，got %+v", guard)
|
||
}
|
||
// 更新 + 删除（确认通道）
|
||
u := ExecSQL(ctx, ConnReq{Driver: "mongodb", Host: "127.0.0.1", Port: 27017, Database: "opsxe2e", SQL: `db.c1.updateOne({"name":"opsx"}, {"$set":{"n":8}})`, ConfirmDanger: true})
|
||
if u.Error != "" {
|
||
t.Fatalf("updateOne: %s", u.Error)
|
||
}
|
||
d := ExecSQL(ctx, ConnReq{Driver: "mongodb", Host: "127.0.0.1", Port: 27017, Database: "opsxe2e", SQL: `db.c1.deleteMany({"name":"opsx"})`, ConfirmDanger: true})
|
||
if d.Error != "" {
|
||
t.Fatalf("deleteMany: %s", d.Error)
|
||
}
|
||
cnt := ExecSQL(ctx, ConnReq{Driver: "mongodb", Host: "127.0.0.1", Port: 27017, Database: "opsxe2e", SQL: `db.c1.count({"name":"opsx"})`})
|
||
if fmt.Sprintf("%v", cnt.Rows[0]["n"]) != "0" {
|
||
t.Fatalf("count after delete = %v", cnt.Rows[0]["n"])
|
||
}
|
||
}
|