plumber/internal/events/bus_test.go

package events

import (
	"context"
	"sync"
	"testing"
	"time"
)

func TestPermalink(t *testing.T) {
	t.Parallel()

	tests := []struct {
		base, root, post, want string
	}{
		{"", "root-1", "post-2", "/questions/root-1#post-post-2"},
		{"https://www.askaplumberfirst.com/", "root-1", "post-2", "https://www.askaplumberfirst.com/questions/root-1#post-post-2"},
		{"https://www.askaplumberfirst.com", "a b", "c/d", "https://www.askaplumberfirst.com/questions/a%20b#post-c%2Fd"},
	}
	for _, tc := range tests {
		if got := Permalink(tc.base, tc.root, tc.post); got != tc.want {
			t.Fatalf("Permalink(%q, %q, %q) = %q, want %q", tc.base, tc.root, tc.post, got, tc.want)
		}
	}
}

func TestNopAndRecording(t *testing.T) {
	t.Parallel()

	Nop{}.Publish(context.Background(), PostCreated{})

	rec := &Recording{}
	rec.Publish(context.Background(), PostCreated{PostEvent: PostEvent{PostID: "a"}})
	rec.Publish(context.Background(), PostUpdated{PostEvent: PostEvent{PostID: "b"}})
	if rec.Len() != 2 {
		t.Fatalf("len = %d", rec.Len())
	}
	got := rec.Snapshot()
	created, ok := got[0].(PostCreated)
	if !ok || created.PostID != "a" {
		t.Fatalf("first = %#v", got[0])
	}
	updated, ok := got[1].(PostUpdated)
	if !ok || updated.PostID != "b" {
		t.Fatalf("second = %#v", got[1])
	}
}

func TestBusDeliversInOrder(t *testing.T) {
	t.Parallel()

	bus := New()
	defer bus.Close()

	var mu sync.Mutex
	var got []string
	done := make(chan struct{})
	bus.Subscribe(func(_ context.Context, ev any) {
		mu.Lock()
		got = append(got, ev.(string))
		if len(got) == 3 {
			close(done)
		}
		mu.Unlock()
	})

	ctx := context.Background()
	bus.Publish(ctx, "one")
	bus.Publish(ctx, "two")
	bus.Publish(ctx, "three")

	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for events")
	}
	mu.Lock()
	defer mu.Unlock()
	if len(got) != 3 || got[0] != "one" || got[1] != "two" || got[2] != "three" {
		t.Fatalf("got %v", got)
	}
}

func TestBusDropsWhenFull(t *testing.T) {
	t.Parallel()

	bus := newBus(1, false)
	bus.Publish(context.Background(), "kept")
	bus.Publish(context.Background(), "dropped")

	select {
	case ev := <-bus.ch:
		if ev != "kept" {
			t.Fatalf("got %v", ev)
		}
	default:
		t.Fatal("expected buffered event")
	}
	select {
	case ev := <-bus.ch:
		t.Fatalf("unexpected extra event %v", ev)
	default:
	}
}

func TestNopReleasesPostCreated(t *testing.T) {
	t.Parallel()

	released := false
	Nop{}.Publish(context.Background(), PostCreated{Release: func() { released = true }})
	if !released {
		t.Fatal("Nop.Publish did not release held upload")
	}
}

func TestBusDropReleasesPostCreated(t *testing.T) {
	t.Parallel()

	bus := newBus(1, false)
	bus.Publish(context.Background(), "kept")
	released := false
	bus.Publish(context.Background(), PostCreated{Release: func() { released = true }})
	if !released {
		t.Fatal("dropped PostCreated did not release held upload")
	}
}

func TestSubscribeReleasePostedToDiscord(t *testing.T) {
	t.Parallel()

	bus := New()
	defer bus.Close()
	SubscribeRelease(bus)

	released := make(chan struct{})
	bus.Publish(context.Background(), PostedToDiscord{
		PostID:  "post-1",
		Release: func() { close(released) },
	})
	select {
	case <-released:
	case <-time.After(time.Second):
		t.Fatal("PostedToDiscord did not release")
	}
}