//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"testing"

	"github.com/apache/cassandra-gocql-driver/v2/lz4"
	"github.com/apache/cassandra-gocql-driver/v2/snappy"
)

func TestProtocolV4Baseline(t *testing.T) {
	session := newSession(t, sessionOptions{disableHostLookup: true, protoVersion: 4})

	rows, err := scanRows(session, "SELECT key FROM system.local")
	if err != nil {
		t.Fatalf("query failed: %v", err)
	}
	if len(rows) == 0 {
		t.Fatalf("expected rows from system.local")
	}
}

func TestProtocolV5SegmentLZ4(t *testing.T) {
	session := newSession(t, sessionOptions{
		disableHostLookup: true,
		protoVersion:      5,
		compressor:        &lz4.LZ4Compressor{},
	})

	rows, err := scanRows(session, "SELECT key FROM system.local")
	if err != nil {
		t.Fatalf("query failed: %v", err)
	}
	if len(rows) == 0 {
		t.Fatalf("expected rows from system.local")
	}
}

func TestProtocolV4CompressionLZ4(t *testing.T) {
	session := newSession(t, sessionOptions{
		disableHostLookup: true,
		protoVersion:      4,
		compressor:        &lz4.LZ4Compressor{},
	})

	rows, err := scanRows(session, "SELECT key FROM system.local")
	if err != nil {
		t.Fatalf("query failed: %v", err)
	}
	if len(rows) == 0 {
		t.Fatalf("expected rows from system.local")
	}
}

func TestProtocolV4CompressionSnappy(t *testing.T) {
	session := newSession(t, sessionOptions{
		disableHostLookup: true,
		protoVersion:      4,
		compressor:        &snappy.SnappyCompressor{},
	})

	rows, err := scanRows(session, "SELECT key FROM system.local")
	if err != nil {
		t.Fatalf("query failed: %v", err)
	}
	if len(rows) == 0 {
		t.Fatalf("expected rows from system.local")
	}
}

func TestProtocolCompatibilityV3ToV5(t *testing.T) {
	setup := newSession(t, sessionOptions{disableHostLookup: true, protoVersion: 4})

	keyspace := uniqueKeyspace()
	if err := setup.Query(createKeyspaceCQL(keyspace)).Exec(); err != nil {
		t.Fatalf("create keyspace failed: %v", err)
	}
	t.Cleanup(func() {
		_ = setup.Query(dropKeyspaceCQL(keyspace)).Exec()
	})

	table := "protocol_compat"
	if err := setup.Query(createTableCQL(keyspace, table)).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	cases := []struct {
		name    string
		version int
	}{
		{name: "v3", version: 3},
		{name: "v4", version: 4},
		{name: "v5", version: 5},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			session := newSession(t, sessionOptions{disableHostLookup: true, protoVersion: tc.version})

			id := tc.version
			want := fmt.Sprintf("proto-%d", tc.version)
			insert := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)
			if err := session.Query(insert, id, want, id).Exec(); err != nil {
				t.Fatalf("insert failed: %v", err)
			}

			var got string
			selectStmt := fmt.Sprintf("SELECT name FROM %s.%s WHERE id = ?", keyspace, table)
			if err := session.Query(selectStmt, id).Scan(&got); err != nil {
				t.Fatalf("select failed: %v", err)
			}
			if got != want {
				t.Fatalf("unexpected name: %q", got)
			}
		})
	}
}

func TestProtocolAuthRequired(t *testing.T) {
	_, err := tryNewSession(t, sessionOptions{disableHostLookup: true, skipAuth: true})
	if err == nil {
		t.Fatalf("expected auth failure without credentials")
	}
}

func TestQueryKeyspaceFlagV5(t *testing.T) {
	session := newSession(t, sessionOptions{disableHostLookup: true, protoVersion: 5})

	keyspace := uniqueKeyspace()
	if err := session.Query(createKeyspaceCQL(keyspace)).Exec(); err != nil {
		t.Fatalf("create keyspace failed: %v", err)
	}
	t.Cleanup(func() {
		_ = session.Query(dropKeyspaceCQL(keyspace)).Exec()
	})

	table := "query_keyspace_flag"
	if err := session.Query(createTableCQL(keyspace, table)).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	insert := "INSERT INTO " + table + " (id, name, idx) VALUES (?, ?, ?)"
	if err := session.Query(insert, 1, "alpha", 1).SetKeyspace(keyspace).Exec(); err != nil {
		t.Fatalf("insert with keyspace flag failed: %v", err)
	}

	var name string
	selectStmt := "SELECT name FROM " + table + " WHERE id = ?"
	if err := session.Query(selectStmt, 1).SetKeyspace(keyspace).Scan(&name); err != nil {
		t.Fatalf("select with keyspace flag failed: %v", err)
	}
	if name != "alpha" {
		t.Fatalf("unexpected name: %q", name)
	}
}