pizzasql/pkg/capabilities/capabilities_test.go

package capabilities

import (
	"encoding/json"
	"strings"
	"testing"

	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
	"github.com/danfragoso/pizzasql-next/pkg/version"
)

func TestCurrentDocumentIsVersionedAndMarshalable(t *testing.T) {
	doc := Current()
	if doc.SchemaVersion != SchemaVersion {
		t.Fatalf("schema version = %d, want %d", doc.SchemaVersion, SchemaVersion)
	}
	if doc.Engine != "pizzasql" || doc.EngineVersion != version.String() {
		t.Fatalf("engine identity = %q/%q", doc.Engine, doc.EngineVersion)
	}

	body, err := json.Marshal(doc)
	if err != nil {
		t.Fatalf("marshal: %v", err)
	}
	var round Document
	if err := json.Unmarshal(body, &round); err != nil {
		t.Fatalf("unmarshal: %v", err)
	}
	if round.Constraints.Unique.SQLState != "23505" {
		t.Fatalf("unique SQLSTATE = %q, want 23505", round.Constraints.Unique.SQLState)
	}
	if round.Transactions.SerializationCode != "40001" {
		t.Fatalf("serialization SQLSTATE = %q, want 40001", round.Transactions.SerializationCode)
	}
	if round.ErrorCodes["0A000"] != "feature_not_supported" {
		t.Fatalf("0A000 mapping = %q", round.ErrorCodes["0A000"])
	}
	if !round.Notifications.Listen.Supported || !round.Notifications.Notify.Supported {
		t.Fatal("LISTEN/NOTIFY should be advertised as supported")
	}
	if round.Notifications.PayloadLimit != 8000 || round.Notifications.DurableReplay {
		t.Fatalf("unexpected notification limits: %+v", round.Notifications)
	}
	if !round.DDL.RenameColumnIndexes.Supported {
		t.Fatal("RENAME COLUMN must advertise index metadata rewriting")
	}
}

func TestJSONArrowOperatorsAreAdvertised(t *testing.T) {
	arrow := Current().JSON.ArrowOperators
	if !arrow.Supported {
		t.Fatal("-> and ->> should be advertised as supported")
	}
	if arrow.SQLState != "" {
		t.Fatalf("arrow operator SQLSTATE = %q, want none", arrow.SQLState)
	}
	if arrow.Status == "rejected" {
		t.Fatalf("arrow operator status = %q, want a supported status", arrow.Status)
	}
}

func TestTransactionalDDLIsAdvertised(t *testing.T) {
	ddl := Current().DDL
	if !ddl.Transactional.Supported {
		t.Fatalf("transactional DDL = %+v, want supported", ddl.Transactional)
	}
	if ddl.Transactional.Status != "staged_until_commit" {
		t.Fatalf("transactional DDL status = %q, want staged_until_commit", ddl.Transactional.Status)
	}
}

func TestAddColumnBackfillIsAdvertised(t *testing.T) {
	doc := Current()
	if !doc.DDL.AddColumnBackfill.Supported {
		t.Fatalf("add_column_backfill = %+v, want supported", doc.DDL.AddColumnBackfill)
	}
	if doc.DDL.AddColumnBackfill.Status != "rewrites_rows" {
		t.Fatalf("add_column_backfill status = %q, want rewrites_rows", doc.DDL.AddColumnBackfill.Status)
	}
	if doc.DDL.AddColumn.Status == "metadata_only" {
		t.Fatalf("add_column status = %q, but a DEFAULT rewrites rows", doc.DDL.AddColumn.Status)
	}
}

// The capability document must describe the current default semantics: constant
// and whitelisted clock defaults are supported, the clock ones are evaluated on
// write, and only the remaining non-constant forms are rejected.
func TestNonConstantDefaultsAreAdvertisedAsEvaluatedAtWriteTime(t *testing.T) {
	defaults := Current().Constraints.DefaultValues
	if !defaults.Supported {
		t.Fatalf("default_values = %+v, want supported", defaults)
	}
	if defaults.Status != "evaluated_at_write_time" {
		t.Fatalf("default_values status = %q, want evaluated_at_write_time", defaults.Status)
	}
	if defaults.SQLState != "0A000" {
		t.Fatalf("default_values SQLSTATE = %q, want 0A000 for the remaining rejected forms", defaults.SQLState)
	}
}

func TestCheckConstraintIsAdvertised(t *testing.T) {
	doc := Current()
	check := doc.Constraints.Check
	if !check.Supported {
		t.Fatalf("check = %+v, want supported", check)
	}
	if check.Status != "enforced" {
		t.Fatalf("check status = %q, want enforced", check.Status)
	}
	if check.SQLState != pgerror.CodeCheckViolation {
		t.Fatalf("check SQLSTATE = %q, want %q", check.SQLState, pgerror.CodeCheckViolation)
	}
	if check.Notes == "" {
		t.Fatal("check notes must describe the enforcement surface")
	}
	if got := doc.ErrorCodes[pgerror.CodeCheckViolation]; got != "check_violation" {
		t.Fatalf("23514 mapping = %q, want check_violation", got)
	}
}

func TestDDLNotesDescribeCheckDependencies(t *testing.T) {
	doc := Current()
	if !strings.Contains(doc.DDL.DropColumn.Notes, "CHECK") || !strings.Contains(doc.DDL.DropColumn.Notes, "0A000") {
		t.Fatalf("drop_column notes = %q, want the persisted CHECK restriction", doc.DDL.DropColumn.Notes)
	}
	if !strings.Contains(doc.DDL.RenameColumn.Notes, "CHECK") || !strings.Contains(doc.DDL.RenameColumn.Notes, "0A000") {
		t.Fatalf("rename_column notes = %q, want the persisted CHECK restriction", doc.DDL.RenameColumn.Notes)
	}
	if !strings.Contains(doc.DDL.AddColumnBackfill.Notes, "23514") {
		t.Fatalf("add_column_backfill notes = %q, want the CHECK backfill failure", doc.DDL.AddColumnBackfill.Notes)
	}
}

// Composite primary keys are advertised as supported, with the SQLSTATE callers
// should expect for a duplicate key.
func TestCompositePrimaryKeyIsAdvertised(t *testing.T) {
	doc := Current()
	composite := doc.Constraints.CompositePK
	if !composite.Supported {
		t.Fatalf("composite_primary_key = %+v, want supported", composite)
	}
	if composite.Status == "rejected" {
		t.Fatalf("composite_primary_key status = %q, want a supported status", composite.Status)
	}
	if composite.SQLState != "23505" {
		t.Fatalf("composite_primary_key SQLSTATE = %q, want 23505", composite.SQLState)
	}
	if !doc.Constraints.PrimaryKey.Supported {
		t.Fatalf("primary_key = %+v, want supported", doc.Constraints.PrimaryKey)
	}

	body, err := json.Marshal(doc)
	if err != nil {
		t.Fatalf("marshal: %v", err)
	}
	if !strings.Contains(string(body), `"composite_primary_key":{"supported":true`) {
		t.Fatalf("document does not advertise composite_primary_key as supported: %s", body)
	}
}

// Foreign keys are advertised as enforced, with the SQLSTATE callers should
// expect for a dangling reference, the referential actions that execute
// immediately, and the explicit exclusions for deferred enforcement.
func TestForeignKeyIsAdvertised(t *testing.T) {
	doc := Current()
	fk := doc.Constraints.ForeignKey
	if !fk.Supported {
		t.Fatalf("foreign_key = %+v, want supported", fk)
	}
	if fk.Status != "enforced" {
		t.Fatalf("foreign_key status = %q, want enforced", fk.Status)
	}
	if fk.SQLState != pgerror.CodeForeignKeyViolation {
		t.Fatalf("foreign_key SQLSTATE = %q, want %q", fk.SQLState, pgerror.CodeForeignKeyViolation)
	}
	for _, want := range []string{
		"UNIQUE", "COMMIT", "MATCH SIMPLE", "MATCH keyword", "self",
		"CASCADE", "SET NULL", "SET DEFAULT", "NOT NULL", "transitively",
		"autocommit", "no later than COMMIT", "statement savepoint", "hand-authored catalog",
		"O(FK component)", "DEFERRABLE", "0A000",
		"PRAGMA foreign_keys", "rejects OFF", "foreign_key_list", "database-wide foreign-key write gate",
		"DROP/RENAME COLUMN", "DROP TABLE", "pg_constraint",
	} {
		if !strings.Contains(fk.Notes, want) {
			t.Fatalf("foreign_key notes = %q, want it to mention %q", fk.Notes, want)
		}
	}
	if got := doc.ErrorCodes[pgerror.CodeForeignKeyViolation]; got != "foreign_key_violation" {
		t.Fatalf("23503 mapping = %q, want foreign_key_violation", got)
	}
}

func TestArraysAreAdvertised(t *testing.T) {
	doc := Current()
	types := doc.Types.Arrays
	if !types.Supported {
		t.Fatalf("types.arrays = %+v, want supported", types)
	}
	if types.Status != "one_dimensional" {
		t.Fatalf("types.arrays status = %q, want one_dimensional", types.Status)
	}
	for _, want := range []string{"TEXT[]", "INTEGER[]", "BOOLEAN[]", "ARRAY[", "array_length", "unnest", "0A000", "22P02"} {
		if !strings.Contains(types.Notes, want) {
			t.Errorf("types.arrays notes do not mention %q", want)
		}
	}

	params := doc.Parameters.Arrays
	if !params.Supported {
		t.Fatalf("parameters.arrays = %+v, want supported", params)
	}
	if params.Status != "text_and_binary" {
		t.Fatalf("parameters.arrays status = %q, want text_and_binary", params.Status)
	}
	for _, want := range []string{"1009", "1007", "1016", "1000", "1022", "1021", "1231", "1015", "0A000", "binary", "22P02"} {
		if !strings.Contains(params.Notes, want) {
			t.Errorf("parameters.arrays notes do not mention %q", want)
		}
	}
	if !strings.Contains(doc.Parameters.MaxOID, "text[]/varchar[]/int4[]") {
		t.Fatalf("parameters.max_oid = %q, want the array OIDs listed", doc.Parameters.MaxOID)
	}
}

// JSONB is advertised as a native type with precise canonicalization, operator,
// function, ordering, parameter, and rejection claims.
func TestJSONBIsAdvertised(t *testing.T) {
	doc := Current()
	types := doc.Types.JSONB
	if !types.Supported {
		t.Fatalf("types.jsonb = %+v, want supported", types)
	}
	if types.Status != "native" {
		t.Fatalf("types.jsonb status = %q, want native", types.Status)
	}
	for _, want := range []string{"canonical", "sorted", "last-wins", "idempotent", "OID 3802", "22P02", "0A000", "jsonpath", "GIN"} {
		if !strings.Contains(types.Notes, want) {
			t.Errorf("types.jsonb notes do not mention %q", want)
		}
	}
	if !doc.Types.JSONText {
		t.Fatal("JSON (text) remains supported and must stay advertised")
	}

	jsonb := doc.JSON.JSONB
	for _, want := range []string{"sorted", "last-wins", "1.50", "idempotent"} {
		if !strings.Contains(jsonb.Canonicalization, want) {
			t.Errorf("json.jsonb.canonicalization does not mention %q", want)
		}
	}
	for _, want := range []string{"exact canonical text bytes", "DISTINCT", "ORDER BY", "numeric"} {
		if !strings.Contains(jsonb.Ordering, want) {
			t.Errorf("json.jsonb.ordering does not mention %q", want)
		}
	}
	if !jsonb.Operators.Supported || jsonb.Operators.SQLState != pgerror.CodeDatatypeMismatch {
		t.Fatalf("json.jsonb.operators = %+v", jsonb.Operators)
	}
	for _, want := range []string{"@>", "<@", "?|", "?&", "#>", "#>>", "#-", "||", "scalar string never matches", "SQL NULL"} {
		if !strings.Contains(jsonb.Operators.Notes, want) {
			t.Errorf("json.jsonb.operators notes do not mention %q", want)
		}
	}
	if !jsonb.Functions.Supported || jsonb.Functions.SQLState != pgerror.CodeInvalidParameter {
		t.Fatalf("json.jsonb.functions = %+v", jsonb.Functions)
	}
	for _, want := range []string{"jsonb_typeof", "jsonb_array_length", "jsonb_strip_nulls", "jsonb_pretty", "jsonb_set", "jsonb_insert", "jsonb_build_object", "jsonb_build_array", "to_jsonb"} {
		if !strings.Contains(jsonb.Functions.Notes, want) {
			t.Errorf("json.jsonb.functions notes do not mention %q", want)
		}
	}
	if !jsonb.SetReturning.Supported || jsonb.SetReturning.Status != "single_from_item" {
		t.Fatalf("json.jsonb.set_returning = %+v", jsonb.SetReturning)
	}
	for _, want := range []string{"jsonb_each", "jsonb_array_elements", "jsonb_object_keys", "0A000"} {
		if !strings.Contains(jsonb.SetReturning.Notes, want) {
			t.Errorf("json.jsonb.set_returning notes do not mention %q", want)
		}
	}
	if !jsonb.Parameters.Supported || jsonb.Parameters.Status != "text_and_binary" {
		t.Fatalf("json.jsonb.parameters = %+v", jsonb.Parameters)
	}
	for _, want := range []string{"3802", "0x01", "pgx", "22P02"} {
		if !strings.Contains(jsonb.Parameters.Notes, want) {
			t.Fatalf("json.jsonb.parameters notes do not mention %q: %q", want, jsonb.Parameters.Notes)
		}
	}
	if jsonb.Rejected.Supported || jsonb.Rejected.SQLState != pgerror.CodeFeatureNotSupported {
		t.Fatalf("json.jsonb.rejected = %+v", jsonb.Rejected)
	}
	for _, want := range []string{"jsonpath", "GIN", "expression indexes", "wildcard"} {
		if !strings.Contains(jsonb.Rejected.Notes, want) {
			t.Errorf("json.jsonb.rejected notes do not mention %q", want)
		}
	}
	if !strings.Contains(doc.Parameters.MaxOID, "3802") {
		t.Fatalf("parameters.max_oid = %q, want OID 3802 listed", doc.Parameters.MaxOID)
	}
	if got := doc.ErrorCodes[pgerror.CodeInvalidParameter]; got != "invalid_parameter_value" {
		t.Fatalf("22023 mapping = %q, want invalid_parameter_value", got)
	}
}

// Binary result support is advertised with the encodable OIDs and the SQLSTATE
// for a type that has no binary encoder. pkg/pgserver covers each claim.
func TestBinaryResultsAreAdvertised(t *testing.T) {
	results := Current().Protocol.BinaryResults
	if !results.Supported {
		t.Fatalf("binary_results = %+v, want supported", results)
	}
	if results.Status != "supported" {
		t.Fatalf("binary_results status = %q, want supported", results.Status)
	}
	for _, claim := range []string{
		"bool (16)", "bytea (17)", "int8 (20)", "int2 (21)", "int4 (23)", "text (25)",
		"float4 (700)", "float8 (701)", "varchar (1043)", "numeric (1700) only as the element type of numeric[] (1231)", "jsonb (3802,",
		"1000, 1005, 1007, 1009, 1015, 1016, 1021, 1022, 1231", "0A000",
		"unrepresentable stored value", "22P02",
		"Describe and simple-query results stay text",
	} {
		if !strings.Contains(results.Notes, claim) {
			t.Errorf("binary_results notes do not mention %q", claim)
		}
	}
}

// Bounded Execute is advertised with its resume semantics, portal lifetime, and
// limits. pkg/pgserver covers each claim with raw-protocol portal tests.
func TestBoundedExecuteIsAdvertised(t *testing.T) {
	bounded := Current().Protocol.BoundedExecute
	if !bounded.Supported {
		t.Fatalf("bounded_execute = %+v, want supported", bounded)
	}
	if bounded.Status != "supported" {
		t.Fatalf("bounded_execute status = %q, want supported", bounded.Status)
	}
	if bounded.SQLState != "34000" {
		t.Fatalf("bounded_execute sqlstate = %q, want 34000", bounded.SQLState)
	}
	for _, claim := range []string{
		"PortalSuspended", "maxRows = 0", "materialized once", "no scroll",
		"WITH HOLD", "RowDescription is sent by the first Execute only",
		"Close('P')", "ROLLBACK TO SAVEPOINT", "42P03", "34000", "25P02", "08P01", "RETURNING",
	} {
		if !strings.Contains(bounded.Notes, claim) {
			t.Errorf("bounded_execute notes do not mention %q", claim)
		}
	}
}

// The catalog document advertises the read-only introspection surface an ORM
// uses: the relation and function matrix, the deterministic OID scheme, and the
// SET/SHOW compatibility surface. pkg/executor covers each relation.
func TestCatalogIntrospectionIsAdvertised(t *testing.T) {
	doc := Current()
	catalog := doc.Catalog
	if !catalog.PgCatalog.Supported || !catalog.InformationSchema.Supported {
		t.Fatalf("catalog = %+v, want both schemas supported", catalog)
	}
	if catalog.PgCatalog.SQLState != pgerror.CodeFeatureNotSupported {
		t.Fatalf("pg_catalog SQLSTATE = %q, want 0A000 for writes", catalog.PgCatalog.SQLState)
	}
	if catalog.Writable {
		t.Fatal("catalog relations are read-only")
	}
	for _, want := range []string{
		"pg_catalog.pg_class", "pg_catalog.pg_namespace", "pg_catalog.pg_attribute",
		"pg_catalog.pg_type", "pg_catalog.pg_index", "pg_catalog.pg_constraint",
		"pg_catalog.pg_attrdef", "pg_catalog.pg_database", "pg_catalog.pg_tablespace",
		"pg_catalog.pg_proc", "pg_catalog.pg_am", "pg_catalog.pg_description",
		"information_schema.tables", "information_schema.columns",
		"information_schema.table_constraints", "information_schema.key_column_usage",
		"information_schema.referential_constraints",
		"information_schema.constraint_column_usage", "information_schema.schemata",
		"information_schema.sequences", "information_schema.views",
	} {
		found := false
		for _, relation := range catalog.Relations {
			if relation == want {
				found = true
				break
			}
		}
		if !found {
			t.Errorf("catalog.relations does not list %q", want)
		}
	}
	for _, want := range []string{
		"current_schema", "current_database", "format_type", "pg_table_is_visible",
		"pg_get_indexdef", "pg_get_constraintdef", "pg_get_expr", "pg_get_serial_sequence",
		"obj_description", "col_description", "has_table_privilege",
	} {
		found := false
		for _, function := range catalog.Functions {
			if function == want {
				found = true
				break
			}
		}
		if !found {
			t.Errorf("catalog.functions does not list %q", want)
		}
	}
	if !strings.Contains(catalog.OIDStrategy, "FNV-1a") || !strings.Contains(catalog.OIDStrategy, "deterministic") {
		t.Fatalf("oid_strategy = %q", catalog.OIDStrategy)
	}
	if !catalog.SetStatements.Supported || catalog.SetStatements.Status != "accepted_and_ignored" {
		t.Fatalf("set_statements = %+v", catalog.SetStatements)
	}
	for _, want := range []string{"search_path", "SHOW", "current_schema"} {
		if !strings.Contains(catalog.SetStatements.Notes, want) {
			t.Errorf("set_statements notes do not mention %q", want)
		}
	}

	body, err := json.Marshal(doc)
	if err != nil {
		t.Fatalf("marshal: %v", err)
	}
	if !strings.Contains(string(body), `"writable":false`) {
		t.Fatalf("document does not advertise the catalog as read-only: %s", body)
	}
}