package audit

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path"
	"strings"
	"testing"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/tidb/pkg/config"
	"github.com/pingcap/tidb/pkg/extension"
	"github.com/pingcap/tidb/pkg/kv"
	"github.com/pingcap/tidb/pkg/sessionctx"
	"github.com/pingcap/tidb/pkg/testkit"
	"github.com/pingcap/tidb/pkg/util/chunk"
	"github.com/pingcap/tidb/pkg/util/sqlexec"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"go.etcd.io/etcd/api/v3/mvccpb"
	clientv3 "go.etcd.io/etcd/client/v3"
	"gopkg.in/natefinch/lumberjack.v2"
)

func TestConfigLogFilePath(t *testing.T) {
	origGlobalCfg := config.GetGlobalConfig()
	defer config.StoreGlobalConfig(origGlobalCfg)
	globalCfg := config.NewConfig()
	config.StoreGlobalConfig(globalCfg)
	require.Empty(t, origGlobalCfg.Log.File.Filename)
	globalCfg.AdvertiseAddress = "192.168.123.211"
	globalCfg.Port = 4000

	wd, err := os.Getwd()
	require.NoError(t, err)
	require.True(t, path.IsAbs(wd))
	require.False(t, strings.HasSuffix(wd, "/"))
	m, err := NewAuditLogManager()
	require.NoError(t, err)
	defer m.Stop()

	checkLogPath := func(expected string) {
		require.Equal(t, expected, m.GetConfig().Filepath)
		//nolint:forcetypeassert
		require.Equal(t, expected, m.logger.writer.(*lumberjack.Logger).Filename)
		require.Equal(t, m.GetConfig(), m.logger.cfg)
	}

	// default case
	globalCfg.Log.File.Filename = ""
	require.Equal(t, wd+"/tidb-audit.log", m.GetConfig().Filepath)
	for _, c := range []struct {
		globalCfgFileName string
		setLogConfigPath  string
		expected          string
	}{
		{
			setLogConfigPath: "",
			expected:         wd + "/tidb-audit.log",
		},
		{
			setLogConfigPath: "audit2.log",
			expected:         wd + "/audit2.log",
		},
		{
			setLogConfigPath: "path/to/audit3.log",
			expected:         wd + "/path/to/audit3.log",
		},
		{
			setLogConfigPath: "./audit-4.log",
			expected:         wd + "/audit-4.log",
		},
		{
			setLogConfigPath: "/audit-5.log",
			expected:         "/audit-5.log",
		},
		{
			setLogConfigPath: "/path/to/audit-6.log",
			expected:         "/path/to/audit-6.log",
		},
		{
			setLogConfigPath: "/tmp/%e-audit.log",
			expected:         "/tmp/192-168-123-211-4000-audit.log",
		},
		{
			globalCfgFileName: "tidb.log",
			setLogConfigPath:  "audit11.log",
			expected:          wd + "/audit11.log",
		},
		{
			globalCfgFileName: "path1/to/tidb.log",
			setLogConfigPath:  "path2/to/audit12.log",
			expected:          wd + "/path1/to/path2/to/audit12.log",
		},
		{
			globalCfgFileName: "./path3/tidb.log",
			setLogConfigPath:  "./path4/audit13.log",
			expected:          wd + "/path3/path4/audit13.log",
		},
		{
			globalCfgFileName: "/abs/path/tidb.log",
			setLogConfigPath:  "audit14.log",
			expected:          "/abs/path/audit14.log",
		},
		{
			globalCfgFileName: "/abs/path/tidb.log",
			setLogConfigPath:  "path5/to/audit14.log",
			expected:          "/abs/path/path5/to/audit14.log",
		},
		{
			globalCfgFileName: "/abs/path/tidb.log",
			setLogConfigPath:  "/path6/to/audit14.log",
			expected:          "/path6/to/audit14.log",
		},
		{
			globalCfgFileName: "path7/tidb.log",
			setLogConfigPath:  "/abs/audit15.log",
			expected:          "/abs/audit15.log",
		},
		{
			globalCfgFileName: "/path8/%e/tidb.log",
			setLogConfigPath:  "audit16.log",
			expected:          "/path8/%e/audit16.log",
		},
	} {
		globalCfg.Log.File.Filename = c.globalCfgFileName
		// SetLogConfigPath with TEXT format
		require.NoError(t, m.SetLogConfigPath(c.setLogConfigPath))
		checkLogPath(c.expected)

		// Update to JSON format
		require.NoError(t, m.SetLogFormat(LogFormatJSON))
		checkLogPath(c.expected + ".json")

		// Update to another path with JSON format
		require.NoError(t, m.SetLogConfigPath("/tmp/abc/def/audit.log"))
		checkLogPath("/tmp/abc/def/audit.log.json")
		require.NoError(t, m.SetLogConfigPath(c.setLogConfigPath))
		checkLogPath(c.expected + ".json")

		// Set back to TEXT format
		require.NoError(t, m.SetLogFormat(LogFormatText))
		checkLogPath(c.expected)
	}
}

func setFn[T any](set any) func(any) error {
	return func(v any) error {
		//nolint:forcetypeassert
		return set.(func(T) error)(v.(T))
	}
}

func getFn[T any](get any) func() any {
	return func() any {
		//nolint:forcetypeassert
		return get.(func() T)()
	}
}

func TestManagerConfig(t *testing.T) {
	m1, err := NewAuditLogManager()
	require.NoError(t, err)
	defer m1.Stop()

	m2, err := NewAuditLogManager()
	require.NoError(t, err)
	defer m2.Stop()

	checkManagerConfigConsistent := func(m *LogManager) {
		require.Equal(t, m.GetConfig(), m.logger.GetConfig())
		writer, ok := m.logger.writer.(*lumberjack.Logger)
		require.True(t, ok)
		require.Equal(t, m.GetConfig().Filepath, writer.Filename)
		require.Equal(t, m.GetConfig().FileMaxSize, int64(writer.MaxSize))
		require.Equal(t, m.GetConfig().FileReservedDays, writer.MaxAge)
		require.Equal(t, m.GetConfig().FileReservedBackups, writer.MaxBackups)
	}

	for i, m := range []*LogManager{m1, m2} {
		checkManagerConfigConsistent(m)
		cases := []struct {
			name   string
			set    any
			get    any
			defVal any
			values []any
		}{
			{
				name:   "SetEnabled",
				set:    m.SetEnabled,
				get:    m.Enabled,
				defVal: false,
				values: []any{true, false},
			},
			{
				name:   "SetLogConfigPath",
				set:    m.SetLogConfigPath,
				get:    m.GetLogConfigPath,
				defVal: DefAuditLogName,
				values: []any{"audit1.log", "path/to/audit2.log", "/tmp/audit3.log"},
			},
			{
				name:   "SetLogFormat",
				set:    m.SetLogFormat,
				get:    m.GetLogFormat,
				defVal: LogFormatText,
				values: []any{LogFormatJSON, LogFormatText},
			},
			{
				name:   "SetFileMaxSize",
				set:    m.SetFileMaxSize,
				get:    m.GetFileMaxSize,
				defVal: DefAuditLogFileMaxSize,
				values: []any{int64(10), int64(50), int64(100), int64(200)},
			},
			{
				name:   "SetFileMaxLifetime",
				set:    m.SetFileMaxLifetime,
				get:    m.GetFileMaxLifetime,
				defVal: DefAuditLogFileMaxLifetime,
				values: []any{int64(500), int64(600), int64(10240)},
			},
			{
				name:   "SetFileReservedBackups",
				set:    m.SetFileReservedBackups,
				get:    m.GetFileReservedBackups,
				defVal: DefAuditLogFileReservedBackups,
				values: []any{1, 10, 100, 1000},
			},
			{
				name:   "SetFileReservedDays",
				set:    m.SetFileReservedDays,
				get:    m.GetFileReservedDays,
				defVal: DefAuditLogFileReservedDays,
				values: []any{1, 20, 30, 50},
			},
			{
				name:   "SetRedactLog",
				set:    m.SetRedactLog,
				get:    m.RedactLog,
				defVal: true,
				values: []any{false, true},
			},
			{
				name:   "SetFilter",
				set:    m.SetFilter,
				get:    m.GetFilter,
				defVal: (*LogFilterRuleBundle)(nil),
				values: []any{
					newLogFilterRuleBundle([]*logFilterRule{
						{
							user:       "u1@%",
							filterName: "f1",
							filter: &logFilter{
								Name: "f1",
							},
						},
					}),
					newLogFilterRuleBundle(nil),
				},
			},
		}

		for _, c := range cases {
			t.Run(fmt.Sprintf("m%d.%s", i+1, c.name), func(t *testing.T) {
				var set func(any) error
				var get func() any
				switch c.set.(type) {
				case func(bool) error:
					set = setFn[bool](c.set)
					get = getFn[bool](c.get)
				case func(string) error:
					set = setFn[string](c.set)
					get = getFn[string](c.get)
				case func(int64) error:
					set = setFn[int64](c.set)
					get = getFn[int64](c.get)
				case func(int) error:
					set = setFn[int](c.set)
					get = getFn[int](c.get)
				case func(*LogFilterRuleBundle) error:
					set = setFn[*LogFilterRuleBundle](c.set)
					get = getFn[*LogFilterRuleBundle](c.get)
				}

				// test the default value as expected after initialized.
				require.Equal(t, c.defVal, get())
				require.NotEmpty(t, c.values)
				for _, v := range c.values {
					require.NoError(t, set(v))
					require.Equal(t, v, get())
					checkManagerConfigConsistent(m)
				}
			})
		}
	}
}

type mockBootstrapContext struct {
	context.Context
	sessPool extension.SessionPool
}

func (c *mockBootstrapContext) ExecuteSQL(ctx context.Context, sql string) ([]chunk.Row, error) {
	r, err := c.sessPool.Get()
	if err != nil {
		return nil, err
	}

	sctx, ok := r.(sessionctx.Context)
	if !ok {
		return nil, errors.Errorf("type '%T' cannot be casted to 'sessionctx.Context'", sctx)
	}

	executor := sctx.GetSQLExecutor()
	ctx = kv.WithInternalSourceType(ctx, kv.InternalTxnBootstrap)
	rs, err := executor.ExecuteInternal(ctx, sql)
	if err != nil {
		return nil, err
	}

	if rs == nil {
		return nil, nil
	}

	defer func() {
		closeErr := rs.Close()
		if err == nil {
			err = closeErr
		}
	}()

	return sqlexec.DrainRecordSet(ctx, rs, 8)
}

// EtcdClient returns the etcd client
func (c *mockBootstrapContext) EtcdClient() *clientv3.Client {
	return nil
}

// SessionPool returns the session pool of domain
func (c *mockBootstrapContext) SessionPool() extension.SessionPool {
	return c.sessPool
}

type mockEtcdCli struct {
	mock.Mock
	EtcdCli
}

func (cli *mockEtcdCli) Put(ctx context.Context, key, val string, opts ...clientv3.OpOption) (*clientv3.PutResponse, error) {
	args := cli.Called(ctx, key, val, opts)
	if resp := args.Get(0); resp != nil {
		//nolint:forcetypeassert
		return resp.(*clientv3.PutResponse), args.Error(1)
	}
	return nil, args.Error(1)
}

func (cli *mockEtcdCli) Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan {
	//nolint:forcetypeassert
	return cli.Called(ctx, key, opts).Get(0).(clientv3.WatchChan)
}

func TestManagerBootstrap(t *testing.T) {
	store, do := testkit.CreateMockStoreAndDomain(t)
	do.SysSessionPool()

	m, err := NewAuditLogManager()
	require.NoError(t, err)

	tk := testkit.NewTestKit(t, store)
	tk.MustQuery("show tables in mysql like '%audit%'").Check(testkit.Rows())

	ctx := &mockBootstrapContext{
		Context:  context.Background(),
		sessPool: do.SysSessionPool(),
	}

	// Bootstrap should create system tables.
	require.NoError(t, m.Bootstrap(ctx))
	tk.MustQuery("select * from mysql.audit_log_filter_rules").Check(testkit.Rows())
	tk.MustQuery("select * from mysql.audit_log_filters").Check(testkit.Rows())
	m.Stop()

	// Bootstrap should load exist rules when bootstrap.
	filter1 := "{}"
	filter2 := `{"filter":[{"class":["QUERY"]}]}`
	tk.MustExec("insert into mysql.audit_log_filters values('f1', ?), ('f2', ?)", filter1, filter2)
	tk.MustExec("insert into mysql.audit_log_filter_rules values('u1@%', 'f1', 0), ('u2@127.0.0.1', 'f2', 1)")
	m, err = NewAuditLogManager()
	require.NoError(t, err)
	require.NoError(t, m.Bootstrap(ctx))
	rules := make(map[string]*logFilterRule)
	require.Equal(t, 2, len(m.logger.GetConfig().Filter.rules))
	for _, rule := range m.logger.GetConfig().Filter.rules {
		rules[rule.user+" "+rule.filterName] = rule
	}
	require.Equal(t, &logFilterRule{
		user:       "u1@%",
		filterName: "f1",
		enabled:    false,
		filter: &logFilter{
			Name: "f1",
		},
	}, rules["u1@% f1"])
	require.Equal(t, &logFilterRule{
		user:       "u2@127.0.0.1",
		filterName: "f2",
		enabled:    true,
		filter: &logFilter{
			Name: "f2",
			Filter: []filterSpec{
				{Classes: []string{"QUERY"}},
			},
		},
	}, rules["u2@127.0.0.1 f2"])
	m.Stop()
}

func TestManagerReloadFilter(t *testing.T) {
	store, do := testkit.CreateMockStoreAndDomain(t)
	tk := testkit.NewTestKit(t, store)

	m, err := NewAuditLogManager()
	require.NoError(t, err)
	require.NoError(t, m.Bootstrap(&mockBootstrapContext{
		Context:  context.Background(),
		sessPool: do.SysSessionPool(),
	}))
	defer m.Stop()

	tk.MustQuery("select * from mysql.audit_log_filters").Check(testkit.Rows())
	tk.MustQuery("select * from mysql.audit_log_filter_rules").Check(testkit.Rows())
	require.Empty(t, m.GetFilter().rules)
	require.Empty(t, m.GetFilter().ruleFuncs)

	tk.MustExec(
		"insert into mysql.audit_log_filters values('f1', '{}'), ('f2', ?), ('f3', ?), ('f4', '{}')",
		`{"filter":[{"class":["QUERY"]}]}`, `{"filter":[{"table":["test.*"]},{"class":[]}]}`,
	)
	tk.MustExec(
		"insert into mysql.audit_log_filter_rules values" +
			"('u1@%', 'f1', 0), " +
			"('u2@localhost', 'f2', 1)," +
			"('root@%', 'f1', 1)," +
			"('u1@%', 'f3', 1)",
	)
	require.NoError(t, m.ReloadFilter())
	filter := m.GetFilter()
	rules := filter.rules
	require.Equal(t, 4, len(rules))
	require.Equal(t, 4, len(m.GetFilter().ruleFuncs))
	findRule := func(rs []*logFilterRule, user, filter string) *logFilterRule {
		for _, rule := range rs {
			if rule.user == user && rule.filterName == filter {
				require.Equal(t, rule.filter.Name, filter)
				return rule
			}
		}
		require.FailNow(t, "cannot find rule", user, filter)
		return nil
	}

	r1 := findRule(rules, "u1@%", "f1")
	require.False(t, r1.enabled)
	require.Nil(t, r1.filter.Filter)

	r2 := findRule(rules, "u2@localhost", "f2")
	require.True(t, r2.enabled)
	require.Equal(t, []filterSpec{{Classes: []string{"QUERY"}}}, r2.filter.Filter)

	r3 := findRule(rules, "root@%", "f1")
	require.True(t, r3.enabled)
	require.Nil(t, r3.filter.Filter)

	r4 := findRule(rules, "u1@%", "f3")
	require.True(t, r4.enabled)
	require.Equal(t, []filterSpec{{Tables: []string{"test.*"}}, {Classes: []string{}}}, r4.filter.Filter)

	// Reload filter if no changes should not change the inner bundle
	require.NoError(t, m.ReloadFilter())
	require.Same(t, filter, m.GetFilter())

	// Test modify filter and then reload
	tk.MustExec(
		"update mysql.audit_log_filters set content = ? where filter_name = 'f2'",
		`{"filter":[{"class":["QUERY", "QUERY_DDL"]}]}`,
	)
	require.NoError(t, m.ReloadFilter())
	require.NotSame(t, filter, m.GetFilter())
	filter = m.GetFilter()
	rules = filter.rules
	// r2 updated
	r2New := findRule(rules, "u2@localhost", "f2")
	require.True(t, r2New.enabled)
	require.Equal(t, []filterSpec{{Classes: []string{"QUERY", "QUERY_DDL"}}}, r2New.filter.Filter)
	// old r2 not changed
	require.True(t, r2.enabled)
	require.Equal(t, []filterSpec{{Classes: []string{"QUERY"}}}, r2.filter.Filter)

	// Test modify enabled
	tk.MustExec("update mysql.audit_log_filter_rules set enabled=1 where user = 'u1@%' and filter_name = 'f1'")
	require.NoError(t, m.ReloadFilter())
	require.NotSame(t, filter, m.GetFilter())
	filter = m.GetFilter()
	rules = filter.rules
	r1New := findRule(rules, "u1@%", "f1")
	require.True(t, r1New.enabled)
	require.Nil(t, r1New.filter.Filter)

	// Test modify filter name
	tk.MustExec("update mysql.audit_log_filter_rules set filter_name='f2' where user = 'u1@%' and filter_name = 'f1'")
	require.NoError(t, m.ReloadFilter())
	require.NotSame(t, filter, m.GetFilter())
	filter = m.GetFilter()
	rules = filter.rules
	r1New = findRule(rules, "u1@%", "f2")
	require.True(t, r1New.enabled)
	require.Equal(t, []filterSpec{{Classes: []string{"QUERY", "QUERY_DDL"}}}, r1New.filter.Filter)
}

func TestManagerNotify(t *testing.T) {
	mockEtcd := &mockEtcdCli{}
	var mockReloadFilter mock.Mock
	injectReloadFilterForTest = func(err error) error {
		return mockReloadFilter.MethodCalled("ReloadFilter", err).Error(0)
	}
	mockWriter := &mockLogWriter{}
	injectWriterForTest = func(_ *log.Config, _ *lumberjack.Logger) LogWriter {
		return mockWriter
	}
	defer func() {
		injectedEtcdCliForTest = nil
		injectReloadFilterForTest = nil
		injectWriterForTest = nil
	}()

	isWithPrefixOpt := func(options []clientv3.OpOption) bool {
		if len(options) != 1 {
			return false
		}
		return clientv3.IsOptsWithPrefix(options)
	}

	noOpt := func(options []clientv3.OpOption) bool {
		return len(options) == 0
	}

	waitChanClosed := func(ch chan struct{}) {
		select {
		case <-ch:
		case <-time.After(time.Second):
			require.FailNow(t, "watch not called")
		}
	}

	validatePutPayLoad := func(v any) {
		s, ok := v.(string)
		require.True(t, ok)
		type Payload struct {
			Time int64 `json:"ts"`
		}
		var p Payload
		require.NoError(t, json.Unmarshal([]byte(s), &p))
		nowTS := time.Now().UnixMilli()
		log.Info(fmt.Sprintf("cmp %d vs %d", nowTS, p.Time))
		require.InDelta(t, nowTS, p.Time, 5000)
	}

	notify := func(ch chan<- clientv3.WatchResponse, key string) {
		select {
		case ch <- clientv3.WatchResponse{
			Events: []*clientv3.Event{
				{
					Type: clientv3.EventTypePut,
					Kv:   &mvccpb.KeyValue{Key: []byte(key)},
				},
			},
		}:
		case <-time.After(5 * time.Second):
			require.FailNow(t, "notify timeout")
		}
	}

	bootstrapCtx, cancelBootstrapCtx := context.WithCancel(context.Background())
	store, do := testkit.CreateMockStoreAndDomain(t)
	tk := testkit.NewTestKit(t, store)

	// m1 is used to do operations
	m1, err := NewAuditLogManager()
	require.NoError(t, err)
	defer m1.Stop()
	mockReloadFilter.On("ReloadFilter", nil).Return(nil).Once()
	require.NoError(t, m1.Bootstrap(&mockBootstrapContext{
		Context:  bootstrapCtx,
		sessPool: do.SysSessionPool(),
	}))
	m1.mu.etcdCli = mockEtcd
	m1.ctx = context.WithValue(m1.ctx, "forTest", "1")
	m1ChildCtx := func(ctx context.Context) bool {
		return ctx.Value("forTest") == "1"
	}
	mockEtcd.AssertExpectations(t)
	mockReloadFilter.AssertExpectations(t)

	// m2 is used to receive notifications
	watchChan := make(chan clientv3.WatchResponse)
	injectedEtcdCliForTest = mockEtcd
	m2, err := NewAuditLogManager()
	require.NoError(t, err)
	defer m2.Stop()
	watched := make(chan struct{})
	mockEtcd.On("Watch", m2.ctx, EtcdKeyNotifyPrefix, mock.MatchedBy(isWithPrefixOpt)).
		Return(clientv3.WatchChan(watchChan), nil).
		Once().
		Run(func(_ mock.Arguments) { close(watched) })
	mockReloadFilter.On("ReloadFilter", nil).Return(nil).Once()
	require.NoError(t, m2.Bootstrap(&mockBootstrapContext{
		Context:  bootstrapCtx,
		sessPool: do.SysSessionPool(),
	}))
	waitChanClosed(watched)
	mockEtcd.AssertExpectations(t)
	mockReloadFilter.AssertExpectations(t)

	defer func() {
		// when managers stop, some `Sync` and `Close` methods will be called.
		mockWriter.On("Sync").Return(nil).Maybe()
		mockWriter.On("Close").Return(nil).Maybe()
	}()

	// cancelBootstrapCtx after used
	cancelBootstrapCtx()

	testFilterUpdate := func(fn func()) {
		if fn != nil {
			mockEtcd.On(
				"Put", mock.MatchedBy(m1ChildCtx), EtcdKeyNotifyFilter, mock.Anything, mock.MatchedBy(noOpt),
			).Return(nil, nil).Once().Run(func(args mock.Arguments) {
				validatePutPayLoad(args.Get(2))
			})
			fn()
			mockEtcd.AssertExpectations(t)
		}

		filterReloaded := make(chan struct{})
		mockReloadFilter.On("ReloadFilter", nil).Return(nil).Once().
			Run(func(_ mock.Arguments) { close(filterReloaded) })
		notify(watchChan, EtcdKeyNotifyFilter)
		waitChanClosed(filterReloaded)
		mockEtcd.AssertExpectations(t)
		mockReloadFilter.AssertExpectations(t)
	}

	// create log filter
	tk.MustQuery("select * from mysql.audit_log_filters").Check(testkit.Rows())
	tk.MustQuery("select * from mysql.audit_log_filter_rules").Check(testkit.Rows())
	testFilterUpdate(func() {
		require.NoError(t, m1.CreateLogFilter(&logFilter{Name: "f1"}, false))
		tk.MustQuery("select FILTER_NAME, CONTENT from mysql.audit_log_filters").
			Check(testkit.Rows("f1 {}"))
		tk.MustQuery("select * from mysql.audit_log_filter_rules").Check(testkit.Rows())
	})
	testFilterUpdate(func() {
		require.NoError(t, m1.CreateLogFilter(&logFilter{Name: "f2", Filter: []filterSpec{
			{Classes: []string{"QUERY"}},
		}}, false))
		tk.MustQuery("select FILTER_NAME, CONTENT from mysql.audit_log_filters order by FILTER_NAME").
			Check(testkit.Rows("f1 {}", `f2 {"filter":[{"class":["QUERY"]}]}`))
		tk.MustQuery("select * from mysql.audit_log_filter_rules").Check(testkit.Rows())
	})

	// CreateLogFilterRule
	testFilterUpdate(func() {
		require.NoError(t, m1.CreateLogFilterRule("root@%", "f1", false))
		tk.MustQuery("select FILTER_NAME, CONTENT from mysql.audit_log_filters order by FILTER_NAME").
			Check(testkit.Rows("f1 {}", `f2 {"filter":[{"class":["QUERY"]}]}`))
		tk.MustQuery("select USER, FILTER_NAME, ENABLED from mysql.audit_log_filter_rules order by USER").
			Check(testkit.Rows("root@% f1 1"))
	})
	testFilterUpdate(func() {
		require.NoError(t, m1.CreateLogFilterRule("user1@%", "f2", false))
		tk.MustQuery("select FILTER_NAME, CONTENT from mysql.audit_log_filters order by FILTER_NAME").
			Check(testkit.Rows("f1 {}", `f2 {"filter":[{"class":["QUERY"]}]}`))
		tk.MustQuery("select USER, FILTER_NAME, ENABLED from mysql.audit_log_filter_rules order by USER").
			Check(testkit.Rows("root@% f1 1", "user1@% f2 1"))
	})

	// SwitchLogFilterRuleEnabled
	testFilterUpdate(func() {
		require.NoError(t, m1.SwitchLogFilterRuleEnabled("root@%", "f1", false))
		tk.MustQuery("select FILTER_NAME, CONTENT from mysql.audit_log_filters order by FILTER_NAME").
			Check(testkit.Rows("f1 {}", `f2 {"filter":[{"class":["QUERY"]}]}`))
		tk.MustQuery("select USER, FILTER_NAME, ENABLED from mysql.audit_log_filter_rules order by USER").
			Check(testkit.Rows("root@% f1 0", "user1@% f2 1"))
	})

	// RemoveLogFilterRule
	testFilterUpdate(func() {
		require.NoError(t, m1.RemoveLogFilterRule("root@%", "f1"))
		tk.MustQuery("select FILTER_NAME, CONTENT from mysql.audit_log_filters order by FILTER_NAME").
			Check(testkit.Rows("f1 {}", `f2 {"filter":[{"class":["QUERY"]}]}`))
		tk.MustQuery("select USER, FILTER_NAME, ENABLED from mysql.audit_log_filter_rules order by USER").
			Check(testkit.Rows("user1@% f2 1"))
	})

	// RemoveLogFilter
	testFilterUpdate(func() {
		require.NoError(t, m1.RemoveLogFilter("f1"))
		tk.MustQuery("select FILTER_NAME, CONTENT from mysql.audit_log_filters order by FILTER_NAME").
			Check(testkit.Rows(`f2 {"filter":[{"class":["QUERY"]}]}`))
		tk.MustQuery("select USER, FILTER_NAME, ENABLED from mysql.audit_log_filter_rules order by USER").
			Check(testkit.Rows("user1@% f2 1"))
	})

	// test ReloadFilter error should not break loop
	filterReloaded := make(chan struct{})
	mockReloadFilter.On("ReloadFilter", nil).Return(errors.New("mockCtx")).Once().
		Run(func(_ mock.Arguments) { close(filterReloaded) })
	notify(watchChan, EtcdKeyNotifyFilter)
	waitChanClosed(filterReloaded)
	// notify again, should also work
	testFilterUpdate(nil)

	// test rotate
	testRotate := func() {
		mockEtcd.On(
			"Put", mock.MatchedBy(m1ChildCtx), EtcdKeyNotifyRotate, mock.Anything, mock.MatchedBy(noOpt),
		).Return(nil, nil).Once().Run(func(args mock.Arguments) {
			validatePutPayLoad(args.Get(2))
		})
		require.NoError(t, m1.GlobalRotateLog())
		mockEtcd.AssertExpectations(t)

		rotated := make(chan struct{})
		mockWriter.On("Rotate").Return(nil).Once().
			Run(func(_ mock.Arguments) { close(rotated) })
		notify(watchChan, EtcdKeyNotifyRotate)
		waitChanClosed(rotated)
		mockWriter.AssertExpectations(t)
	}
	testRotate()

	// rotate error should not break loop
	rotated := make(chan struct{})
	mockWriter.On("Rotate").Return(errors.New("mockErr")).Once().
		Run(func(_ mock.Arguments) { close(rotated) })
	notify(watchChan, EtcdKeyNotifyRotate)
	waitChanClosed(rotated)
	mockWriter.AssertExpectations(t)
	testRotate()

	// test renew watch etcd ch
	watched = make(chan struct{})
	watchChan2 := make(chan clientv3.WatchResponse)
	mockEtcd.On("Watch", m2.ctx, EtcdKeyNotifyPrefix, mock.MatchedBy(isWithPrefixOpt)).
		Return(clientv3.WatchChan(watchChan2), nil).
		Once().
		Run(func(_ mock.Arguments) { close(watched) })
	close(watchChan)
	select {
	case <-watched:
		require.FailNow(t, "reWatch should not be called in MinEtcdReWatchInterval")
	case <-time.After(time.Second):
	}
	time.Sleep(MinEtcdReWatchInterval - time.Second)
	waitChanClosed(watched)
	watchChan = watchChan2
	// after renew watch chan, notify should still work
	testRotate()
}