package sqluser

import (
	"reflect"
	"testing"

	"github.com/pingcap/tidb/parser/ast"
	"github.com/smartystreets/goconvey/convey"
	"go.uber.org/mock/gomock"

	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/testdata"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterproxy/mock_clusterproxy"
)

func TestGetRoles(t *testing.T) {
	convey.Convey("TestGetRoles", t, func() {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		username := "user1"
		result := []byte(
			`{"Grants for user1@%":["R1JBTlQgVVNBR0UgT04gKi4qIFRPICd1c2VyMSdAJyUn","R1JBTlQgJ3JvbGVfbWVtYmVyJ0AnJScgVE8gJ3VzZXIxJ0AnJSc="]}`,
		)
		ctx := testdata.Ctx
		proxy := mock_clusterproxy.NewMockClusterProxy(ctrl)

		convey.Convey("should be ok if all is well", func() {
			sql := BuildShowRolesSql(username)
			proxy.EXPECT().QuerySQL(gomock.Any(), sql).Return(result, nil)

			r1, err := GetRoles(ctx, proxy, username)
			convey.So(err, convey.ShouldBeNil)
			convey.So(r1, convey.ShouldResemble, []string{"role_member"})
		})
	})
}

func TestParseSql(t *testing.T) {
	type args struct {
		sql string
	}
	tests := []struct {
		name    string
		args    args
		want    []ast.StmtNode
		wantErr bool
	}{
		{
			name:    "empty sql",
			args:    args{sql: ""},
			want:    nil,
			wantErr: false,
		},
		{
			name: "multiple sqls",
			args: args{sql: "select * from mysql.user; delete from d.t where f1='abc'"},
			want: []ast.StmtNode{
				&ast.SelectStmt{},
				&ast.DeleteStmt{},
			},
			wantErr: false,
		},
		{
			name: "single sql",
			args: args{sql: "select * from mysql.user"},
			want: []ast.StmtNode{
				&ast.SelectStmt{},
			},
			wantErr: false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := parseSql(tt.args.sql)
			if (err != nil) != tt.wantErr {
				t.Errorf("parseSql() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if len(got) != len(tt.want) {
				t.Errorf("expected len(want)=%d, got=%d", len(tt.want), len(got))
				return
			}
			for idx, gotStmtNode := range got {
				wantStmtNode := tt.want[idx]
				if reflect.TypeOf(gotStmtNode) != reflect.TypeOf(wantStmtNode) {
					t.Errorf(
						"parseSql() stmtNode type mismatch got = %v, want %v",
						reflect.TypeOf(gotStmtNode),
						reflect.TypeOf(wantStmtNode),
					)
				}
			}
		})
	}
}

func TestCheckGrants(t *testing.T) {
	type args struct {
		target string
		grants string
	}
	tests := []struct {
		name           string
		args           args
		raiseErr       bool
		errMsgContains string
	}{
		{
			name:           "single grants",
			args:           args{target: "r1", grants: "GRANT CREATE ON mysql.* TO r1"},
			raiseErr:       false,
			errMsgContains: "",
		},
		{
			name:           "multiple grants",
			args:           args{target: "r1", grants: "GRANT CREATE ON mysql.* TO r1; GRANT SELECT ON a.* TO r1; "},
			raiseErr:       false,
			errMsgContains: "",
		},
		{
			name:           "empty grants",
			args:           args{target: "r1", grants: ""},
			raiseErr:       false,
			errMsgContains: "",
		},
		{
			name:           "invalid sql",
			args:           args{target: "r1", grants: "abcd"},
			raiseErr:       true,
			errMsgContains: "invalid grant",
		},
		{
			name:           "the target of grant is mismatched",
			args:           args{target: "r1", grants: "GRANT CREATE ON *.* TO r2"},
			raiseErr:       true,
			errMsgContains: "invalid grantee",
		},
		{
			name:           "too many targets in one grant",
			args:           args{target: "r1", grants: "GRANT CREATE ON *.* TO r1,r2"},
			raiseErr:       true,
			errMsgContains: "too many",
		},
		{
			name:           "the sql is not grant statement",
			args:           args{target: "r1", grants: "select user TO mysq.user"},
			raiseErr:       true,
			errMsgContains: "invalid grant",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := CheckGrants(tt.args.target, tt.args.grants)
			if (got != nil) != tt.raiseErr {
				t.Errorf("checkGrants() = %v, want raiseErr: %t", got, tt.raiseErr)
			}
		})
	}
}

func TestCheckRevocation(t *testing.T) {
	type args struct {
		target     string
		revocation string
	}
	tests := []struct {
		name           string
		args           args
		raiseErr       bool
		errMsgContains string
	}{
		{
			name:           "single revocation",
			args:           args{target: "r1", revocation: "REVOKE CREATE ON mysql.* FROM r1"},
			raiseErr:       false,
			errMsgContains: "",
		},
		{
			name: "multiple revocation",
			args: args{
				target:     "r1",
				revocation: "REVOKE CREATE ON mysql.* FROM r1; REVOKE SELECT ON a.* FROM r1; ",
			},
			raiseErr:       false,
			errMsgContains: "",
		},
		{
			name:           "empty revocation",
			args:           args{target: "r1", revocation: ""},
			raiseErr:       false,
			errMsgContains: "",
		},
		{
			name:           "invalid sql",
			args:           args{target: "r1", revocation: "abcd"},
			raiseErr:       true,
			errMsgContains: "invalid revoke",
		},
		{
			name:           "the target of revocation is mismatched",
			args:           args{target: "r1", revocation: "REVOKE CREATE ON *.* FROM r2"},
			raiseErr:       true,
			errMsgContains: "invalid grantee",
		},
		{
			name:           "too many targets in one revocation",
			args:           args{target: "r1", revocation: "REVOKE CREATE ON *.* FROM r1,r2"},
			raiseErr:       true,
			errMsgContains: "too many",
		},
		{
			name:           "the sql is not revoke statement",
			args:           args{target: "r1", revocation: "select user FROM mysq.user"},
			raiseErr:       true,
			errMsgContains: "invalid revoke",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := CheckRevocation(tt.args.target, tt.args.revocation)
			if (got != nil) != tt.raiseErr {
				t.Errorf("checkRevocation() = %v, want raiseErr: %t", got, tt.raiseErr)
			}
		})
	}
}

func TestGetBuiltinRoles(t *testing.T) {
	type args struct {
		roles []string
	}
	tests := []struct {
		name string
		args args
		want []string
	}{
		{
			name: "empty roles",
			args: args{roles: nil},
			want: nil,
		},
		{
			name: "each role is custom role",
			args: args{roles: []string{"r1", "r2"}},
			want: nil,
		},
		{
			name: "include 'role_admin'",
			args: args{roles: []string{"r1", "role_admin", "r2"}},
			want: []string{BuiltinRoleNameAdmin},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := GetBuiltinRoles(tt.args.roles); !reflect.DeepEqual(got, tt.want) {
				t.Errorf("getBuiltinRoles() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestGetCustomRoles(t *testing.T) {
	type args struct {
		roles []string
	}
	tests := []struct {
		name string
		args args
		want []string
	}{
		{
			name: "empty roles",
			args: args{roles: nil},
			want: nil,
		},
		{
			name: "each role is builtin role",
			args: args{roles: []string{"role_admin", "role_cdc_admin"}},
			want: nil,
		},
		{
			name: "include 'role_admin'",
			args: args{roles: []string{"r1", "role_admin", "r2"}},
			want: []string{"r1", "r2"},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := GetCustomRoles(tt.args.roles); !reflect.DeepEqual(got, tt.want) {
				t.Errorf("getCustomRoles() = %v, want %v", got, tt.want)
			}
		})
	}
}

func Test_parseGrantAndRevokeSql(t *testing.T) {
	type args struct {
		grants     string
		revocation string
	}
	tests := []struct {
		name string
		args args
		want []string
	}{
		{
			name: "empty grants and revocation",
			args: args{grants: " ", revocation: " "},
			want: nil,
		},
		{
			name: "empty grants and revocation with ';'",
			args: args{grants: " ; ", revocation: " ; "},
			want: nil,
		},
		{
			name: "empty grants",
			args: args{grants: "", revocation: "revoke delete on *.* from 'u1'"},
			want: []string{"revoke delete on *.* from 'u1'"},
		},
		{
			name: "empty revocation",
			args: args{grants: "grant delete on *.* to 'u1'", revocation: ""},
			want: []string{"grant delete on *.* to 'u1'"},
		},
		{
			name: "both grant and revocation have leading and trailing white spaces",
			args: args{grants: "\t grant delete on *.* to 'u1' \t", revocation: "\t revoke delete on *.* from 'u1' \t"},
			want: []string{"grant delete on *.* to 'u1'", "revoke delete on *.* from 'u1'"},
		},
		{
			name: "multiple grants and revocation",
			args: args{
				grants:     "grant delete on *.* to 'u1'; grant update on *.* to 'u1';",
				revocation: "revoke delete on *.* from 'u1'; revoke update on *.* from 'u1'",
			},
			want: []string{
				"grant delete on *.* to 'u1'", "grant update on *.* to 'u1'",
				"revoke delete on *.* from 'u1'", "revoke update on *.* from 'u1'",
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := ParseGrantAndRevokeSql(tt.args.grants, tt.args.revocation); !reflect.DeepEqual(got, tt.want) {
				t.Errorf("parseGrantAndRevokeSql() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestStringifyOperator(t *testing.T) {
	testTable := []struct {
		name   string
		o      *globalserverlessv1.SqlUserOperator
		result string
	}{
		{
			name:   "should return '<nil>' if operator is nil",
			o:      nil,
			result: "<nil>",
		},
		{
			name: "normal case",
			o: &globalserverlessv1.SqlUserOperator{
				Username:   "u",
				AuthMethod: globalserverlessv1.SqlUser_AUTH_METHOD_TOKEN,
				Credential: "token",
			},
			result: "name=u,authMethod=AUTH_METHOD_TOKEN,len(credential)=5",
		},
	}

	for _, each := range testTable {
		t.Run(each.name, func(t *testing.T) {
			got := StringifyOperator(each.o)

			if got != each.result {
				t.Errorf("got: %s, want: %s", got, each.result)
			}
		})
	}
}

func TestParseRoles(t *testing.T) {
	testTable := []struct {
		name        string
		roles       []string
		builtinRole globalserverlessv1.SqlUser_BuiltinRole
		customRoles []string
	}{
		{
			name:        "should return empty",
			roles:       nil,
			builtinRole: globalserverlessv1.SqlUser_BUILTIN_ROLE_UNSPECIFIED,
			customRoles: nil,
		},
		{
			name:        "both builtin role and custom role",
			roles:       []string{BuiltinRoleNameAdmin, "c1"},
			builtinRole: globalserverlessv1.SqlUser_BUILTIN_ROLE_USER_ADMIN,
			customRoles: []string{"c1"},
		},
		{
			name:        "custom role only",
			roles:       []string{"c1"},
			builtinRole: globalserverlessv1.SqlUser_BUILTIN_ROLE_UNSPECIFIED,
			customRoles: []string{"c1"},
		},
	}

	for _, each := range testTable {
		t.Run(each.name, func(t *testing.T) {
			builtinRole, customRoles := ParseRoles(each.roles)

			if builtinRole != each.builtinRole {
				t.Errorf("builtin role, got: %s, want: %s", builtinRole, each.builtinRole)
			}
			if !reflect.DeepEqual(customRoles, each.customRoles) {
				t.Errorf("custom roles, got: %s, want: %s", customRoles, each.customRoles)
			}
		})
	}
}

func Test_GetBatchRoles(t *testing.T) {
	convey.Convey("Test_getBatchRoles", t, func() {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		usernames := []string{"user1", "user2"}
		sqlResultMap := map[string][]string{}
		sqlResultMap["user1"] = []string{"role_member", "role_admin"}
		sqlResultMap["user2"] = []string{"role_member"}

		result := []byte(
			`{"FROM_USER": ["cm9sZV9tZW1iZXI=", "cm9sZV9hZG1pbg==", "cm9sZV9tZW1iZXI="],"TO_USER": ["dXNlcjE=", "dXNlcjE=", "dXNlcjI="]}`,
		)
		ctx := t.Context()
		proxy := mock_clusterproxy.NewMockClusterProxy(ctrl)

		convey.Convey("should be ok if all is well", func() {
			sql := buildBatchShowRolesSql(usernames)
			proxy.EXPECT().QuerySQL(gomock.Any(), sql).Return(result, nil)

			resp, err := getBatchRoles(ctx, proxy, usernames)
			convey.So(err, convey.ShouldBeNil)
			convey.So(resp, convey.ShouldResemble, sqlResultMap)
		})
	})
}