tests: add integration test suite (llm-d#177)

* tests: add integration test suite

Signed-off-by: Vishwanath Suresh <vsuresh@redhat.com>

* fix lint issues

Signed-off-by: Vishwanath Suresh <vsuresh@redhat.com>

* modify some integration tests

Signed-off-by: Vishwanath Suresh <vsuresh@redhat.com>

---------

Signed-off-by: Vishwanath Suresh <vsuresh@redhat.com>

Author: vishbhat

.github/workflows/ci-pr-checks.yaml

@@ -58,6 +58,9 @@ jobs:
       - name: Test
         run: make test
 
+      - name: Integration tests
+        run: make test-integration
+
   # Python: lint (only if Python files exist)
   python-lint:
     runs-on: ubuntu-latest

CONTRIBUTING.md

@@ -88,6 +88,32 @@ We use three tiers of testing:
 
 Strong e2e coverage is required for deployed systems to prevent performance regression. Appropriate test coverage is an important part of code review.
 
+### Running Tests Locally
+
+```bash
+# Unit tests (default, no external dependencies)
+make test
+
+# Integration tests (each test spawns its own in-process mock server)
+make test-integration
+
+# Both unit and integration tests
+make test-all
+
+# E2E tests (requires docker/podman, kind, helm, kubectl)
+make test-e2e
+```
+
+Integration tests use the `//go:build integration` build tag and live in `test/integration/`. They are excluded from `make test` and only run when the tag is explicitly passed via `make test-integration`. Each test spawns its own mock server (miniredis, `httptest.Server`) — no external services are required.
+
+To add a new integration test, create a `*_test.go` file in `test/integration/` with the build tag:
+
+```go
+//go:build integration
+
+package integration_test
+```
+
 ## Security
 
 See [SECURITY.md](SECURITY.md) for our vulnerability disclosure process.

Makefile

@@ -177,6 +177,16 @@ test-e2e: ## Run e2e integration tests against a Kind cluster
 		$(if $(FOCUS),-ginkgo.focus="$(FOCUS)",) \
 		$(if $(SKIP),-ginkgo.skip="$(SKIP)",)
 
+.PHONY: test-integration
+test-integration: ## Run integration tests (each test spawns its own mock server)
+	@echo "Running integration tests..."
+	@go test -v -tags=integration ./test/integration/ || \
+		(echo "❌ Integration tests failed" && exit 1)
+	@echo "✅ Integration tests passed!"
+
+.PHONY: test-all
+test-all: test test-integration ## Run all tests (unit + integration)
+
 .PHONY: check-dco
 check-dco: ## Check that all commits since main have a DCO Signed-off-by trailer
 	@scripts/check-dco.sh

test/integration/gate_factory_redis_test.go

@@ -0,0 +1,143 @@
+//go:build integration
+
+package integration_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/llm-d-incubation/llm-d-async/pipeline"
+	"github.com/llm-d-incubation/llm-d-async/pkg/async/inference/flowcontrol"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestGateFactory_RedisQuota_ConcurrencyParsing validates that GateFactory
+// correctly parses "redis-quota" params and produces a working AttributeGate.
+// Packages exercised: flowcontrol (GateFactory) -> pkg/redis (RedisQuotaGate).
+func TestGateFactory_RedisQuota_ConcurrencyParsing(t *testing.T) {
+	s := miniredis.RunT(t)
+	factory := flowcontrol.NewGateFactory("")
+
+	gate, err := factory.CreateGate("redis-quota", map[string]string{
+		"address":   s.Addr(),
+		"attribute": "model",
+		"mode":      "concurrency",
+		"limit":     "2",
+		"window":    "30s",
+		"prefix":    "test:",
+	})
+	require.NoError(t, err)
+	require.NotNil(t, gate)
+
+	// Budget always returns 1.0 for quota gates.
+	assert.Equal(t, 1.0, gate.Budget(context.Background()))
+
+	attrGate, ok := gate.(pipeline.AttributeGate)
+	require.True(t, ok, "redis-quota gate should implement AttributeGate")
+
+	ctx := context.Background()
+	attrs := map[string]string{"model": "gpt-4"}
+
+	// Acquire twice (limit=2) — both should succeed.
+	ok1, rel1, err := attrGate.Acquire(ctx, attrs)
+	require.NoError(t, err)
+	assert.True(t, ok1)
+
+	ok2, rel2, err := attrGate.Acquire(ctx, attrs)
+	require.NoError(t, err)
+	assert.True(t, ok2)
+
+	// Third acquire — should fail.
+	ok3, _, err := attrGate.Acquire(ctx, attrs)
+	require.NoError(t, err)
+	assert.False(t, ok3, "Third acquire should be denied (limit=2)")
+
+	// Release one and retry.
+	rel1()
+	ok4, rel4, err := attrGate.Acquire(ctx, attrs)
+	require.NoError(t, err)
+	assert.True(t, ok4, "Should succeed after release")
+
+	// Cleanup.
+	rel2()
+	if rel4 != nil {
+		rel4()
+	}
+}
+
+// TestGateFactory_RedisQuota_RateLimitParsing validates rate-limit mode parsing.
+func TestGateFactory_RedisQuota_RateLimitParsing(t *testing.T) {
+	s := miniredis.RunT(t)
+	factory := flowcontrol.NewGateFactory("")
+
+	gate, err := factory.CreateGate("redis-quota", map[string]string{
+		"address": s.Addr(),
+		"mode":    "rate-limit",
+		"limit":   "3",
+		"window":  "1m",
+	})
+	require.NoError(t, err)
+	require.NotNil(t, gate)
+
+	attrGate, ok := gate.(pipeline.AttributeGate)
+	require.True(t, ok)
+
+	ctx := context.Background()
+	attrs := map[string]string{"userid": "alice"}
+
+	for i := 0; i < 3; i++ {
+		ok, _, err := attrGate.Acquire(ctx, attrs)
+		require.NoError(t, err)
+		assert.True(t, ok, "Request %d should be allowed", i+1)
+	}
+
+	// Fourth should be rate limited.
+	ok4, _, err := attrGate.Acquire(ctx, attrs)
+	require.NoError(t, err)
+	assert.False(t, ok4, "Fourth request should be rate limited")
+}
+
+// TestGateFactory_RedisQuota_MissingParams validates error handling for missing
+// required parameters.
+func TestGateFactory_RedisQuota_MissingParams(t *testing.T) {
+	factory := flowcontrol.NewGateFactory("")
+
+	_, err := factory.CreateGate("redis-quota", map[string]string{
+		"limit": "5",
+	})
+	assert.Error(t, err, "Should fail when address is missing")
+
+	s := miniredis.RunT(t)
+	_, err = factory.CreateGate("redis-quota", map[string]string{
+		"address": s.Addr(),
+	})
+	assert.Error(t, err, "Should fail when limit is missing")
+}
+
+// TestGateFactory_RedisQuota_DefaultParams verifies that omitted optional params
+// fall back to documented defaults (attribute=userid, mode=rate-limit, etc.).
+func TestGateFactory_RedisQuota_DefaultParams(t *testing.T) {
+	s := miniredis.RunT(t)
+	factory := flowcontrol.NewGateFactory("")
+
+	gate, err := factory.CreateGate("redis-quota", map[string]string{
+		"address": s.Addr(),
+		"limit":   "1",
+	})
+	require.NoError(t, err)
+
+	attrGate, ok := gate.(pipeline.AttributeGate)
+	require.True(t, ok)
+
+	// Default attribute is "userid", default mode is "rate-limit".
+	ctx := context.Background()
+	ok1, _, err := attrGate.Acquire(ctx, map[string]string{"userid": "bob"})
+	require.NoError(t, err)
+	assert.True(t, ok1)
+
+	ok2, _, err := attrGate.Acquire(ctx, map[string]string{"userid": "bob"})
+	require.NoError(t, err)
+	assert.False(t, ok2, "Second acquire should be rate limited with default params")
+}

test/integration/merge_policy_test.go

@@ -0,0 +1,98 @@
+//go:build integration
+
+package integration_test
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	asyncapi "github.com/llm-d-incubation/llm-d-async/api"
+	"github.com/llm-d-incubation/llm-d-async/pipeline"
+	ap "github.com/llm-d-incubation/llm-d-async/pkg/async"
+	"github.com/stretchr/testify/assert"
+)
+
+// TestRandomRobinPolicy_ConcurrentProducers validates that RandomRobinPolicy
+// correctly merges messages from multiple channels when multiple goroutines
+// are producing concurrently. Every message sent must appear exactly once
+// on the merged output channel.
+func TestRandomRobinPolicy_ConcurrentProducers(t *testing.T) {
+	const numChannels = 4
+	const msgsPerChannel = 25
+
+	channels := make([]pipeline.RequestChannel, numChannels)
+	for i := range numChannels {
+		channels[i] = pipeline.RequestChannel{
+			Channel:            make(chan *asyncapi.InternalRequest, msgsPerChannel),
+			IGWBaseURL:         "http://localhost:8080",
+			InferenceObjective: "latency",
+			RequestPathURL:     "/v1/completions",
+		}
+	}
+
+	policy := ap.NewRandomRobinPolicy()
+	merged := policy.MergeRequestChannels(channels)
+
+	// Produce messages concurrently on all channels.
+	var wg sync.WaitGroup
+	for chIdx := range numChannels {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+			for m := range msgsPerChannel {
+				ir := asyncapi.NewInternalRequest(
+					asyncapi.InternalRouting{},
+					&asyncapi.RequestMessage{
+						ID:       msgID(idx, m),
+						Created:  time.Now().Unix(),
+						Deadline: time.Now().Add(time.Minute).Unix(),
+						Payload:  map[string]any{"model": "test"},
+					},
+				)
+				channels[idx].Channel <- ir
+			}
+			close(channels[idx].Channel)
+		}(chIdx)
+	}
+
+	// Consume all messages from the merged channel.
+	received := make(map[string]bool)
+	done := make(chan struct{})
+	go func() {
+		for msg := range merged.Channel {
+			received[msg.InternalRequest.PublicRequest.ReqID()] = true
+		}
+		close(done)
+	}()
+
+	wg.Wait()
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("Timed out waiting for merged channel to close")
+	}
+
+	expected := numChannels * msgsPerChannel
+	assert.Equal(t, expected, len(received), "Expected %d unique messages, got %d", expected, len(received))
+
+	// Verify every expected message was received.
+	for chIdx := range numChannels {
+		for m := range msgsPerChannel {
+			id := msgID(chIdx, m)
+			assert.True(t, received[id], "Missing message %s", id)
+		}
+	}
+}
+
+func msgID(channelIdx, msgIdx int) string {
+	return "ch" + itoa(channelIdx) + "-msg" + itoa(msgIdx)
+}
+
+func itoa(i int) string {
+	const digits = "0123456789"
+	if i < 10 {
+		return string(digits[i])
+	}
+	return itoa(i/10) + string(digits[i%10])
+}