From 436f97f678df6268d3b32c9b4aac575ead049c36 Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Mon, 5 Jan 2026 14:34:03 +0100 Subject: [PATCH 01/14] add exchange declaration --- internal/queue/consumer.go | 7 +++- internal/queue/consumer_test.go | 3 ++ internal/queue/producer.go | 17 ++++++--- internal/queue/producer_test.go | 66 +++++++++++++++++++-------------- internal/queue/types.go | 55 +++++++++++++++------------ internal/queue/types_test.go | 4 ++ 6 files changed, 94 insertions(+), 58 deletions(-) diff --git a/internal/queue/consumer.go b/internal/queue/consumer.go index 5b160e4e..c38e8013 100644 --- a/internal/queue/consumer.go +++ b/internal/queue/consumer.go @@ -106,7 +106,12 @@ func (c *AmqpConsumer) ensureAmqpChannel() error { } if c.client.amqpChannel == nil || c.client.amqpChannel.IsClosed() { - err := c.client.resetChannel(c.queueName, false) + err := c.client.resetChannel() + if err != nil { + return err + } + + err = c.client.declareQueue(c.queueName) if err != nil { return err } diff --git a/internal/queue/consumer_test.go b/internal/queue/consumer_test.go index 627e678f..cd804d9f 100644 --- a/internal/queue/consumer_test.go +++ b/internal/queue/consumer_test.go @@ -14,6 +14,9 @@ import ( "github.com/stretchr/testify/assert" ) +const queueName = "test-queue" + +const queueWithDeclareError = "queue-with-declare-error" const queueWithConsumeError = "queue-with-consume-error" const queueWithQosError = "queue-with-qos-error" diff --git a/internal/queue/producer.go b/internal/queue/producer.go index 078d0058..376ea87d 100644 --- a/internal/queue/producer.go +++ b/internal/queue/producer.go @@ -40,7 +40,12 @@ func (p *AmqpProducer) resetConnectionOrChannelIfNecessary() error { } if p.client.amqpChannel == nil || p.client.amqpChannel.IsClosed() { - err := p.client.resetChannel(p.queueName, false) + err := p.client.resetChannel() + if err != nil { + return err + } + + err = p.client.declareExchange(p.exchangeName) if err != nil { return err } @@ -63,10 +68,10 @@ func waitForMsgConfirmation(ctx context.Context, confirmation confirmation) erro func (p *AmqpProducer) publishMsg(msg []byte, headers map[string]interface{}) (confirmation, error) { confirmation, err := p.client.amqpChannel.PublishWithDeferredConfirm( - "", // exchange - p.queueName, // routing key - false, // mandatory - false, // immediate + "", // exchange + p.exchangeName, // routing key + false, // mandatory + false, // immediate amqp.Publishing{ ContentType: "application/json", Body: msg, @@ -86,7 +91,7 @@ func NewAmqpProducer(uri string, queueName string) *AmqpProducer { uri: uri, connectFunc: amqpConnect, }, - queueName: queueName, + exchangeName: queueName, } } diff --git a/internal/queue/producer_test.go b/internal/queue/producer_test.go index 4339bfb0..55c22390 100644 --- a/internal/queue/producer_test.go +++ b/internal/queue/producer_test.go @@ -14,25 +14,28 @@ import ( "github.com/stretchr/testify/assert" ) -const queueName = "test-queue" -const queueWithPublishNoAck = "queue-with-publish-no-ack" -const queueWithPublishError = "queue-with-publish-error" -const queueWithPublishHangs = "queue-with-publish-hangs" -const queueWithDeclareError = "queue-with-declare-error" +const exchangeName = "test-exchange" +const exchangeWithDeclareError = "exchange-with-declare-error" +const exchangeWithPublishNoAck = "exchange-with-publish-no-ack" +const exchangeWithPublishError = "exchange-with-publish-error" +const exchangeWithPublishHangs = "exchange-with-publish-hangs" + +// TODO : Put this in a helper file because it is also used by consumer tests type MockAmqpChannel struct { msgs [][]byte headers []map[string]interface{} isclosed bool confirmMode bool queueName string + exchangeName string queueDurable bool confirmModeError bool } func (ch *MockAmqpChannel) PublishWithDeferredConfirm(_ string, key string, _, _ bool, msg amqp.Publishing) (confirmation, error) { - if key == queueWithPublishError { + if key == exchangeWithPublishError { return nil, errors.New("publish error") } ch.msgs = append(ch.msgs, msg.Body) @@ -40,11 +43,11 @@ func (ch *MockAmqpChannel) PublishWithDeferredConfirm(_ string, key string, _, _ done_ch := make(chan struct{}, 1) - if key != queueWithPublishHangs { + if key != exchangeWithPublishHangs { done_ch <- struct{}{} } - ack := key != queueWithPublishNoAck + ack := key != exchangeWithPublishNoAck confirmation := &MockConfirmation{ done: done_ch, ack: ack, @@ -64,6 +67,14 @@ func (ch *MockAmqpChannel) Confirm(_ bool) error { return nil } +func (ch *MockAmqpChannel) ExchangeDeclare(name string, _ string, _, _, _, _ bool, _ amqp.Table) error { + if name == exchangeWithDeclareError { + return errors.New("exchange declare error") + } + ch.exchangeName = name + return nil +} + func (ch *MockAmqpChannel) QueueDeclare(name string, durable, _, _, _ bool, _ amqp.Table) (amqp.Queue, error) { if name == queueWithDeclareError { return amqp.Queue{}, errors.New("queue declare error") @@ -145,7 +156,7 @@ func TestPush(t *testing.T) { amqpChannel: mockAmqpChannel, }, }, - queueName: queueName, + exchangeName: exchangeName, } headers := map[string]interface{}{"header1": "value1"} @@ -169,7 +180,7 @@ func TestPushFailure(t *testing.T) { }{ { name: "message is not acked", - queueName: queueWithPublishNoAck, + queueName: exchangeWithPublishNoAck, errMsg: "confirmation not acknowledged", context: context.Background(), }, @@ -180,12 +191,12 @@ func TestPushFailure(t *testing.T) { cancel() return ctx }(), - queueName: queueWithPublishHangs, + queueName: exchangeWithPublishHangs, errMsg: "context canceled", }, { name: "publish returns error", - queueName: queueWithPublishError, + queueName: exchangeWithPublishError, errMsg: "publish error", context: context.Background(), }, @@ -201,7 +212,7 @@ func TestPushFailure(t *testing.T) { amqpChannel: mockAmqpChannel, }, }, - queueName: tt.queueName, + exchangeName: tt.queueName, } err := amqpProducer.Push(tt.context, nil, []byte("TestMessage")) @@ -335,11 +346,11 @@ func TestPushNoConnectionFailure(t *testing.T) { assert.ErrorContains(t, err, "connection error") } -func TestPushQueueDeclare(t *testing.T) { +func TestPushExchangeDeclare(t *testing.T) { /* - arrange: create a queue with no amqp channel + arrange: create a producer with no amqp channel act: push a message to the queue - assert: channel with confirm mode is established and queue is declared + assert: channel with confirm mode is established and exchange is declared */ mockAmqpConnection := &MockAmqpConnection{} amqpProducer := &AmqpProducer{ @@ -347,37 +358,36 @@ func TestPushQueueDeclare(t *testing.T) { amqpChannel: nil, amqpConnection: mockAmqpConnection, }, - queueName: queueName, + exchangeName: exchangeName, } amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) assert.Equal(t, mockAmqpConnection.amqpChannel.confirmMode, true, "expected channel to be in confirm mode") - assert.Equal(t, mockAmqpConnection.amqpChannel.queueName, queueName, "expected queue name to be "+queueName) - assert.Equal(t, mockAmqpConnection.amqpChannel.queueDurable, true, "expected queue to be durable") + assert.Equal(t, mockAmqpConnection.amqpChannel.exchangeName, exchangeName, "expected exchange name to be "+exchangeName) } -func TestPushQueueDeclareFailure(t *testing.T) { +func TestPushExchangeDeclareFailure(t *testing.T) { /* - arrange: create a queue with no amqp channel where the queue declare fails + arrange: create a queue with no amqp channel where the exchange declare fails act: push a message to the queue assert: push returns an error */ tests := []struct { name string - queueName string + exchangeName string mockAmqpConnection *MockAmqpConnection errMsg string }{ { - name: "queue declare error", - queueName: queueWithDeclareError, + name: "exchange declare error", + exchangeName: exchangeWithDeclareError, mockAmqpConnection: &MockAmqpConnection{}, - errMsg: "queue declare error", + errMsg: "exchange declare error", }, { - name: "confirm error", - queueName: queueName, + name: "confirm error", + exchangeName: exchangeName, mockAmqpConnection: &MockAmqpConnection{ confirmModeError: true, }, @@ -392,7 +402,7 @@ func TestPushQueueDeclareFailure(t *testing.T) { amqpChannel: nil, amqpConnection: tt.mockAmqpConnection, }, - queueName: tt.queueName, + exchangeName: tt.exchangeName, } err := amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) diff --git a/internal/queue/types.go b/internal/queue/types.go index 63614750..4844be9c 100644 --- a/internal/queue/types.go +++ b/internal/queue/types.go @@ -34,8 +34,8 @@ func (c *Client) resetConnection() error { return nil } -// resetChannel opens a new AMQP channel and declares the queue. -func (c *Client) resetChannel(queue string, passive bool) error { +// resetChannel opens a new AMQP channel +func (c *Client) resetChannel() error { if c.amqpChannel != nil && !c.amqpChannel.IsClosed() { return nil } @@ -51,29 +51,37 @@ func (c *Client) resetChannel(queue string, passive bool) error { return fmt.Errorf("failed to put channel in confirm mode: %w", err) } - if passive { - _, err = ch.QueueDeclarePassive( - queue, // queueName - true, // durable - false, // delete when unused - false, // exclusive - false, // no-wait - nil, // arguments - ) - } else { - _, err = ch.QueueDeclare( - queue, // queueName - true, // durable - false, // delete when unused - false, // exclusive - false, // no-wait - nil, // arguments - ) + return nil +} + +func (c *Client) declareExchange(exchangeName string) error { + err := c.amqpChannel.ExchangeDeclare( + exchangeName, + "direct", // use direct exchange + true, // durable - don't delete on broker restart + false, // autoDelete - don't delete when no longer in use + false, // internal set to false - be suitable for publishing + false, // wait for confirmation + nil, // no additional arguments + ) + if err != nil { + return fmt.Errorf("failed to declare AMQP exchange: %w", err) } + return nil +} + +func (c *Client) declareQueue(queueName string) error { + _, err := c.amqpChannel.QueueDeclare( + queueName, + true, // durable + false, // delete when unused + false, // exclusive + false, // no-wait + nil, // arguments + ) if err != nil { return fmt.Errorf("failed to declare AMQP queue: %w", err) } - return nil } @@ -100,8 +108,8 @@ func (c *Client) Close() error { } type AmqpProducer struct { - client *Client - queueName string + client *Client + exchangeName string } type Producer interface { @@ -113,6 +121,7 @@ type amqpChannel interface { IsClosed() bool Close() error Confirm(noWait bool) error + ExchangeDeclare(name string, kind string, durable, autoDelete, internal, noWait bool, args amqp.Table) error QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) diff --git a/internal/queue/types_test.go b/internal/queue/types_test.go index c515bf71..2a8dc11f 100644 --- a/internal/queue/types_test.go +++ b/internal/queue/types_test.go @@ -38,6 +38,10 @@ func (f *fakeChannel) Close() error { return f.closeErr } func (f *fakeChannel) Confirm(noWait bool) error { return f.confirmErr } +func (f *fakeChannel) ExchangeDeclare(name string, kind string, durable, autoDelete, internal, noWait bool, args amqp.Table) error { + return nil +} + func (f *fakeChannel) QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) { return amqp.Queue{Name: name}, f.declareErr } From fc43cc68ba552b732f8c4d42372be1c3b2952810 Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Mon, 5 Jan 2026 14:58:44 +0100 Subject: [PATCH 02/14] add exchange declaration to consumer --- internal/queue/consumer.go | 17 ++++++++++++----- internal/queue/consumer_test.go | 29 ++++++++++++++++++++++------- 2 files changed, 34 insertions(+), 12 deletions(-) diff --git a/internal/queue/consumer.go b/internal/queue/consumer.go index c38e8013..e2cff9be 100644 --- a/internal/queue/consumer.go +++ b/internal/queue/consumer.go @@ -18,15 +18,17 @@ import ( // AmqpConsumer is an AMQP consumer for workflow job events. type AmqpConsumer struct { - client *Client - queueName string - channel <-chan amqp.Delivery + client *Client + exchangeName string + queueName string + channel <-chan amqp.Delivery } // NewAmqpConsumer creates a new AmqpConsumer with the given dependencies. -func NewAmqpConsumer(uri, queueName string) *AmqpConsumer { +func NewAmqpConsumer(uri, exchangeName string, queueName string) *AmqpConsumer { return &AmqpConsumer{ - queueName: queueName, + exchangeName: exchangeName, + queueName: queueName, client: &Client{ uri: uri, connectFunc: amqpConnect, @@ -111,6 +113,11 @@ func (c *AmqpConsumer) ensureAmqpChannel() error { return err } + err = c.client.declareExchange(c.exchangeName) + if err != nil { + return err + } + err = c.client.declareQueue(c.queueName) if err != nil { return err diff --git a/internal/queue/consumer_test.go b/internal/queue/consumer_test.go index cd804d9f..89e91521 100644 --- a/internal/queue/consumer_test.go +++ b/internal/queue/consumer_test.go @@ -343,7 +343,7 @@ func TestPullQueueDeclare(t *testing.T) { /* arrange: create a consumer with no amqp channel act: pull a message from the queue - assert: channel is established and queue is declared + assert: channel is established and exchange and queue is declared */ msgChannel := make(chan amqp.Delivery, 1) msgChannel <- amqp.Delivery{Body: []byte("TestMessage")} @@ -360,38 +360,50 @@ func TestPullQueueDeclare(t *testing.T) { amqpChannel: nil, amqpConnection: mockAmqpConnection, }, - queueName: queueName, + exchangeName: exchangeName, + queueName: queueName, } _, err := amqpConsumer.Pull(context.Background()) assert.NoError(t, err) assert.True(t, mockAmqpChannelConsumer.confirmMode, "expected channel to be in confirm mode") + assert.Equal(t, exchangeName, mockAmqpChannelConsumer.exchangeName, "expected exchange name to be "+exchangeName) assert.Equal(t, queueName, mockAmqpChannelConsumer.queueName, "expected queue name to be "+queueName) assert.True(t, mockAmqpChannelConsumer.queueDurable, "expected queue to be durable") } func TestPullQueueDeclareFailure(t *testing.T) { /* - arrange: create a consumer with no amqp channel where the queue declare fails + arrange: create a consumer with no amqp channel where a declaration fails act: pull a message from the queue assert: pull returns an error */ tests := []struct { name string + exchangeName string queueName string mockAmqpConnection *MockAmqpConnectionConsumer errMsg string }{ + { + name: "exchange declare error", + exchangeName: exchangeWithDeclareError, + queueName: queueName, + mockAmqpConnection: &MockAmqpConnectionConsumer{}, + errMsg: "exchange declare error", + }, { name: "queue declare error", + exchangeName: exchangeName, queueName: queueWithDeclareError, mockAmqpConnection: &MockAmqpConnectionConsumer{}, errMsg: "queue declare error", }, { - name: "confirm error", - queueName: queueName, + name: "confirm error", + exchangeName: exchangeWithDeclareError, + queueName: queueName, mockAmqpConnection: &MockAmqpConnectionConsumer{ confirmModeError: true, }, @@ -406,7 +418,8 @@ func TestPullQueueDeclareFailure(t *testing.T) { amqpChannel: nil, amqpConnection: tt.mockAmqpConnection, }, - queueName: tt.queueName, + exchangeName: tt.exchangeName, + queueName: tt.queueName, } _, err := amqpConsumer.Pull(context.Background()) @@ -477,11 +490,13 @@ func TestNewAmqpConsumer(t *testing.T) { assert: consumer is properly initialized */ uri := "amqp://guest:guest@localhost:5672/" + exchangeName := "test-exchange" queueName := "test-queue" - consumer := NewAmqpConsumer(uri, queueName) + consumer := NewAmqpConsumer(uri, exchangeName, queueName) assert.NotNil(t, consumer) + assert.Equal(t, exchangeName, consumer.exchangeName) assert.Equal(t, queueName, consumer.queueName) assert.NotNil(t, consumer.client) assert.Equal(t, uri, consumer.client.uri) From c658666b00bc9f4ce21e389f6fae0392353eb395 Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Tue, 6 Jan 2026 10:53:17 +0100 Subject: [PATCH 03/14] add queue bind --- internal/queue/consumer.go | 2 +- internal/queue/types.go | 5 +++++ internal/queue/types_test.go | 3 +++ 3 files changed, 9 insertions(+), 1 deletion(-) diff --git a/internal/queue/consumer.go b/internal/queue/consumer.go index e2cff9be..6b849e28 100644 --- a/internal/queue/consumer.go +++ b/internal/queue/consumer.go @@ -118,7 +118,7 @@ func (c *AmqpConsumer) ensureAmqpChannel() error { return err } - err = c.client.declareQueue(c.queueName) + err = c.client.bindQueue(c.config.QueueName, c.config.RoutingKey, c.config.ExchangeName) if err != nil { return err } diff --git a/internal/queue/types.go b/internal/queue/types.go index 4844be9c..a27643b7 100644 --- a/internal/queue/types.go +++ b/internal/queue/types.go @@ -124,6 +124,7 @@ type amqpChannel interface { ExchangeDeclare(name string, kind string, durable, autoDelete, internal, noWait bool, args amqp.Table) error QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) + QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) Qos(prefetchCount, prefetchSize int, global bool) error } @@ -164,6 +165,10 @@ func (ch *amqpChannelWrapper) Qos(prefetchCount, prefetchSize int, global bool) return ch.Channel.Qos(prefetchCount, prefetchSize, global) } +func (ch *amqpChannelWrapper) QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error { + return ch.Channel.QueueBind(name, key, exchange, noWait, args) +} + type Consumer interface { Pull(ctx context.Context) (amqp.Delivery, error) Close() error diff --git a/internal/queue/types_test.go b/internal/queue/types_test.go index 2a8dc11f..73560c0f 100644 --- a/internal/queue/types_test.go +++ b/internal/queue/types_test.go @@ -45,6 +45,9 @@ func (f *fakeChannel) ExchangeDeclare(name string, kind string, durable, autoDel func (f *fakeChannel) QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) { return amqp.Queue{Name: name}, f.declareErr } +func (f *fakeChannel) QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error { + return nil +} func (f *fakeChannel) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) { return amqp.Queue{Name: name}, f.declarePassErr } From 2420d3eefe1b5b95241059b45b449e897c4eef87 Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Tue, 6 Jan 2026 10:53:43 +0100 Subject: [PATCH 04/14] add DefaultQueueConfig --- cmd/planner/main.go | 4 +- cmd/webhook-gateway/main.go | 3 +- internal/planner/consumer.go | 4 +- internal/queue/config.go | 29 ++++++++ internal/queue/consumer.go | 33 +++++---- internal/queue/consumer_test.go | 104 +++++++++++++-------------- internal/queue/producer.go | 14 ++-- internal/queue/producer_test.go | 123 ++++++++++++++++++-------------- internal/queue/types.go | 18 ++++- internal/queue/types_test.go | 1 - 10 files changed, 197 insertions(+), 136 deletions(-) create mode 100644 internal/queue/config.go diff --git a/cmd/planner/main.go b/cmd/planner/main.go index bcfccb5c..ea8da28f 100644 --- a/cmd/planner/main.go +++ b/cmd/planner/main.go @@ -23,6 +23,7 @@ import ( "github.com/canonical/github-runner-operators/internal/database" "github.com/canonical/github-runner-operators/internal/planner" + "github.com/canonical/github-runner-operators/internal/queue" "github.com/canonical/github-runner-operators/internal/telemetry" "github.com/canonical/github-runner-operators/internal/version" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" @@ -33,7 +34,6 @@ const ( portEnvVar = "APP_PORT" serviceName = "github-runner-planner" rabbitMQUriEnvVar = "RABBITMQ_CONNECT_STRING" - queueName = "webhook-queue" shutdownTimeout = 30 * time.Second ) @@ -71,7 +71,7 @@ func main() { metrics := planner.NewMetrics(db) - consumer := planner.NewJobConsumer(rabbitMQUri, queueName, db, metrics) + consumer := planner.NewJobConsumer(rabbitMQUri, queue.DefaultQueueConfig(), db, metrics) var consumerWg sync.WaitGroup consumerWg.Add(1) diff --git a/cmd/webhook-gateway/main.go b/cmd/webhook-gateway/main.go index 50a6e23c..e378d35a 100644 --- a/cmd/webhook-gateway/main.go +++ b/cmd/webhook-gateway/main.go @@ -19,7 +19,6 @@ import ( "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" ) -const queueName = "webhook-queue" const webhookPath = "/webhook" const portEnvVar = "APP_PORT" const rabbitMQUriEnvVar = "RABBITMQ_CONNECT_STRING" @@ -51,7 +50,7 @@ func main() { log.Fatalln(webhookSecretEnvVar + " environment variable not set") } - p := queue.NewAmqpProducer(uri, queueName) + p := queue.NewAmqpProducer(uri, queue.DefaultQueueConfig()) handler := &webhook.Handler{ WebhookSecret: webhookSecret, Producer: p, diff --git a/internal/planner/consumer.go b/internal/planner/consumer.go index af64cccf..38ec97fc 100644 --- a/internal/planner/consumer.go +++ b/internal/planner/consumer.go @@ -32,9 +32,9 @@ const ( maxBackoff = 5 * time.Minute ) -func NewJobConsumer(amqpUri, queueName string, db JobDatabase, metrics *Metrics) *JobConsumer { +func NewJobConsumer(amqpUri string, queueConfig queue.QueueConfig, db JobDatabase, metrics *Metrics) *JobConsumer { return &JobConsumer{ - consumer: queue.NewAmqpConsumer(amqpUri, queueName), + consumer: queue.NewAmqpConsumer(amqpUri, queueConfig), db: db, metrics: metrics, } diff --git a/internal/queue/config.go b/internal/queue/config.go new file mode 100644 index 00000000..da334189 --- /dev/null +++ b/internal/queue/config.go @@ -0,0 +1,29 @@ +/* + * Copyright 2026 Canonical Ltd. + * See LICENSE file for licensing details. + */ + +package queue + +// Default AMQP configuration values. +const ( + defaultExchangeName = "github-workflow-jobs" + defaultQueueName = "github-workflow-jobs" + defaultRoutingKey = "workflow-job" +) + +// QueueConfig holds AMQP exchange, queue, and routing configuration. +type QueueConfig struct { + ExchangeName string + QueueName string + RoutingKey string +} + +// DefaultQueueConfig returns a QueueConfig with default values. +func DefaultQueueConfig() QueueConfig { + return QueueConfig{ + ExchangeName: defaultExchangeName, + QueueName: defaultQueueName, + RoutingKey: defaultRoutingKey, + } +} diff --git a/internal/queue/consumer.go b/internal/queue/consumer.go index 6b849e28..e96ab814 100644 --- a/internal/queue/consumer.go +++ b/internal/queue/consumer.go @@ -18,21 +18,19 @@ import ( // AmqpConsumer is an AMQP consumer for workflow job events. type AmqpConsumer struct { - client *Client - exchangeName string - queueName string - channel <-chan amqp.Delivery + client *Client + config QueueConfig + channel <-chan amqp.Delivery } // NewAmqpConsumer creates a new AmqpConsumer with the given dependencies. -func NewAmqpConsumer(uri, exchangeName string, queueName string) *AmqpConsumer { +func NewAmqpConsumer(uri string, config QueueConfig) *AmqpConsumer { return &AmqpConsumer{ - exchangeName: exchangeName, - queueName: queueName, client: &Client{ uri: uri, connectFunc: amqpConnect, }, + config: config, } } @@ -83,13 +81,13 @@ func (c *AmqpConsumer) ensureChannel() error { } channel, err := c.client.amqpChannel.Consume( - c.queueName, // queue - "", // consumer tag, empty string means a unique random tag will be generated - false, // whether rabbitmq auto-acknowledges messages - true, // whether only this consumer can access the queue - false, // no-local - false, // false means wait for server confirmation that consumer is registered - nil, // args + c.config.QueueName, // queue + "", // consumer tag, empty string means a unique random tag will be generated + false, // whether rabbitmq auto-acknowledges messages + true, // whether only this consumer can access the queue + false, // no-local + false, // false means wait for server confirmation that consumer is registered + nil, // args ) if err != nil { return fmt.Errorf("cannot consume amqp messages: %w", err) @@ -113,7 +111,12 @@ func (c *AmqpConsumer) ensureAmqpChannel() error { return err } - err = c.client.declareExchange(c.exchangeName) + err = c.client.declareExchange(c.config.ExchangeName) + if err != nil { + return err + } + + err = c.client.declareQueue(c.config.QueueName) if err != nil { return err } diff --git a/internal/queue/consumer_test.go b/internal/queue/consumer_test.go index 89e91521..a5945c4e 100644 --- a/internal/queue/consumer_test.go +++ b/internal/queue/consumer_test.go @@ -1,5 +1,5 @@ /* - * Copyright 2025 Canonical Ltd. + * Copyright 2026 Canonical Ltd. * See LICENSE file for licensing details. */ @@ -14,12 +14,6 @@ import ( "github.com/stretchr/testify/assert" ) -const queueName = "test-queue" - -const queueWithDeclareError = "queue-with-declare-error" -const queueWithConsumeError = "queue-with-consume-error" -const queueWithQosError = "queue-with-qos-error" - type MockAmqpChannelConsumer struct { MockAmqpChannel msgChannel chan amqp.Delivery @@ -30,10 +24,12 @@ type MockAmqpChannelConsumer struct { qosCalls int consumerTag string consumerAutoAck bool + consumeError bool + qosError bool } func (ch *MockAmqpChannelConsumer) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) { - if queue == queueWithConsumeError { + if ch.consumeError { return nil, errors.New("consume error") } ch.consumeCalls++ @@ -46,7 +42,7 @@ func (ch *MockAmqpChannelConsumer) Consume(queue, consumer string, autoAck, excl } func (ch *MockAmqpChannelConsumer) Qos(prefetchCount, prefetchSize int, global bool) error { - if ch.queueName == queueWithQosError { + if ch.qosError { return errors.New("qos error") } ch.qosCalls++ @@ -62,6 +58,9 @@ type MockAmqpConnectionConsumer struct { isclosed bool errMode bool confirmModeError bool + exchangeDeclareError bool + queueDeclareError bool + bindError bool returnChannelOnSetup bool } @@ -75,7 +74,10 @@ func (m *MockAmqpConnectionConsumer) Channel() (amqpChannel, error) { } m.amqpChannelConsumer = &MockAmqpChannelConsumer{ MockAmqpChannel: MockAmqpChannel{ - confirmModeError: m.confirmModeError, + confirmModeError: m.confirmModeError, + exchangeDeclareError: m.exchangeDeclareError, + queueDeclareError: m.queueDeclareError, + bindError: m.bindError, }, } return m.amqpChannelConsumer, nil @@ -115,8 +117,8 @@ func TestPull(t *testing.T) { amqpChannelConsumer: mockAmqpChannel, }, }, - queueName: queueName, - channel: msgChannel, + config: DefaultQueueConfig(), + channel: msgChannel, } msg, err := amqpConsumer.Pull(context.Background()) @@ -146,8 +148,8 @@ func TestPullContextCanceled(t *testing.T) { amqpChannelConsumer: mockAmqpChannel, }, }, - queueName: queueName, - channel: msgChannel, + config: DefaultQueueConfig(), + channel: msgChannel, } _, err := amqpConsumer.Pull(ctx) @@ -175,8 +177,8 @@ func TestPullChannelClosed(t *testing.T) { amqpChannelConsumer: mockAmqpChannel, }, }, - queueName: queueName, - channel: msgChannel, + config: DefaultQueueConfig(), + channel: msgChannel, } _, err := amqpConsumer.Pull(context.Background()) @@ -225,7 +227,7 @@ func TestPullNoChannel(t *testing.T) { amqpChannel: tt.channel, amqpConnection: mockAmqpConnection, }, - queueName: queueName, + config: DefaultQueueConfig(), } msg, err := amqpConsumer.Pull(context.Background()) @@ -255,7 +257,7 @@ func TestPullNoChannelFailure(t *testing.T) { amqpChannel: nil, amqpConnection: mockAmqpConnection, }, - queueName: queueName, + config: DefaultQueueConfig(), } _, err := amqpConsumer.Pull(context.Background()) @@ -304,7 +306,7 @@ func TestPullNoConnection(t *testing.T) { return mockAmqpConnection, nil }, }, - queueName: queueName, + config: DefaultQueueConfig(), } msg, err := amqpConsumer.Pull(context.Background()) @@ -330,7 +332,7 @@ func TestPullNoConnectionFailure(t *testing.T) { return nil, errors.New("connection error") }, }, - queueName: queueName, + config: DefaultQueueConfig(), } _, err := amqpConsumer.Pull(context.Background()) @@ -339,11 +341,11 @@ func TestPullNoConnectionFailure(t *testing.T) { assert.ErrorContains(t, err, "connection error") } -func TestPullQueueDeclare(t *testing.T) { +func TestPullQueueDeclareAndBind(t *testing.T) { /* arrange: create a consumer with no amqp channel act: pull a message from the queue - assert: channel is established and exchange and queue is declared + assert: channel is established, exchange and queue declared, and queue bound */ msgChannel := make(chan amqp.Delivery, 1) msgChannel <- amqp.Delivery{Body: []byte("TestMessage")} @@ -355,25 +357,28 @@ func TestPullQueueDeclare(t *testing.T) { amqpChannelConsumer: mockAmqpChannelConsumer, returnChannelOnSetup: true, } + config := DefaultQueueConfig() amqpConsumer := &AmqpConsumer{ client: &Client{ amqpChannel: nil, amqpConnection: mockAmqpConnection, }, - exchangeName: exchangeName, - queueName: queueName, + config: config, } _, err := amqpConsumer.Pull(context.Background()) assert.NoError(t, err) assert.True(t, mockAmqpChannelConsumer.confirmMode, "expected channel to be in confirm mode") - assert.Equal(t, exchangeName, mockAmqpChannelConsumer.exchangeName, "expected exchange name to be "+exchangeName) - assert.Equal(t, queueName, mockAmqpChannelConsumer.queueName, "expected queue name to be "+queueName) + assert.Equal(t, config.ExchangeName, mockAmqpChannelConsumer.exchangeName, "expected exchange name to be "+config.ExchangeName) + assert.Equal(t, config.QueueName, mockAmqpChannelConsumer.queueName, "expected queue name to be "+config.QueueName) assert.True(t, mockAmqpChannelConsumer.queueDurable, "expected queue to be durable") + assert.Equal(t, config.QueueName, mockAmqpChannelConsumer.boundQueue, "expected queue to be bound") + assert.Equal(t, config.ExchangeName, mockAmqpChannelConsumer.boundExchange, "expected queue to be bound to exchange") + assert.Equal(t, config.RoutingKey, mockAmqpChannelConsumer.boundRoutingKey, "expected routing key to be "+config.RoutingKey) } -func TestPullQueueDeclareFailure(t *testing.T) { +func TestPullDeclarationFailure(t *testing.T) { /* arrange: create a consumer with no amqp channel where a declaration fails act: pull a message from the queue @@ -381,29 +386,26 @@ func TestPullQueueDeclareFailure(t *testing.T) { */ tests := []struct { name string - exchangeName string - queueName string mockAmqpConnection *MockAmqpConnectionConsumer errMsg string }{ { name: "exchange declare error", - exchangeName: exchangeWithDeclareError, - queueName: queueName, - mockAmqpConnection: &MockAmqpConnectionConsumer{}, + mockAmqpConnection: &MockAmqpConnectionConsumer{exchangeDeclareError: true}, errMsg: "exchange declare error", }, { name: "queue declare error", - exchangeName: exchangeName, - queueName: queueWithDeclareError, - mockAmqpConnection: &MockAmqpConnectionConsumer{}, + mockAmqpConnection: &MockAmqpConnectionConsumer{queueDeclareError: true}, errMsg: "queue declare error", }, { - name: "confirm error", - exchangeName: exchangeWithDeclareError, - queueName: queueName, + name: "queue bind error", + mockAmqpConnection: &MockAmqpConnectionConsumer{bindError: true}, + errMsg: "bind error", + }, + { + name: "confirm error", mockAmqpConnection: &MockAmqpConnectionConsumer{ confirmModeError: true, }, @@ -418,13 +420,12 @@ func TestPullQueueDeclareFailure(t *testing.T) { amqpChannel: nil, amqpConnection: tt.mockAmqpConnection, }, - exchangeName: tt.exchangeName, - queueName: tt.queueName, + config: DefaultQueueConfig(), } _, err := amqpConsumer.Pull(context.Background()) - assert.Error(t, err, "expected error when queue declare fails") + assert.Error(t, err, "expected error when declaration fails") assert.ErrorContains(t, err, tt.errMsg) }) } @@ -437,9 +438,7 @@ func TestPullQosFailure(t *testing.T) { assert: pull returns an error */ mockAmqpChannelConsumer := &MockAmqpChannelConsumer{ - MockAmqpChannel: MockAmqpChannel{ - queueName: queueWithQosError, - }, + qosError: true, } mockAmqpConnection := &MockAmqpConnectionConsumer{ amqpChannelConsumer: mockAmqpChannelConsumer, @@ -450,7 +449,7 @@ func TestPullQosFailure(t *testing.T) { amqpChannel: nil, amqpConnection: mockAmqpConnection, }, - queueName: queueWithQosError, + config: DefaultQueueConfig(), } _, err := amqpConsumer.Pull(context.Background()) @@ -465,16 +464,19 @@ func TestPullConsumeFailure(t *testing.T) { act: pull a message from the queue assert: pull returns an error */ + mockAmqpChannelConsumer := &MockAmqpChannelConsumer{ + consumeError: true, + } mockAmqpConnection := &MockAmqpConnectionConsumer{ returnChannelOnSetup: true, - amqpChannelConsumer: &MockAmqpChannelConsumer{}, + amqpChannelConsumer: mockAmqpChannelConsumer, } amqpConsumer := &AmqpConsumer{ client: &Client{ amqpChannel: nil, amqpConnection: mockAmqpConnection, }, - queueName: queueWithConsumeError, + config: DefaultQueueConfig(), } _, err := amqpConsumer.Pull(context.Background()) @@ -490,15 +492,13 @@ func TestNewAmqpConsumer(t *testing.T) { assert: consumer is properly initialized */ uri := "amqp://guest:guest@localhost:5672/" - exchangeName := "test-exchange" - queueName := "test-queue" + config := DefaultQueueConfig() - consumer := NewAmqpConsumer(uri, exchangeName, queueName) + consumer := NewAmqpConsumer(uri, config) assert.NotNil(t, consumer) - assert.Equal(t, exchangeName, consumer.exchangeName) - assert.Equal(t, queueName, consumer.queueName) assert.NotNil(t, consumer.client) assert.Equal(t, uri, consumer.client.uri) assert.NotNil(t, consumer.client.connectFunc) + assert.Equal(t, config, consumer.config) } diff --git a/internal/queue/producer.go b/internal/queue/producer.go index 376ea87d..cfdda8c8 100644 --- a/internal/queue/producer.go +++ b/internal/queue/producer.go @@ -45,7 +45,7 @@ func (p *AmqpProducer) resetConnectionOrChannelIfNecessary() error { return err } - err = p.client.declareExchange(p.exchangeName) + err = p.client.declareExchange(p.config.ExchangeName) if err != nil { return err } @@ -68,10 +68,10 @@ func waitForMsgConfirmation(ctx context.Context, confirmation confirmation) erro func (p *AmqpProducer) publishMsg(msg []byte, headers map[string]interface{}) (confirmation, error) { confirmation, err := p.client.amqpChannel.PublishWithDeferredConfirm( - "", // exchange - p.exchangeName, // routing key - false, // mandatory - false, // immediate + p.config.ExchangeName, // exchange + p.config.RoutingKey, // routing key + false, // mandatory + false, // immediate amqp.Publishing{ ContentType: "application/json", Body: msg, @@ -85,13 +85,13 @@ func (p *AmqpProducer) publishMsg(msg []byte, headers map[string]interface{}) (c return confirmation, nil } -func NewAmqpProducer(uri string, queueName string) *AmqpProducer { +func NewAmqpProducer(uri string, config QueueConfig) *AmqpProducer { return &AmqpProducer{ client: &Client{ uri: uri, connectFunc: amqpConnect, }, - exchangeName: queueName, + config: config, } } diff --git a/internal/queue/producer_test.go b/internal/queue/producer_test.go index 55c22390..51a2ff08 100644 --- a/internal/queue/producer_test.go +++ b/internal/queue/producer_test.go @@ -14,13 +14,6 @@ import ( "github.com/stretchr/testify/assert" ) -const exchangeName = "test-exchange" -const exchangeWithDeclareError = "exchange-with-declare-error" - -const exchangeWithPublishNoAck = "exchange-with-publish-no-ack" -const exchangeWithPublishError = "exchange-with-publish-error" -const exchangeWithPublishHangs = "exchange-with-publish-hangs" - // TODO : Put this in a helper file because it is also used by consumer tests type MockAmqpChannel struct { msgs [][]byte @@ -31,11 +24,22 @@ type MockAmqpChannel struct { exchangeName string queueDurable bool confirmModeError bool + // Publish error modes + publishError bool + publishNoAck bool + publishHangs bool + // Declaration error modes + exchangeDeclareError bool + queueDeclareError bool + // Binding tracking + boundQueue string + boundExchange string + boundRoutingKey string + bindError bool } -func (ch *MockAmqpChannel) PublishWithDeferredConfirm(_ string, key string, _, _ bool, msg amqp.Publishing) (confirmation, error) { - - if key == exchangeWithPublishError { +func (ch *MockAmqpChannel) PublishWithDeferredConfirm(exchange string, key string, _, _ bool, msg amqp.Publishing) (confirmation, error) { + if ch.publishError { return nil, errors.New("publish error") } ch.msgs = append(ch.msgs, msg.Body) @@ -43,11 +47,11 @@ func (ch *MockAmqpChannel) PublishWithDeferredConfirm(_ string, key string, _, _ done_ch := make(chan struct{}, 1) - if key != exchangeWithPublishHangs { + if !ch.publishHangs { done_ch <- struct{}{} } - ack := key != exchangeWithPublishNoAck + ack := !ch.publishNoAck confirmation := &MockConfirmation{ done: done_ch, ack: ack, @@ -68,7 +72,7 @@ func (ch *MockAmqpChannel) Confirm(_ bool) error { } func (ch *MockAmqpChannel) ExchangeDeclare(name string, _ string, _, _, _, _ bool, _ amqp.Table) error { - if name == exchangeWithDeclareError { + if ch.exchangeDeclareError { return errors.New("exchange declare error") } ch.exchangeName = name @@ -76,7 +80,7 @@ func (ch *MockAmqpChannel) ExchangeDeclare(name string, _ string, _, _, _, _ boo } func (ch *MockAmqpChannel) QueueDeclare(name string, durable, _, _, _ bool, _ amqp.Table) (amqp.Queue, error) { - if name == queueWithDeclareError { + if ch.queueDeclareError { return amqp.Queue{}, errors.New("queue declare error") } ch.queueName = name @@ -88,6 +92,16 @@ func (ch *MockAmqpChannel) QueueDeclarePassive(name string, durable, autoDelete, return amqp.Queue{}, nil } +func (ch *MockAmqpChannel) QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error { + if ch.bindError { + return errors.New("bind error") + } + ch.boundQueue = name + ch.boundRoutingKey = key + ch.boundExchange = exchange + return nil +} + func (ch *MockAmqpChannel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) { return nil, nil } @@ -115,11 +129,12 @@ func (c *MockConfirmation) Acked() bool { } type MockAmqpConnection struct { - channelCalls int - amqpChannel *MockAmqpChannel - isclosed bool - errMode bool - confirmModeError bool + channelCalls int + amqpChannel *MockAmqpChannel + isclosed bool + errMode bool + confirmModeError bool + exchangeDeclareError bool } func (m *MockAmqpConnection) Channel() (amqpChannel, error) { @@ -128,7 +143,8 @@ func (m *MockAmqpConnection) Channel() (amqpChannel, error) { } m.channelCalls++ m.amqpChannel = &MockAmqpChannel{ - confirmModeError: m.confirmModeError, + confirmModeError: m.confirmModeError, + exchangeDeclareError: m.exchangeDeclareError, } return m.amqpChannel, nil } @@ -144,7 +160,7 @@ func (m *MockAmqpConnection) Close() error { func TestPush(t *testing.T) { /* - arrange: create a queue with a fake amqp connection + arrange: create a producer with a fake amqp connection act: push a message to the queue assert: message was published to the amqp channel */ @@ -156,7 +172,7 @@ func TestPush(t *testing.T) { amqpChannel: mockAmqpChannel, }, }, - exchangeName: exchangeName, + config: DefaultQueueConfig(), } headers := map[string]interface{}{"header1": "value1"} @@ -168,21 +184,21 @@ func TestPush(t *testing.T) { func TestPushFailure(t *testing.T) { /* - arrange: create a queue with a fake confirm handler that always fails + arrange: create a producer with a fake confirm handler that always fails act: push a message to the queue that will fail to publish assert: the push return an error */ tests := []struct { - name string - queueName string - context context.Context - errMsg string + name string + mockChannel *MockAmqpChannel + context context.Context + errMsg string }{ { - name: "message is not acked", - queueName: exchangeWithPublishNoAck, - errMsg: "confirmation not acknowledged", - context: context.Background(), + name: "message is not acked", + mockChannel: &MockAmqpChannel{publishNoAck: true}, + errMsg: "confirmation not acknowledged", + context: context.Background(), }, { name: "context is done", @@ -191,28 +207,27 @@ func TestPushFailure(t *testing.T) { cancel() return ctx }(), - queueName: exchangeWithPublishHangs, - errMsg: "context canceled", + mockChannel: &MockAmqpChannel{publishHangs: true}, + errMsg: "context canceled", }, { - name: "publish returns error", - queueName: exchangeWithPublishError, - errMsg: "publish error", - context: context.Background(), + name: "publish returns error", + mockChannel: &MockAmqpChannel{publishError: true}, + errMsg: "publish error", + context: context.Background(), }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - mockAmqpChannel := &MockAmqpChannel{} amqpProducer := &AmqpProducer{ client: &Client{ - amqpChannel: mockAmqpChannel, + amqpChannel: tt.mockChannel, amqpConnection: &MockAmqpConnection{ - amqpChannel: mockAmqpChannel, + amqpChannel: tt.mockChannel, }, }, - exchangeName: tt.queueName, + config: DefaultQueueConfig(), } err := amqpProducer.Push(tt.context, nil, []byte("TestMessage")) @@ -252,6 +267,7 @@ func TestPushNoChannel(t *testing.T) { amqpChannel: tt.channel, amqpConnection: mockAmqpConnection, }, + config: DefaultQueueConfig(), } amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) @@ -276,6 +292,7 @@ func TestPushNoChannelFailure(t *testing.T) { amqpChannel: nil, amqpConnection: mockAmqpConnection, }, + config: DefaultQueueConfig(), } err := amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) assert.Error(t, err, "expected error when channel fails to open") @@ -316,6 +333,7 @@ func TestPushNoConnection(t *testing.T) { return mockAmqpConnection, nil }, }, + config: DefaultQueueConfig(), } amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) @@ -339,6 +357,7 @@ func TestPushNoConnectionFailure(t *testing.T) { return nil, errors.New("connection error") }, }, + config: DefaultQueueConfig(), } err := amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) @@ -350,44 +369,42 @@ func TestPushExchangeDeclare(t *testing.T) { /* arrange: create a producer with no amqp channel act: push a message to the queue - assert: channel with confirm mode is established and exchange is declared + assert: channel with confirm mode is established and exchange is declared with config name */ mockAmqpConnection := &MockAmqpConnection{} + config := DefaultQueueConfig() amqpProducer := &AmqpProducer{ client: &Client{ amqpChannel: nil, amqpConnection: mockAmqpConnection, }, - exchangeName: exchangeName, + config: config, } amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) - assert.Equal(t, mockAmqpConnection.amqpChannel.confirmMode, true, "expected channel to be in confirm mode") - assert.Equal(t, mockAmqpConnection.amqpChannel.exchangeName, exchangeName, "expected exchange name to be "+exchangeName) + assert.True(t, mockAmqpConnection.amqpChannel.confirmMode, "expected channel to be in confirm mode") + assert.Equal(t, config.ExchangeName, mockAmqpConnection.amqpChannel.exchangeName, "expected exchange name to be "+config.ExchangeName) } func TestPushExchangeDeclareFailure(t *testing.T) { /* - arrange: create a queue with no amqp channel where the exchange declare fails + arrange: create a producer with no amqp channel where the exchange declare fails act: push a message to the queue assert: push returns an error */ tests := []struct { name string - exchangeName string mockAmqpConnection *MockAmqpConnection errMsg string }{ { name: "exchange declare error", - exchangeName: exchangeWithDeclareError, - mockAmqpConnection: &MockAmqpConnection{}, + mockAmqpConnection: &MockAmqpConnection{exchangeDeclareError: true}, errMsg: "exchange declare error", }, { - name: "confirm error", - exchangeName: exchangeName, + name: "confirm error", mockAmqpConnection: &MockAmqpConnection{ confirmModeError: true, }, @@ -402,12 +419,12 @@ func TestPushExchangeDeclareFailure(t *testing.T) { amqpChannel: nil, amqpConnection: tt.mockAmqpConnection, }, - exchangeName: tt.exchangeName, + config: DefaultQueueConfig(), } err := amqpProducer.Push(context.Background(), nil, []byte("TestMessage")) - assert.Error(t, err, "expected error when queue declare fails") + assert.Error(t, err, "expected error when exchange declare fails") assert.ErrorContains(t, err, tt.errMsg) }) } diff --git a/internal/queue/types.go b/internal/queue/types.go index a27643b7..6b52d0d1 100644 --- a/internal/queue/types.go +++ b/internal/queue/types.go @@ -85,6 +85,20 @@ func (c *Client) declareQueue(queueName string) error { return nil } +func (c *Client) bindQueue(queueName, routingKey, exchangeName string) error { + err := c.amqpChannel.QueueBind( + queueName, + routingKey, + exchangeName, + false, // no-wait + nil, // arguments + ) + if err != nil { + return fmt.Errorf("failed to bind AMQP queue: %w", err) + } + return nil +} + // Close gracefully closes the AMQP channel and connection. func (c *Client) Close() error { c.mu.Lock() @@ -108,8 +122,8 @@ func (c *Client) Close() error { } type AmqpProducer struct { - client *Client - exchangeName string + client *Client + config QueueConfig } type Producer interface { diff --git a/internal/queue/types_test.go b/internal/queue/types_test.go index 73560c0f..634e08a6 100644 --- a/internal/queue/types_test.go +++ b/internal/queue/types_test.go @@ -195,7 +195,6 @@ func TestAmqpConsumer_Table(t *testing.T) { conn := &fakeConnection{isClosedRet: false} cons := &AmqpConsumer{ - queueName: "q", client: &Client{ amqpChannel: ch, amqpConnection: conn, From ce15ff785c13f5a4b26bb457009b9f64cdbc4591 Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Tue, 6 Jan 2026 11:11:07 +0100 Subject: [PATCH 05/14] move MockAmqpChannel to mock_test.go --- internal/queue/mock_test.go | 172 ++++++++++++++++++++++++++++++++ internal/queue/producer_test.go | 145 --------------------------- 2 files changed, 172 insertions(+), 145 deletions(-) create mode 100644 internal/queue/mock_test.go diff --git a/internal/queue/mock_test.go b/internal/queue/mock_test.go new file mode 100644 index 00000000..615bef55 --- /dev/null +++ b/internal/queue/mock_test.go @@ -0,0 +1,172 @@ +/* + * Copyright 2026 Canonical Ltd. + * See LICENSE file for licensing details. + */ + +package queue + +import ( + "errors" + + amqp "github.com/rabbitmq/amqp091-go" +) + +// MockAmqpChannel implements amqpChannel for unit testing. +type MockAmqpChannel struct { + // Message tracking + msgs [][]byte + headers []map[string]interface{} + + // State + isclosed bool + closeCalls int + closeErr error + confirmMode bool + + // Declaration tracking + queueName string + exchangeName string + queueDurable bool + + // Binding tracking + boundQueue string + boundExchange string + boundRoutingKey string + + // Publish behavior + publishError bool + publishNoAck bool + publishHangs bool + + // Error modes + confirmModeError bool + exchangeDeclareError bool + queueDeclareError bool + bindError bool +} + +func (ch *MockAmqpChannel) PublishWithDeferredConfirm(exchange string, key string, _, _ bool, msg amqp.Publishing) (confirmation, error) { + if ch.publishError { + return nil, errors.New("publish error") + } + ch.msgs = append(ch.msgs, msg.Body) + ch.headers = append(ch.headers, msg.Headers) + + doneCh := make(chan struct{}, 1) + + if !ch.publishHangs { + doneCh <- struct{}{} + } + + ack := !ch.publishNoAck + conf := &MockConfirmation{ + done: doneCh, + ack: ack, + } + return conf, nil +} + +func (ch *MockAmqpChannel) IsClosed() bool { + return ch.isclosed +} + +func (ch *MockAmqpChannel) Confirm(_ bool) error { + if ch.confirmModeError { + return errors.New("confirm error") + } + ch.confirmMode = true + return nil +} + +func (ch *MockAmqpChannel) ExchangeDeclare(name string, _ string, _, _, _, _ bool, _ amqp.Table) error { + if ch.exchangeDeclareError { + return errors.New("exchange declare error") + } + ch.exchangeName = name + return nil +} + +func (ch *MockAmqpChannel) QueueDeclare(name string, durable, _, _, _ bool, _ amqp.Table) (amqp.Queue, error) { + if ch.queueDeclareError { + return amqp.Queue{}, errors.New("queue declare error") + } + ch.queueName = name + ch.queueDurable = durable + return amqp.Queue{Name: name}, nil +} + +func (ch *MockAmqpChannel) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) { + return amqp.Queue{Name: name}, nil +} + +func (ch *MockAmqpChannel) QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error { + if ch.bindError { + return errors.New("bind error") + } + ch.boundQueue = name + ch.boundRoutingKey = key + ch.boundExchange = exchange + return nil +} + +func (ch *MockAmqpChannel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) { + return nil, nil +} + +func (ch *MockAmqpChannel) Qos(prefetchCount, prefetchSize int, global bool) error { + return nil +} + +func (ch *MockAmqpChannel) Close() error { + ch.closeCalls++ + ch.isclosed = true + return ch.closeErr +} + +// MockConfirmation implements confirmation for unit testing. +type MockConfirmation struct { + done <-chan struct{} + ack bool +} + +func (c *MockConfirmation) Done() <-chan struct{} { + return c.done +} + +func (c *MockConfirmation) Acked() bool { + return c.ack +} + +// MockAmqpConnection implements amqpConnection for unit testing. +type MockAmqpConnection struct { + channelCalls int + amqpChannel *MockAmqpChannel + isclosed bool + closeCalls int + closeErr error + errMode bool + confirmModeError bool + exchangeDeclareError bool +} + +func (m *MockAmqpConnection) Channel() (amqpChannel, error) { + if m.errMode { + return nil, errors.New("failed to open channel") + } + m.channelCalls++ + m.amqpChannel = &MockAmqpChannel{ + confirmModeError: m.confirmModeError, + exchangeDeclareError: m.exchangeDeclareError, + } + return m.amqpChannel, nil +} + +func (m *MockAmqpConnection) IsClosed() bool { + return m.isclosed +} + +func (m *MockAmqpConnection) Close() error { + m.closeCalls++ + m.isclosed = true + return m.closeErr +} diff --git a/internal/queue/producer_test.go b/internal/queue/producer_test.go index 51a2ff08..0f036c77 100644 --- a/internal/queue/producer_test.go +++ b/internal/queue/producer_test.go @@ -10,154 +10,9 @@ import ( "errors" "testing" - amqp "github.com/rabbitmq/amqp091-go" "github.com/stretchr/testify/assert" ) -// TODO : Put this in a helper file because it is also used by consumer tests -type MockAmqpChannel struct { - msgs [][]byte - headers []map[string]interface{} - isclosed bool - confirmMode bool - queueName string - exchangeName string - queueDurable bool - confirmModeError bool - // Publish error modes - publishError bool - publishNoAck bool - publishHangs bool - // Declaration error modes - exchangeDeclareError bool - queueDeclareError bool - // Binding tracking - boundQueue string - boundExchange string - boundRoutingKey string - bindError bool -} - -func (ch *MockAmqpChannel) PublishWithDeferredConfirm(exchange string, key string, _, _ bool, msg amqp.Publishing) (confirmation, error) { - if ch.publishError { - return nil, errors.New("publish error") - } - ch.msgs = append(ch.msgs, msg.Body) - ch.headers = append(ch.headers, msg.Headers) - - done_ch := make(chan struct{}, 1) - - if !ch.publishHangs { - done_ch <- struct{}{} - } - - ack := !ch.publishNoAck - confirmation := &MockConfirmation{ - done: done_ch, - ack: ack, - } - return confirmation, nil -} - -func (ch *MockAmqpChannel) IsClosed() bool { - return ch.isclosed -} - -func (ch *MockAmqpChannel) Confirm(_ bool) error { - if ch.confirmModeError { - return errors.New("confirm error") - } - ch.confirmMode = true - return nil -} - -func (ch *MockAmqpChannel) ExchangeDeclare(name string, _ string, _, _, _, _ bool, _ amqp.Table) error { - if ch.exchangeDeclareError { - return errors.New("exchange declare error") - } - ch.exchangeName = name - return nil -} - -func (ch *MockAmqpChannel) QueueDeclare(name string, durable, _, _, _ bool, _ amqp.Table) (amqp.Queue, error) { - if ch.queueDeclareError { - return amqp.Queue{}, errors.New("queue declare error") - } - ch.queueName = name - ch.queueDurable = durable - return amqp.Queue{}, nil -} - -func (ch *MockAmqpChannel) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) { - return amqp.Queue{}, nil -} - -func (ch *MockAmqpChannel) QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error { - if ch.bindError { - return errors.New("bind error") - } - ch.boundQueue = name - ch.boundRoutingKey = key - ch.boundExchange = exchange - return nil -} - -func (ch *MockAmqpChannel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) { - return nil, nil -} - -func (ch *MockAmqpChannel) Qos(prefetchCount, prefetchSize int, global bool) error { - return nil -} - -func (ch *MockAmqpChannel) Close() error { - ch.isclosed = true - return nil -} - -type MockConfirmation struct { - done <-chan struct{} - ack bool -} - -func (c *MockConfirmation) Done() <-chan struct{} { - return c.done -} - -func (c *MockConfirmation) Acked() bool { - return c.ack -} - -type MockAmqpConnection struct { - channelCalls int - amqpChannel *MockAmqpChannel - isclosed bool - errMode bool - confirmModeError bool - exchangeDeclareError bool -} - -func (m *MockAmqpConnection) Channel() (amqpChannel, error) { - if m.errMode { - return nil, errors.New("failed to open channel") - } - m.channelCalls++ - m.amqpChannel = &MockAmqpChannel{ - confirmModeError: m.confirmModeError, - exchangeDeclareError: m.exchangeDeclareError, - } - return m.amqpChannel, nil -} - -func (m *MockAmqpConnection) IsClosed() bool { - return m.isclosed -} - -func (m *MockAmqpConnection) Close() error { - m.isclosed = true - return nil -} - func TestPush(t *testing.T) { /* arrange: create a producer with a fake amqp connection From 51540a39a7abd7e93d6e55eb722ddf3039a9bde1 Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Tue, 6 Jan 2026 11:15:43 +0100 Subject: [PATCH 06/14] remove fakeChannel implementation --- internal/queue/mock_test.go | 28 +++++++++-- internal/queue/types_test.go | 96 ++++++------------------------------ 2 files changed, 39 insertions(+), 85 deletions(-) diff --git a/internal/queue/mock_test.go b/internal/queue/mock_test.go index 615bef55..6b24650e 100644 --- a/internal/queue/mock_test.go +++ b/internal/queue/mock_test.go @@ -19,6 +19,7 @@ type MockAmqpChannel struct { // State isclosed bool + isClosedRet bool closeCalls int closeErr error confirmMode bool @@ -43,6 +44,14 @@ type MockAmqpChannel struct { exchangeDeclareError bool queueDeclareError bool bindError bool + + // Consumer support + consumeCh <-chan amqp.Delivery + consumeErr error + qosErr error + + // Additional error modes for types_test + queueDeclarePassiveErr error } func (ch *MockAmqpChannel) PublishWithDeferredConfirm(exchange string, key string, _, _ bool, msg amqp.Publishing) (confirmation, error) { @@ -67,7 +76,7 @@ func (ch *MockAmqpChannel) PublishWithDeferredConfirm(exchange string, key strin } func (ch *MockAmqpChannel) IsClosed() bool { - return ch.isclosed + return ch.isclosed || ch.isClosedRet } func (ch *MockAmqpChannel) Confirm(_ bool) error { @@ -96,7 +105,7 @@ func (ch *MockAmqpChannel) QueueDeclare(name string, durable, _, _, _ bool, _ am } func (ch *MockAmqpChannel) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) { - return amqp.Queue{Name: name}, nil + return amqp.Queue{Name: name}, ch.queueDeclarePassiveErr } func (ch *MockAmqpChannel) QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error { @@ -110,11 +119,11 @@ func (ch *MockAmqpChannel) QueueBind(name, key, exchange string, noWait bool, ar } func (ch *MockAmqpChannel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) { - return nil, nil + return ch.consumeCh, ch.consumeErr } func (ch *MockAmqpChannel) Qos(prefetchCount, prefetchSize int, global bool) error { - return nil + return ch.qosErr } func (ch *MockAmqpChannel) Close() error { @@ -141,7 +150,10 @@ func (c *MockConfirmation) Acked() bool { type MockAmqpConnection struct { channelCalls int amqpChannel *MockAmqpChannel + channel amqpChannel + channelErr error isclosed bool + isClosedRet bool closeCalls int closeErr error errMode bool @@ -153,6 +165,12 @@ func (m *MockAmqpConnection) Channel() (amqpChannel, error) { if m.errMode { return nil, errors.New("failed to open channel") } + if m.channelErr != nil { + return nil, m.channelErr + } + if m.channel != nil { + return m.channel, nil + } m.channelCalls++ m.amqpChannel = &MockAmqpChannel{ confirmModeError: m.confirmModeError, @@ -162,7 +180,7 @@ func (m *MockAmqpConnection) Channel() (amqpChannel, error) { } func (m *MockAmqpConnection) IsClosed() bool { - return m.isclosed + return m.isclosed || m.isClosedRet } func (m *MockAmqpConnection) Close() error { diff --git a/internal/queue/types_test.go b/internal/queue/types_test.go index 634e08a6..0cc8abfb 100644 --- a/internal/queue/types_test.go +++ b/internal/queue/types_test.go @@ -11,70 +11,6 @@ import ( "github.com/stretchr/testify/assert" ) -// fakeChannel implements amqpChannel for unit testing without external deps. -type fakeChannel struct { - closed bool - closeCalls int - closeErr error - isClosedRet bool - - publishErr error - confirmErr error - qosErr error - consumeErr error - declareErr error - declarePassErr error - - consumeCh <-chan amqp.Delivery -} - -func (f *fakeChannel) PublishWithDeferredConfirm(exchange string, key string, mandatory, immediate bool, msg amqp.Publishing) (confirmation, error) { - return nil, f.publishErr -} -func (f *fakeChannel) IsClosed() bool { return f.closed || f.isClosedRet } -func (f *fakeChannel) Close() error { - f.closeCalls++ - f.closed = true - return f.closeErr -} -func (f *fakeChannel) Confirm(noWait bool) error { return f.confirmErr } -func (f *fakeChannel) ExchangeDeclare(name string, kind string, durable, autoDelete, internal, noWait bool, args amqp.Table) error { - return nil -} - -func (f *fakeChannel) QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) { - return amqp.Queue{Name: name}, f.declareErr -} -func (f *fakeChannel) QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error { - return nil -} -func (f *fakeChannel) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error) { - return amqp.Queue{Name: name}, f.declarePassErr -} -func (f *fakeChannel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) { - return f.consumeCh, f.consumeErr -} -func (f *fakeChannel) Qos(prefetchCount, prefetchSize int, global bool) error { return f.qosErr } - -// fakeConnection implements amqpConnection for unit testing. -type fakeConnection struct { - closed bool - closeCalls int - closeErr error - isClosedRet bool - - channel amqpChannel - chErr error -} - -func (f *fakeConnection) Channel() (amqpChannel, error) { return f.channel, f.chErr } -func (f *fakeConnection) IsClosed() bool { return f.closed || f.isClosedRet } -func (f *fakeConnection) Close() error { - f.closeCalls++ - f.closed = true - return f.closeErr -} - // fakeJobDB implements JobDatabase with no-op methods for consumer tests. type fakeJobDB struct{} @@ -107,8 +43,8 @@ func TestClient_Close_Table(t *testing.T) { { name: "both open ok", client: &Client{ - amqpChannel: &fakeChannel{isClosedRet: false}, - amqpConnection: &fakeConnection{isClosedRet: false}, + amqpChannel: &MockAmqpChannel{isClosedRet: false}, + amqpConnection: &MockAmqpConnection{isClosedRet: false}, }, expectErr: nil, expectChCalls: 1, @@ -119,8 +55,8 @@ func TestClient_Close_Table(t *testing.T) { { name: "channel error first", client: &Client{ - amqpChannel: &fakeChannel{isClosedRet: false, closeErr: chErr}, - amqpConnection: &fakeConnection{isClosedRet: false, closeErr: connErr}, + amqpChannel: &MockAmqpChannel{isClosedRet: false, closeErr: chErr}, + amqpConnection: &MockAmqpConnection{isClosedRet: false, closeErr: connErr}, }, expectErr: chErr, expectChCalls: 1, @@ -129,8 +65,8 @@ func TestClient_Close_Table(t *testing.T) { { name: "connection error only", client: &Client{ - amqpChannel: &fakeChannel{isClosedRet: false}, - amqpConnection: &fakeConnection{isClosedRet: false, closeErr: connErr}, + amqpChannel: &MockAmqpChannel{isClosedRet: false}, + amqpConnection: &MockAmqpConnection{isClosedRet: false, closeErr: connErr}, }, expectErr: connErr, expectChCalls: 1, @@ -148,8 +84,8 @@ func TestClient_Close_Table(t *testing.T) { { name: "already closed resources", client: &Client{ - amqpChannel: &fakeChannel{isClosedRet: true}, - amqpConnection: &fakeConnection{isClosedRet: true}, + amqpChannel: &MockAmqpChannel{isClosedRet: true}, + amqpConnection: &MockAmqpConnection{isClosedRet: true}, }, expectErr: nil, expectChCalls: 0, @@ -167,16 +103,16 @@ func TestClient_Close_Table(t *testing.T) { assert.NoError(t, err) } - if fc, ok := tt.client.amqpChannel.(*fakeChannel); ok { + if fc, ok := tt.client.amqpChannel.(*MockAmqpChannel); ok { assert.Equal(t, tt.expectChCalls, fc.closeCalls) if tt.expectChClosed { - assert.True(t, fc.closed) + assert.True(t, fc.isclosed) } } - if fconn, ok := tt.client.amqpConnection.(*fakeConnection); ok { + if fconn, ok := tt.client.amqpConnection.(*MockAmqpConnection); ok { assert.Equal(t, tt.expectConnCalls, fconn.closeCalls) if tt.expectConnClose { - assert.True(t, fconn.closed) + assert.True(t, fconn.isclosed) } } }) @@ -191,8 +127,8 @@ func TestAmqpConsumer_Table(t *testing.T) { act: Close the consumer. assert: Check no error and verify close calls. */ - ch := &fakeChannel{isClosedRet: false} - conn := &fakeConnection{isClosedRet: false} + ch := &MockAmqpChannel{isClosedRet: false} + conn := &MockAmqpConnection{isClosedRet: false} cons := &AmqpConsumer{ client: &Client{ @@ -215,8 +151,8 @@ func TestClient_Close_Idempotent(t *testing.T) { act: Close the client twice. assert: First close should close both, second close should be a no-op. */ - ch := &fakeChannel{isClosedRet: false} - conn := &fakeConnection{isClosedRet: false} + ch := &MockAmqpChannel{isClosedRet: false} + conn := &MockAmqpConnection{isClosedRet: false} c := &Client{ amqpChannel: ch, From 738637379ca51fc63693b937ab57311fd5c15371 Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Tue, 6 Jan 2026 12:59:08 +0100 Subject: [PATCH 07/14] fix planner main_test --- cmd/planner/main_test.go | 37 +++++++++++++++++++++++++++++++++---- 1 file changed, 33 insertions(+), 4 deletions(-) diff --git a/cmd/planner/main_test.go b/cmd/planner/main_test.go index 6e645cc5..f4d990e0 100644 --- a/cmd/planner/main_test.go +++ b/cmd/planner/main_test.go @@ -19,6 +19,7 @@ import ( "testing" "time" + "github.com/canonical/github-runner-operators/internal/queue" amqp "github.com/rabbitmq/amqp091-go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -74,7 +75,7 @@ func TestMain_FlavorPressure(t *testing.T) { assert.Eventually(t, func() bool { press := ctx.getFlavorPressure(flavor) return press[flavor] > pressure - }, 20*time.Minute, 500*time.Millisecond) + }, 2*time.Minute, 100*time.Millisecond, "expected flavor pressure to increase after webhook processing") // Trigger graceful shutdown and verify services stop cleanly ctx.shutdownMain() @@ -112,7 +113,7 @@ func (ctx *testContext) constructWebhookPayload(labels []string) []byte { return body } -// declareQueue declares a RabbitMQ queue for testing purposes. +// declareQueue declares the exchange, queue, and binding for testing. func (ctx *testContext) declareQueue() { ctx.t.Helper() @@ -124,6 +125,21 @@ func (ctx *testContext) declareQueue() { require.NoError(ctx.t, err, "open channel") defer ch.Close() + config := queue.DefaultQueueConfig() + + // Declare exchange + err = ch.ExchangeDeclare( + config.ExchangeName, + "direct", // type + true, // durable + false, // auto-delete + false, // internal + false, // no-wait + nil, + ) + require.NoError(ctx.t, err, "declare exchange") + + // Declare queue _, err = ch.QueueDeclare( ctx.queueName, true, // durable @@ -133,9 +149,19 @@ func (ctx *testContext) declareQueue() { nil, ) require.NoError(ctx.t, err, "declare queue") + + // Bind queue to exchange + err = ch.QueueBind( + ctx.queueName, + config.RoutingKey, + config.ExchangeName, + false, // no-wait + nil, + ) + require.NoError(ctx.t, err, "bind queue") } -// publishAndWaitAck publishes a message to the given RabbitMQ queue and waits for an ack. +// publishAndWaitAck publishes a message to the exchange with routing key and waits for an ack. func (ctx *testContext) publishAndWaitAck(body []byte) { ctx.t.Helper() @@ -152,8 +178,11 @@ func (ctx *testContext) publishAndWaitAck(body []byte) { confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1)) + config := queue.DefaultQueueConfig() err = ch.Publish( - "", ctx.queueName, false, false, + config.ExchangeName, // publish to exchange instead of directly to queue + config.RoutingKey, // use routing key + false, false, amqp.Publishing{ ContentType: "application/json", Body: body, From 0037c88b96415873b29dc06883be2ac01aec86e1 Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Tue, 6 Jan 2026 13:06:09 +0100 Subject: [PATCH 08/14] fix webhook_test.go --- cmd/webhook-gateway/webhook_test.go | 64 +++++++++++++++++++++++++---- 1 file changed, 55 insertions(+), 9 deletions(-) diff --git a/cmd/webhook-gateway/webhook_test.go b/cmd/webhook-gateway/webhook_test.go index f3dbb9dd..ce201ed1 100644 --- a/cmd/webhook-gateway/webhook_test.go +++ b/cmd/webhook-gateway/webhook_test.go @@ -19,19 +19,23 @@ import ( "testing" "time" + "github.com/canonical/github-runner-operators/internal/queue" + amqp "github.com/rabbitmq/amqp091-go" "github.com/stretchr/testify/assert" ) func TestHTTPRequestIsForwarded(t *testing.T) { const payload = `{"message":"Hello, Bob!"}` + amqpUri := getAmqpUriFromEnv(t) + + setupQueue(t, amqpUri) go main() secret := getSecretFromEnv(t) sendPayloadToHTTPServer(t, payload, secret) - amqpUri := getAmqpUriFromEnv(t) msg := consumeMessage(t, amqpUri) assert.Equal(t, payload, msg, "expected message body to match") @@ -98,7 +102,7 @@ func createSignature(message string, secret string) string { return "sha256=" + hex.EncodeToString(h.Sum(nil)) } -func consumeMessage(t *testing.T, amqpUri string) string { +func setupQueue(t *testing.T, amqpUri string) { conn, err := amqp.Dial(amqpUri) if err != nil { t.Fatalf("Failed to connect to RabbitMQ: %v", err) @@ -110,20 +114,62 @@ func consumeMessage(t *testing.T, amqpUri string) string { } defer ch.Close() + config := queue.DefaultQueueConfig() + + err = ch.ExchangeDeclare( + config.ExchangeName, // name) + "direct", // type + true, // durable + false, // auto-deleted + false, // internal + false, // no-wait + nil, // arguments + ) + + if err != nil { + t.Fatalf("Failed to declare an exchange: %v", err) + } _, err = ch.QueueDeclare( - "webhook-queue", // name - true, // durable - false, // delete when unused - false, // exclusive - false, // no-wait - nil, // arguments + config.QueueName, // name + true, // durable + false, // delete when unused + false, // exclusive + false, // no-wait + nil, // arguments ) if err != nil { t.Fatalf("Failed to declare a queue: %v", err) } - deliveryChan, err := ch.Consume("webhook-queue", "", true, false, false, false, nil) + err = ch.QueueBind( + config.QueueName, // queue name + config.RoutingKey, // routing key + config.ExchangeName, // exchange + false, + nil, + ) + + if err != nil { + t.Fatalf("Failed to bind a queue: %v", err) + } +} + +func consumeMessage(t *testing.T, amqpUri string) string { + conn, err := amqp.Dial(amqpUri) + if err != nil { + t.Fatalf("Failed to connect to RabbitMQ: %v", err) + } + defer conn.Close() + ch, err := conn.Channel() + if err != nil { + t.Fatalf("Failed to open a channel: %v", err) + } + defer ch.Close() + + config := queue.DefaultQueueConfig() + + deliveryChan, err := ch.Consume(config.QueueName, "", true, false, false, false, nil) if err != nil { t.Fatalf("Failed to register a consumer: %v", err) From 435b180cd831477a2c28f94b0a1c491486be6034 Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Tue, 6 Jan 2026 13:10:49 +0100 Subject: [PATCH 09/14] refactor webhook_test.go --- cmd/webhook-gateway/webhook_test.go | 36 ++++++++++++++++++----------- 1 file changed, 22 insertions(+), 14 deletions(-) diff --git a/cmd/webhook-gateway/webhook_test.go b/cmd/webhook-gateway/webhook_test.go index ce201ed1..18d462cd 100644 --- a/cmd/webhook-gateway/webhook_test.go +++ b/cmd/webhook-gateway/webhook_test.go @@ -102,21 +102,37 @@ func createSignature(message string, secret string) string { return "sha256=" + hex.EncodeToString(h.Sum(nil)) } -func setupQueue(t *testing.T, amqpUri string) { +// returns an AMQP channel and a cleanup function to close the connection and channel +func setupAMQPChannel(t *testing.T, amqpUri string) (*amqp.Channel, func()) { conn, err := amqp.Dial(amqpUri) if err != nil { t.Fatalf("Failed to connect to RabbitMQ: %v", err) } - defer conn.Close() + ch, err := conn.Channel() if err != nil { + conn.Close() t.Fatalf("Failed to open a channel: %v", err) } - defer ch.Close() + + cleanup := func() { + ch.Close() + conn.Close() + } + + return ch, cleanup +} + +// setupQueue declares the exchange, queue, and binding for the test +// this is required because the webhook-gateway does not create the queue and +// published messages could be discarded if the queue does not exist +func setupQueue(t *testing.T, amqpUri string) { + ch, cleanup := setupAMQPChannel(t, amqpUri) + defer cleanup() config := queue.DefaultQueueConfig() - err = ch.ExchangeDeclare( + err := ch.ExchangeDeclare( config.ExchangeName, // name) "direct", // type true, // durable @@ -156,16 +172,8 @@ func setupQueue(t *testing.T, amqpUri string) { } func consumeMessage(t *testing.T, amqpUri string) string { - conn, err := amqp.Dial(amqpUri) - if err != nil { - t.Fatalf("Failed to connect to RabbitMQ: %v", err) - } - defer conn.Close() - ch, err := conn.Channel() - if err != nil { - t.Fatalf("Failed to open a channel: %v", err) - } - defer ch.Close() + ch, cleanup := setupAMQPChannel(t, amqpUri) + defer cleanup() config := queue.DefaultQueueConfig() From e1de3b1bb26c0ce8677dec663466469bb9493b07 Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Tue, 6 Jan 2026 13:25:54 +0100 Subject: [PATCH 10/14] remove redundant comments --- cmd/planner/main_test.go | 3 --- 1 file changed, 3 deletions(-) diff --git a/cmd/planner/main_test.go b/cmd/planner/main_test.go index f4d990e0..57b3d89b 100644 --- a/cmd/planner/main_test.go +++ b/cmd/planner/main_test.go @@ -127,7 +127,6 @@ func (ctx *testContext) declareQueue() { config := queue.DefaultQueueConfig() - // Declare exchange err = ch.ExchangeDeclare( config.ExchangeName, "direct", // type @@ -139,7 +138,6 @@ func (ctx *testContext) declareQueue() { ) require.NoError(ctx.t, err, "declare exchange") - // Declare queue _, err = ch.QueueDeclare( ctx.queueName, true, // durable @@ -150,7 +148,6 @@ func (ctx *testContext) declareQueue() { ) require.NoError(ctx.t, err, "declare queue") - // Bind queue to exchange err = ch.QueueBind( ctx.queueName, config.RoutingKey, From 0f61ffdede2bda8be7b84a113eab737b49d859ec Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Tue, 6 Jan 2026 13:30:49 +0100 Subject: [PATCH 11/14] update CONTRIBUTING.md --- CONTRIBUTING.md | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 5f6188b9..de9ea153 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -113,7 +113,7 @@ docker run -d --rm --name postgres -e POSTGRES_PASSWORD=password -p 5432:5432 po Run `webhook-gateway` integration tests using: ```shell -APP_PORT=8080 APP_WEBHOOK_SECRET_VALUE=fake RABBITMQ_CONNECT_STRING="amqp://guest:guest@localhost:5672/" go test -cover -v ./cmd/webhook-gateway -integration +APP_WEBHOOK_SECRET_VALUE=fake APP_PORT=8080 RABBITMQ_CONNECT_STRING="amqp://guest:guest@localhost:5672/" go test -v -cover -tags=integration -race ./cmd/webhook-gateway/... ``` It assumes you have access to a RabbitMQ server running reachable at $RABBITMQ_CONNECT_STRING. @@ -126,11 +126,10 @@ docker run -d --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4-manag Run `planner` integration tests using: ```shell -POSTGRESQL_DB_CONNECT_STRING="postgres://postgres:postgres@localhost:5432/gh_runner_operators?sslmode=disable" APP_PORT=8080 go test ./cmd/planner -v -count=1 +APP_PORT=8080 POSTGRESQL_DB_CONNECT_STRING="postgres://postgres:password@localhost:5432/postgres?sslmode=disable" RABBITMQ_CONNECT_STRING="amqp://guest:guest@localhost:5672/" go test -v -cover -tags=integration -race ./cmd/planner/.. ``` -It assumes you are connected to a local PostgreSQL database "gh_runner_operators" as user "postgres" on host "localhost" at port "5432". - +It assumes you can reach postgres and rabbitmq servers as described above. ### Test design From ecf7d6c9ebdd5ab3384cf4039e4b1a60f21af8b7 Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Tue, 6 Jan 2026 13:38:59 +0100 Subject: [PATCH 12/14] Update CONTRIBUTING.md Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- CONTRIBUTING.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index de9ea153..746972a3 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -126,7 +126,7 @@ docker run -d --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4-manag Run `planner` integration tests using: ```shell -APP_PORT=8080 POSTGRESQL_DB_CONNECT_STRING="postgres://postgres:password@localhost:5432/postgres?sslmode=disable" RABBITMQ_CONNECT_STRING="amqp://guest:guest@localhost:5672/" go test -v -cover -tags=integration -race ./cmd/planner/.. +APP_PORT=8080 POSTGRESQL_DB_CONNECT_STRING="postgres://postgres:password@localhost:5432/postgres?sslmode=disable" RABBITMQ_CONNECT_STRING="amqp://guest:guest@localhost:5672/" go test -v -cover -tags=integration -race ./cmd/planner/... ``` It assumes you can reach postgres and rabbitmq servers as described above. From c16ca5f09d5aa5bef1616e8d030d3e2af8c5c721 Mon Sep 17 00:00:00 2001 From: Christopher Bartz Date: Wed, 7 Jan 2026 14:33:40 +0100 Subject: [PATCH 13/14] simplify NewJobConsumer signature --- cmd/planner/main.go | 3 ++- internal/planner/consumer.go | 4 ++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/cmd/planner/main.go b/cmd/planner/main.go index ea8da28f..e2142002 100644 --- a/cmd/planner/main.go +++ b/cmd/planner/main.go @@ -71,7 +71,8 @@ func main() { metrics := planner.NewMetrics(db) - consumer := planner.NewJobConsumer(rabbitMQUri, queue.DefaultQueueConfig(), db, metrics) + amqpConsumer := queue.NewAmqpConsumer(rabbitMQUri, queue.DefaultQueueConfig()) + consumer := planner.NewJobConsumer(amqpConsumer, db, metrics) var consumerWg sync.WaitGroup consumerWg.Add(1) diff --git a/internal/planner/consumer.go b/internal/planner/consumer.go index 0cbd4df2..7bb30dc5 100644 --- a/internal/planner/consumer.go +++ b/internal/planner/consumer.go @@ -32,9 +32,9 @@ const ( maxBackoff = 5 * time.Minute ) -func NewJobConsumer(amqpUri string, queueConfig queue.QueueConfig, db JobDatabase, metrics *Metrics) *JobConsumer { +func NewJobConsumer(consumer queue.Consumer, db JobDatabase, metrics *Metrics) *JobConsumer { return &JobConsumer{ - consumer: queue.NewAmqpConsumer(amqpUri, queueConfig), + consumer: consumer, db: db, metrics: metrics, } From f59981e488f57fa0cd0226b0d481f891a5a18ca9 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 7 Jan 2026 14:14:12 +0000 Subject: [PATCH 14/14] Initial plan