195 lines
6.5 KiB
Go
195 lines
6.5 KiB
Go
package logstore
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.pcux.cn/xuans/Smail/config"
|
|
"git.pcux.cn/xuans/Smail/db/migration"
|
|
"git.pcux.cn/xuans/Smail/utils/logpolicy"
|
|
_ "modernc.org/sqlite"
|
|
)
|
|
|
|
func testStore(t *testing.T) *Store {
|
|
t.Helper()
|
|
database, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "logs.db"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
database.SetMaxOpenConns(1)
|
|
t.Cleanup(func() { _ = database.Close() })
|
|
runner := migration.Runner{DB: database, Dialect: migration.DialectSQLite, Migrations: migration.DefaultMigrations()}
|
|
if _, err := runner.Run(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
store, err := New(database, migration.DialectSQLite)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return store
|
|
}
|
|
|
|
func TestLogStoreEmptyEntriesAreNotInserted(t *testing.T) {
|
|
s := testStore(t)
|
|
ctx := context.Background()
|
|
for _, kind := range []string{"operation", "mail"} {
|
|
for _, payload := range []string{`{}`, `null`, `[]`, `" "`, `{"message":"***"}`} {
|
|
if err := s.Write(ctx, Entry{Kind: kind, CreatedAt: time.Now(), Action: "---", Payload: []byte(payload)}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
var count int
|
|
table, _ := logTable(kind)
|
|
if err := s.db.QueryRow("SELECT COUNT(*) FROM "+table+" WHERE kind = ?", kind).Scan(&count); err != nil || count != 0 {
|
|
t.Fatalf("空记录已落库：%s %d %v", kind, count, err)
|
|
}
|
|
for _, entry := range []Entry{
|
|
{Action: "登录验证", Result: "success", Payload: []byte(`{}`)},
|
|
{Action: "update", Result: "success", Payload: []byte(`{"changes":[]}`)},
|
|
{Result: "failure", Payload: []byte(`{}`)},
|
|
{Result: "unknown", Payload: []byte(`{}`)},
|
|
{Payload: []byte(`{"errorCode":0}`)},
|
|
{Payload: []byte(`{"message":"投递成功"}`)},
|
|
{Payload: []byte(`{"unknown":null}`)},
|
|
} {
|
|
entry.Kind, entry.CreatedAt = kind, time.Now()
|
|
if err := s.Write(ctx, entry); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
items, err := s.Query(ctx, Query{Kind: kind})
|
|
if err != nil || len(items) != 7 {
|
|
t.Fatalf("有效证据丢失：%s %d %v", kind, len(items), err)
|
|
}
|
|
}
|
|
if err := s.Write(ctx, Entry{Kind: "mail", CreatedAt: time.Now(), Payload: []byte(`invalid`)}); err != ErrInvalid {
|
|
t.Fatal("过滤不能掩盖非法载荷")
|
|
}
|
|
}
|
|
|
|
func TestLogRetentionBoundariesAndBusinessSafety(t *testing.T) {
|
|
s := testStore(t)
|
|
now := time.Date(2026, 9, 8, 9, 37, 0, 0, time.UTC)
|
|
s.now = func() time.Time { return now.Add(-logpolicy.Retention) }
|
|
cutoff := logpolicy.Cutoff(now)
|
|
ctx := context.Background()
|
|
if _, err := s.db.Exec("CREATE TABLE business_probe (value TEXT)"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := s.db.Exec("INSERT INTO business_probe VALUES ('keep')"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, kind := range []string{"operation", "mail"} {
|
|
for index := 0; index < 2005; index++ {
|
|
at := cutoff.Add(-time.Nanosecond)
|
|
if index >= 2002 {
|
|
at = cutoff.Add(time.Duration(index - 2002))
|
|
}
|
|
if err := s.Write(ctx, Entry{ID: fmt.Sprintf("%s-%d", kind, index), Kind: kind, CreatedAt: at, Result: "success", Payload: []byte(`{"message":"test"}`)}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
}
|
|
s.now = func() time.Time { return now }
|
|
for _, kind := range []string{"operation", "mail"} {
|
|
items, err := s.Query(ctx, Query{Kind: kind})
|
|
if err != nil || len(items) != 3 {
|
|
t.Fatalf("visible %s = %d, %v", kind, len(items), err)
|
|
}
|
|
}
|
|
deleted, err := s.Cleanup(ctx, now)
|
|
if err != nil || deleted != 4004 {
|
|
t.Fatalf("cleanup = %d, %v", deleted, err)
|
|
}
|
|
deleted, err = s.Cleanup(ctx, now)
|
|
if err != nil || deleted != 0 {
|
|
t.Fatalf("repeated cleanup = %d, %v", deleted, err)
|
|
}
|
|
var count int
|
|
if err := s.db.QueryRow("SELECT COUNT(*) FROM business_probe").Scan(&count); err != nil || count != 1 {
|
|
t.Fatal("business data changed")
|
|
}
|
|
if err := s.Write(ctx, Entry{Kind: "mail", CreatedAt: cutoff.Add(-1), Payload: []byte(`{}`)}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := s.db.QueryRow("SELECT COUNT(*) FROM mail_log").Scan(&count); err != nil || count != 3 {
|
|
t.Fatal("expired replay was inserted")
|
|
}
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
cancel()
|
|
if _, err := s.Cleanup(ctx, now); err == nil {
|
|
t.Fatal("cancelled cleanup succeeded")
|
|
}
|
|
}
|
|
|
|
func TestLogDialectStatementsAndRetiredAuditWriter(t *testing.T) {
|
|
s := testStore(t)
|
|
for _, dialect := range []migration.Dialect{migration.DialectSQLite, migration.DialectMySQL, migration.DialectPostgres} {
|
|
s.dialect = dialect
|
|
got := s.bind("SELECT id FROM mail_log WHERE created_at < ? LIMIT ?")
|
|
if dialect == migration.DialectPostgres {
|
|
if !strings.Contains(got, "$1 LIMIT $2") {
|
|
t.Fatal(got)
|
|
}
|
|
} else if strings.Count(got, "?") != 2 {
|
|
t.Fatal(got)
|
|
}
|
|
}
|
|
s.dialect = migration.DialectSQLite
|
|
audit := &AuditWriter{Store: s}
|
|
event, err := config.NormalizeConfigAuditEvent(config.ConfigAuditEvent{
|
|
Operation: config.ConfigOperationUpdate, Section: config.SectionRoot, ActorID: "1", ActorRole: "super_admin", ActorName: "tester",
|
|
FromGeneration: 1, ToGeneration: 2, CreatedAt: time.Now(), Result: "success",
|
|
Changes: []config.ConfigAuditChange{{Field: "root.logLevel", Before: "info", After: "debug"}},
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for i := 0; i < 2; i++ {
|
|
if err := audit.WriteConfigAudit(context.Background(), event); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
items, err := audit.Query(context.Background(), config.ConfigAuditQuery{})
|
|
if err != nil || len(items) != 0 {
|
|
t.Fatalf("已停用的配置审计仍写入数据库：%+v, %v", items, err)
|
|
}
|
|
if err := s.Write(context.Background(), Entry{Kind: "config", CreatedAt: time.Now(), Payload: []byte(`{}`)}); err != ErrInvalid {
|
|
t.Fatal("数据库写入入口必须拒绝配置审计")
|
|
}
|
|
}
|
|
|
|
func TestRetiredAuditImportPreservesLegacyFiles(t *testing.T) {
|
|
s := testStore(t)
|
|
root := t.TempDir()
|
|
directory := filepath.Join(root, "log")
|
|
if err := os.MkdirAll(directory, 0700); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
path := filepath.Join(directory, "config-audit-v2.jsonl")
|
|
history := []byte("旧审计内容不再解析或导入\n")
|
|
if err := os.WriteFile(path, history, 0600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
audit := &AuditWriter{Store: s}
|
|
for index := 0; index < 2; index++ {
|
|
if err := audit.ImportLegacy(context.Background(), root); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
data, err := os.ReadFile(path)
|
|
if err != nil || string(data) != string(history) {
|
|
t.Fatal("旧审计文件被修改")
|
|
}
|
|
var count int
|
|
if err := s.db.QueryRow("SELECT COUNT(*) FROM operation_log").Scan(&count); err != nil || count != 0 {
|
|
t.Fatal("已停用的旧审计导入仍写入数据库")
|
|
}
|
|
}
|