zzet--gortex/internal/contracts/topics_brokers_test.go

package contracts

import "testing"

// Per-broker producer / consumer detection coverage for the four
// supported message-broker families: Kafka, RabbitMQ, NATS, Redis.
// Each broker has a producer test, a consumer test, and a dynamic-
// topic suppression test. The contract-pairing tests live in
// topics_pairing_test.go.

// -- Kafka --------------------------------------------------------

func TestTopicExtractor_Kafka_Producer_KafkaGoWriter(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

import "github.com/segmentio/kafka-go"

func publish(w *kafka.Writer) {
	w.WriteMessages(ctx, kafka.Message{Topic: "orders.created", Value: payload})
}
`)
	contracts := ext.Extract("publisher.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 kafka producer contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "kafka", "orders.created", RoleProvider)
}

func TestTopicExtractor_Kafka_Producer_ConfluentProduce(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func publish(p *kafka.Producer) {
	p.Produce("user.created", payload)
}
`)
	contracts := ext.Extract("publisher.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 kafka producer contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "kafka", "user.created", RoleProvider)
}

func TestTopicExtractor_Kafka_Consumer_SubscribeTopics(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func consume(c *kafka.Consumer) {
	c.SubscribeTopics([]string{"user.created"}, nil)
}
`)
	contracts := ext.Extract("consumer.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 kafka consumer contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "kafka", "user.created", RoleConsumer)
}

func TestTopicExtractor_Kafka_Consumer_ReaderConfig(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

import "github.com/segmentio/kafka-go"

func make() *kafka.Reader {
	return kafka.NewReader(kafka.ReaderConfig{
		Brokers: []string{"localhost:9092"},
		Topic:   "events.audit",
	})
}
`)
	contracts := ext.Extract("reader.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 kafka reader-config consumer contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "kafka", "events.audit", RoleConsumer)
}

// Dynamic topic — passed as a variable, not a literal. The string-
// literal regex shouldn't fire; the contract pairer never sees it.
func TestTopicExtractor_Kafka_DynamicTopic_NotEmitted(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func publish(p *kafka.Producer, topic string) {
	p.Produce(topic, payload)
}
`)
	contracts := ext.Extract("publisher.go", src, nil, nil)
	if len(contracts) != 0 {
		t.Fatalf("expected 0 contracts for dynamic topic, got %d (%+v)", len(contracts), contracts)
	}
}

// -- RabbitMQ -----------------------------------------------------

func TestTopicExtractor_RabbitMQ_Producer_PublishWithContext(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

import amqp "github.com/rabbitmq/amqp091-go"

func publish(ch *amqp.Channel) {
	ch.PublishWithContext(ctx, "user.events", "user.created", false, false, amqp.Publishing{Body: payload})
}
`)
	contracts := ext.Extract("publisher.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 rabbitmq producer contract, got %d (%+v)", len(contracts), contracts)
	}
	// Exchange is the pairing identity; routing key is metadata.
	assertBroker(t, contracts[0], "rabbitmq", "user.events", RoleProvider)
}

func TestTopicExtractor_RabbitMQ_Producer_PublishExchange(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func publish(ch *amqp.Channel) {
	ch.Publish("orders", "order.placed", false, false, amqp.Publishing{Body: payload})
}
`)
	contracts := ext.Extract("publisher.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 rabbitmq producer contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "rabbitmq", "orders", RoleProvider)
}

func TestTopicExtractor_RabbitMQ_Consumer_Consume(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func consume(ch *amqp.Channel) {
	msgs, _ := ch.Consume("order-events", "consumer-1", true, false, false, false, nil)
	_ = msgs
}
`)
	contracts := ext.Extract("consumer.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 rabbitmq consumer contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "rabbitmq", "order-events", RoleConsumer)
}

func TestTopicExtractor_RabbitMQ_DynamicTopic_NotEmitted(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func consume(ch *amqp.Channel, queue string) {
	ch.Consume(queue, "consumer-1", true, false, false, false, nil)
}
`)
	contracts := ext.Extract("consumer.go", src, nil, nil)
	if len(contracts) != 0 {
		t.Fatalf("expected 0 contracts for dynamic queue, got %d (%+v)", len(contracts), contracts)
	}
}

// -- NATS ---------------------------------------------------------

func TestTopicExtractor_NATS_Producer_Publish(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

import "github.com/nats-io/nats.go"

func publish(nc *nats.Conn) {
	nc.Publish("orders.created", []byte("payload"))
}
`)
	contracts := ext.Extract("publisher.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 nats producer contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "nats", "orders.created", RoleProvider)
}

func TestTopicExtractor_NATS_Producer_PublishRequest(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func ask(nc *nats.Conn) {
	nc.PublishRequest("orders.created", "reply.box", payload)
}
`)
	contracts := ext.Extract("publisher.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 nats publish-request contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "nats", "orders.created", RoleProvider)
}

func TestTopicExtractor_NATS_Consumer_Subscribe(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func consume(nc *nats.Conn) {
	nc.Subscribe("orders.created", handler)
}
`)
	contracts := ext.Extract("consumer.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 nats consumer contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "nats", "orders.created", RoleConsumer)
}

func TestTopicExtractor_NATS_Consumer_QueueSubscribe(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func consume(nc *nats.Conn) {
	nc.QueueSubscribe("orders.created", "workers", handler)
}
`)
	contracts := ext.Extract("consumer.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 nats queue-subscribe contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "nats", "orders.created", RoleConsumer)
}

func TestTopicExtractor_NATS_DynamicTopic_NotEmitted(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func publish(nc *nats.Conn, subject string) {
	nc.Publish(subject, payload)
}
`)
	contracts := ext.Extract("publisher.go", src, nil, nil)
	if len(contracts) != 0 {
		t.Fatalf("expected 0 contracts for dynamic subject, got %d (%+v)", len(contracts), contracts)
	}
}

// -- Redis pub/sub ------------------------------------------------

func TestTopicExtractor_Redis_Producer_Publish(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

import "github.com/redis/go-redis/v9"

func publish(rdb *redis.Client) {
	rdb.Publish(ctx, "channel.users", "payload")
}
`)
	contracts := ext.Extract("publisher.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 redis producer contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "redis", "channel.users", RoleProvider)
}

func TestTopicExtractor_Redis_Consumer_Subscribe(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func consume(rdb *redis.Client) {
	rdb.Subscribe(ctx, "channel.users")
}
`)
	contracts := ext.Extract("consumer.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 redis consumer contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "redis", "channel.users", RoleConsumer)
}

func TestTopicExtractor_Redis_Consumer_PSubscribe(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func consume(rdb *redis.Client) {
	rdb.PSubscribe(ctx, "channel.*")
}
`)
	contracts := ext.Extract("consumer.go", src, nil, nil)
	if len(contracts) != 1 {
		t.Fatalf("expected 1 redis psubscribe contract, got %d (%+v)", len(contracts), contracts)
	}
	assertBroker(t, contracts[0], "redis", "channel.*", RoleConsumer)
}

func TestTopicExtractor_Redis_DynamicTopic_NotEmitted(t *testing.T) {
	ext := &TopicExtractor{}
	src := []byte(`
package main

func publish(rdb *redis.Client, ch string) {
	rdb.Publish(ctx, ch, "payload")
}
`)
	contracts := ext.Extract("publisher.go", src, nil, nil)
	if len(contracts) != 0 {
		t.Fatalf("expected 0 contracts for dynamic channel, got %d (%+v)", len(contracts), contracts)
	}
}

// assertBroker is a topic-specific assertion helper. It checks the
// contract's role, ID shape, broker meta, and raw topic name in one
// call so per-broker tests stay terse.
func assertBroker(t *testing.T, c Contract, broker, topic string, role Role) {
	t.Helper()
	wantID := "topic::" + broker + "::" + topic
	if c.ID != wantID {
		t.Errorf("contract ID: want %q, got %q", wantID, c.ID)
	}
	if c.Type != ContractTopic {
		t.Errorf("contract Type: want %q, got %q", ContractTopic, c.Type)
	}
	if c.Role != role {
		t.Errorf("contract Role: want %q, got %q", role, c.Role)
	}
	if got, _ := c.Meta["broker"].(string); got != broker {
		t.Errorf("Meta[broker]: want %q, got %q", broker, got)
	}
	if got, _ := c.Meta["topic"].(string); got != topic {
		t.Errorf("Meta[topic]: want %q, got %q", topic, got)
	}
}