package index
|
|
import (
|
"database/sql"
|
"os"
|
"path/filepath"
|
"testing"
|
)
|
|
// newV1Store 手工构建一个 v1 schema 的老库（模拟迁移前状态）
|
func newV1Store(t *testing.T) (*Store, string) {
|
t.Helper()
|
dbPath := filepath.Join(t.TempDir(), "kb-v1.db")
|
db, err := sql.Open("sqlite3", dbPath+"?_journal_mode=WAL")
|
if err != nil {
|
t.Fatal(err)
|
}
|
schema := `
|
CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
CREATE TABLE nodes (
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
path TEXT NOT NULL UNIQUE,
|
title TEXT, section TEXT, tags TEXT, entities TEXT, wikilinks TEXT,
|
content_fts TEXT,
|
created_at TEXT DEFAULT (datetime('now')),
|
updated_at TEXT DEFAULT (datetime('now')));
|
CREATE TABLE edges (
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
from_node INTEGER NOT NULL, to_node INTEGER NOT NULL,
|
relation TEXT NOT NULL, label TEXT,
|
UNIQUE(from_node, to_node, relation, label));
|
`
|
if _, err := db.Exec(schema); err != nil {
|
t.Fatal(err)
|
}
|
if _, err := db.Exec(`INSERT INTO nodes (path, title, tags, entities, wikilinks, content_fts)
|
VALUES ('FAQ/001-测试.md', '测试', '["a"]', '[]', '[]', '内容')`); err != nil {
|
t.Fatal(err)
|
}
|
return &Store{db: db, dbPath: dbPath}, dbPath
|
}
|
|
func TestMigrateV1ToV2(t *testing.T) {
|
store, dbPath := newV1Store(t)
|
defer store.Close()
|
|
// 迁移前备份文件必须存在
|
if err := store.migrate(); err != nil {
|
t.Fatalf("migrate: %v", err)
|
}
|
if _, err := os.Stat(dbPath + ".bak"); err != nil {
|
t.Fatalf("备份文件不存在: %v", err)
|
}
|
|
// v2 列必须存在
|
cols := store.nodeColumns(t)
|
for _, want := range []string{"aliases", "status", "size", "mtime", "content_hash"} {
|
if !cols[want] {
|
t.Errorf("nodes 缺列 %s", want)
|
}
|
}
|
if !store.edgeHasProvenance(t) {
|
t.Error("edges 缺 provenance 列")
|
}
|
if !store.tableExists(t, "unresolved_links") {
|
t.Error("缺 unresolved_links 表")
|
}
|
|
// 数据保留
|
var title string
|
if err := store.db.QueryRow(`SELECT title FROM nodes WHERE path='FAQ/001-测试.md'`).Scan(&title); err != nil {
|
t.Fatal(err)
|
}
|
if title != "测试" {
|
t.Errorf("迁移丢数据: %s", title)
|
}
|
}
|
|
func TestMigrateIdempotent(t *testing.T) {
|
store, _ := newV1Store(t)
|
defer store.Close()
|
if err := store.migrate(); err != nil {
|
t.Fatalf("第一次: %v", err)
|
}
|
if err := store.migrate(); err != nil {
|
t.Fatalf("第二次(幂等): %v", err)
|
}
|
}
|
|
func TestNewDBIsV2(t *testing.T) {
|
dbPath := filepath.Join(t.TempDir(), "kb-v2.db")
|
store, err := Open(dbPath)
|
if err != nil {
|
t.Fatal(err)
|
}
|
defer store.Close()
|
if !store.edgeHasProvenance(t) {
|
t.Error("新建库应为 v2 schema")
|
}
|
}
|
|
// 测试辅助：列检查
|
func (s *Store) nodeColumns(t *testing.T) map[string]bool {
|
t.Helper()
|
rows, err := s.db.Query(`SELECT name FROM pragma_table_info('nodes')`)
|
if err != nil {
|
t.Fatal(err)
|
}
|
defer rows.Close()
|
m := map[string]bool{}
|
for rows.Next() {
|
var name string
|
rows.Scan(&name)
|
m[name] = true
|
}
|
return m
|
}
|
|
func (s *Store) edgeHasProvenance(t *testing.T) bool {
|
t.Helper()
|
var n int
|
s.db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('edges') WHERE name='provenance'`).Scan(&n)
|
return n > 0
|
}
|
|
func (s *Store) tableExists(t *testing.T, name string) bool {
|
t.Helper()
|
var n int
|
s.db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?`, name).Scan(&n)
|
return n > 0
|
}
|
