25881f3ff8
- Pub/sub broker with fan-out to SSE clients - POST /webhook — receives Gitea push events, optional HMAC verification - GET /events — SSE stream of push events - GET /health — health check endpoint - Zero external dependencies (stdlib only) - Docker-based build via goctl wrapper (golang:1.26-alpine) - systemd unit, README with deployment docs - Tests: broker pub/sub, slow subscriber drop, ref parsing, signature verification - 6.0M static binary
141 lines
3.1 KiB
Go
141 lines
3.1 KiB
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")
|
|
}
|
|
}
|