converge/converge-relay/broker_test.go

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"net/http"
	"testing"
	"time"
)

func TestBrokerSubscribePublishUnsubscribe(t *testing.T) {
	broker := NewBroker()
	go broker.Run()
	defer broker.Stop()

	ch1 := broker.Subscribe()
	ch2 := broker.Subscribe()

	evt := PushEvent{FullName: "org/repo", Branch: "main", Ref: "refs/heads/main"}

	// Publish
	broker.Publish(evt)

	// Both subscribers should receive
	select {
	case got := <-ch1:
		if got != evt {
			t.Errorf("ch1: got %+v, want %+v", got, evt)
		}
	case <-time.After(time.Second):
		t.Fatal("ch1: timed out waiting for event")
	}

	select {
	case got := <-ch2:
		if got != evt {
			t.Errorf("ch2: got %+v, want %+v", got, evt)
		}
	case <-time.After(time.Second):
		t.Fatal("ch2: timed out waiting for event")
	}

	// Unsubscribe ch2, publish again
	broker.Unsubscribe(ch2)
	// Drain ch2 (Unsubscribe closes it)
	for range ch2 {
	}

	evt2 := PushEvent{FullName: "org/repo2", Branch: "dev", Ref: "refs/heads/dev"}
	broker.Publish(evt2)

	// ch1 gets it
	select {
	case got := <-ch1:
		if got != evt2 {
			t.Errorf("ch1 second: got %+v, want %+v", got, evt2)
		}
	case <-time.After(time.Second):
		t.Fatal("ch1: timed out waiting for second event")
	}

	// ch2 is closed, should not receive
	_, open := <-ch2
	if open {
		t.Error("ch2 should be closed after unsubscribe")
	}
}

func TestBrokerSlowSubscriberDropped(t *testing.T) {
	broker := NewBroker()
	go broker.Run()
	defer broker.Stop()

	// Channel with buffer 0 — publish will always fail to send
	ch := make(chan PushEvent)
	broker.subs[ch] = struct{}{} // bypass Subscribe for manual control

	evt := PushEvent{FullName: "org/repo", Branch: "main"}
	broker.Publish(evt)

	// Give it time to process
	time.Sleep(50 * time.Millisecond)

	// Should not have blocked the broker; Publish returned fine
	// and we didn't deadlock.
	select {
	case <-ch:
		t.Error("slow subscriber should have been dropped")
	default:
		// expected: no event on unbuffered channel
	}
	// ch is closed by broker.Stop() via defer
}

func TestRefToBranch(t *testing.T) {
	tests := []struct {
		ref  string
		want string
	}{
		{"refs/heads/main", "main"},
		{"refs/heads/feature/x", "feature/x"},
		{"refs/tags/v1.0", ""},
		{"refs/heads/", ""},
		{"main", ""},
	}
	for _, tt := range tests {
		got := refToBranch(tt.ref)
		if got != tt.want {
			t.Errorf("refToBranch(%q) = %q, want %q", tt.ref, got, tt.want)
		}
	}
}

func TestVerifySignature(t *testing.T) {
	secret := "test-secret"
	body := []byte(`{"ref":"refs/heads/main"}`)

	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write(body)
	validSig := "sha256=" + hex.EncodeToString(mac.Sum(nil))

	headers := http.Header{}
	headers.Set("X-Gitea-Signature", validSig)

	if !verifySignature(body, secret, headers) {
		t.Error("valid signature should pass")
	}

	headers.Set("X-Gitea-Signature", "sha256=bad")
	if verifySignature(body, secret, headers) {
		t.Error("bad signature should fail")
	}

	headers.Del("X-Gitea-Signature")
	headers.Set("X-Hub-Signature-256", validSig)
	if !verifySignature(body, secret, headers) {
		t.Error("X-Hub-Signature-256 with valid signature should pass")
	}
}