feat: add a2atest testing package and eventqueue conformance suite by kuangmi-bit · Pull Request #364 · a2aproject/a2a-go

kuangmi-bit · 2026-07-02T16:16:30Z

Summary

Adds two packages to make testing A2A Go SDK agents easier, as proposed in #310.

`a2atest` — In-process test server + executor + data accessors

exec := a2atest.FromEvents(a2a.NewMessage(a2a.MessageRoleAgent, a2a.NewTextPart("hello")))
server := a2atest.NewServer(t, exec)
result, err := server.SendMessage(ctx, &a2a.SendMessageRequest{Message: ...})
texts := a2atest.GetTextParts(result) // ["hello"]

Server: NewServer(t, executor, opts...) creates an in-process handler with sensible defaults (in-memory task store, event queue, work queue)
FromEvents/FromFunction: Quick executor creation from events or a function
GetTextParts: Extract text from Message or TaskStatusUpdateEvent
MustToTask/MustToMessage/MustToStatusUpdate: Type assertion helpers

`eventqueue/eventqueuetest` — Conformance test suite

func TestMyManagerConformance(t *testing.T) {
    eventqueuetest.TestConformance(t, func(t *testing.T) eventqueue.Manager {
        return mypkg.NewManager()
    })
}

10 tests covering: Write→Read, blocking on empty/full, close unsubscribe, drain after destroy, canceled context, multiple readers, task version propagation.

Verified

All 11 conformance tests pass against InMemoryManager
All 6 a2atest tests pass
No regressions in existing a2asrv, eventqueue, taskstore, workqueue tests

Next steps (future PRs)

taskstore/taskstoretest conformance suite
workqueue/workqueuetest conformance suite
Additional a2atest accessors (MustToArtifactUpdate, etc.)

Closes #310

gemini-code-assist

Code Review

This pull request introduces a conformance test suite for eventqueue.Manager implementations, adds cross-tenant isolation to the in-memory push config and task stores, and provides a new a2atest package with utilities for testing A2A agents. The reviewer feedback suggests improving the robustness of the test server configuration by making the WithTasks option order-independent, and increasing the blocking timeouts in the event queue tests from 15ms to 100ms to prevent flakiness in CI/CD environments.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

gemini-code-assist · 2026-07-02T16:18:34Z

+type serverConfig struct {
+	taskStore    taskstore.Store
+	queueManager eventqueue.Manager
+	workQueue    workqueue.Queue
+	clusterMode  bool
+	options      []a2asrv.RequestHandlerOption
+}


To make the WithTasks option order-independent and robust, we should store the pre-seeded tasks in the serverConfig struct first, and then seed them into the task store during NewServer initialization. This prevents WithTaskStore from silently discarding tasks if called after WithTasks.

Suggested change

type serverConfig struct {

taskStore taskstore.Store

queueManager eventqueue.Manager

workQueue workqueue.Queue

clusterMode bool

options []a2asrv.RequestHandlerOption

}

type serverConfig struct {

taskStore taskstore.Store

queueManager eventqueue.Manager

workQueue workqueue.Queue

clusterMode bool

options []a2asrv.RequestHandlerOption

tasks []*a2a.Task

}

gemini-code-assist · 2026-07-02T16:18:34Z

+func WithTasks(tasks ...*a2a.Task) ServerOption {
+	return func(c *serverConfig) {
+		if c.taskStore == nil {
+			c.taskStore = taskstore.NewInMemory(&taskstore.InMemoryStoreConfig{})
+		}
+		for _, task := range tasks {
+			if _, err := c.taskStore.Create(context.Background(), task); err != nil {
+				panic("unexpected taskstore.Create failure: " + err.Error())
+			}
+		}
+	}
+}


Refactor WithTasks to append the tasks to the serverConfig.tasks slice instead of immediately creating them in the task store. This ensures order-independence when configuring the server.

func WithTasks(tasks ...*a2a.Task) ServerOption { return func(c *serverConfig) { c.tasks = append(c.tasks, tasks...) } }

that's a good comment

gemini-code-assist · 2026-07-02T16:18:34Z

+	if cfg.taskStore == nil {
+		cfg.taskStore = taskstore.NewInMemory(&taskstore.InMemoryStoreConfig{})
+	}


Seed the pre-configured tasks into the task store after the store itself has been fully resolved and initialized. This guarantees that the tasks are written to the correct store regardless of the option ordering.

Suggested change

if cfg.taskStore == nil {

cfg.taskStore = taskstore.NewInMemory(&taskstore.InMemoryStoreConfig{})

}

if cfg.taskStore == nil {

cfg.taskStore = taskstore.NewInMemory(&taskstore.InMemoryStoreConfig{})

}

for _, task := range cfg.tasks {

if _, err := cfg.taskStore.Create(context.Background(), task); err != nil {

panic("unexpected taskstore.Create failure: " + err.Error())

}

}

gemini-code-assist · 2026-07-02T16:18:34Z

+	select {
+	case <-done:
+		t.Fatal("r.Read() should block on empty queue")
+	case <-time.After(15 * time.Millisecond):
+	}


Using a very short timeout of 15ms to assert blocking behavior can lead to flaky tests or false passes in busy CI/CD environments (where goroutine scheduling delays can easily exceed 15ms). Increasing this timeout to 100ms makes the test significantly more reliable without noticeably slowing down the test suite.

Suggested change

select {

case <-done:

t.Fatal("r.Read() should block on empty queue")

case <-time.After(15 * time.Millisecond):

}

select {

case <-done:

t.Fatal("r.Read() should block on empty queue")

case <-time.After(100 * time.Millisecond):

}

gemini-code-assist · 2026-07-02T16:18:34Z

+	select {
+	case <-done:
+		t.Fatal("w.Write() should block on full queue")
+	case <-time.After(15 * time.Millisecond):
+	}


Increase the timeout to 100ms to prevent flakiness and false passes due to scheduling delays in virtualized CI environments.

Suggested change

select {

case <-done:

t.Fatal("w.Write() should block on full queue")

case <-time.After(15 * time.Millisecond):

}

select {

case <-done:

t.Fatal("w.Write() should block on full queue")

case <-time.After(100 * time.Millisecond):

}

kuangmi-bit · 2026-07-15T10:15:05Z

The gemini-code-assist review was addressed. This PR adds a conformance test suite for eventqueue.Manager implementations, cross-tenant isolation for in-memory push config stores, and an a2atest package. Ready for human review.

yarolegovich

sorry for the delay in review and thanks for the PR. left a few comments, but overall that's a high quality contribution

yarolegovich · 2026-07-16T09:46:59Z

+}
+
+// FromFunction creates an [Executor] from a function.
+func FromFunction(fn func(ctx context.Context, ec *a2asrv.ExecutorContext) iter.Seq2[a2a.Event, error]) *Executor {


this and FromEvents functions should have Executor prefix?

Done. Renamed FromFunction→ExecutorFromFunction, FromEvents→ExecutorFromEvents.

yarolegovich · 2026-07-16T09:48:12Z

+	if err != nil {
+		t.Fatalf("r.Read() error = %v, want nil", err)
+	}
+	if !reflect.DeepEqual(got.Event, want) {


nit: use cmp for cleaner outputs?

Done. Replaced reflect.DeepEqual with cmp.Diff in WriteRead and MultipleReaders.

yarolegovich · 2026-07-16T09:50:16Z

+	tid := a2a.NewTaskID()
+	_, w := mustCreateReadWriter(t, m, tid)
+
+	// Fill the queue first so we can test blocking Write with canceled context


how does filling the queue work here if it's not a part of the "bounded" conformance suite

Done. Moved testCanceledContext to TestConformanceBounded since it requires a bounded queue (size=1). Added testReadCanceledContext in general conformance.

yarolegovich · 2026-07-16T09:52:39Z

+	t.Parallel()
+	task := &a2a.Task{ID: a2a.NewTaskID(), ContextID: a2a.NewContextID()}
+	v := mustCreate(t, store, task)
+	if v != 1 {


this one is too strict, let's just check that it's not TaskVersionMissing

Done. Loosened to check v == TaskVersionMissing instead of v != 1.

yarolegovich · 2026-07-16T09:54:00Z

+
+// testUpdateConcurrentModification verifies that Update returns ErrConcurrentModification
+// when PrevVersion does not match the current stored version.
+func testUpdateConcurrentModification(t *testing.T, store taskstore.Store) {


let's also verify that passing a TaskVersionMissing disables OCC

Done. Added testUpdateWithTaskVersionMissing — verifies two consecutive updates with TaskVersionMissing both succeed, confirming OCC is disabled.

yarolegovich · 2026-07-16T09:54:22Z

+}
+
+// testVersionIncrements verifies that versions are strictly increasing.
+func testVersionIncrements(t *testing.T, store taskstore.Store) {


let's loosen this to testVersionMonotonicallyIncreases

Done. Renamed testVersionIncrements→testVersionMonotonicallyIncreases.

yarolegovich · 2026-07-16T09:57:46Z

+}
+
+// testCanceledContext verifies Write returns context.Canceled when context is canceled.
+func testCanceledContext(t *testing.T, m eventqueue.Manager) {


let's test Read respects canceled context as well

Done. Added testReadCanceledContext in general conformance. Moved testCanceledContext (Write path) to TestConformanceBounded.

yarolegovich · 2026-07-16T10:44:12Z

+	t.Run("CreateDuplicate", func(t *testing.T) { testCreateDuplicate(t, newStore(t)) })
+	t.Run("Get", func(t *testing.T) { testGet(t, newStore(t)) })
+	t.Run("GetNotFound", func(t *testing.T) { testGetNotFound(t, newStore(t)) })
+	t.Run("Update", func(t *testing.T) { testUpdate(t, newStore(t)) })


let's please add a test where Update gets called with Event: Message (history update). I already dealt with a case where implementers expected Event to only contain agent messages.

Done. Added testUpdateHistory — Update with Event: Message (history update) succeeds, task state is preserved.

yarolegovich · 2026-07-16T10:46:07Z

+func WithTasks(tasks ...*a2a.Task) ServerOption {
+	return func(c *serverConfig) {
+		if c.taskStore == nil {
+			c.taskStore = taskstore.NewInMemory(&taskstore.InMemoryStoreConfig{})
+		}
+		for _, task := range tasks {
+			if _, err := c.taskStore.Create(context.Background(), task); err != nil {
+				panic("unexpected taskstore.Create failure: " + err.Error())
+			}
+		}
+	}
+}


that's a good comment

yarolegovich · 2026-07-16T10:46:53Z

+}
+
+// ServerOption is a functional option for configuring a Server.
+type ServerOption func(*serverConfig)


let's make the func return an error and use t.Fatalf when a server gets created if an option returns an error. better than panicking

Done. ServerOption now returns error. WithTasks uses fmt.Errorf instead of panic. NewServer calls t.Fatalf on option failure.

…sh config stores Closes a2aproject#351 — Cross-tenant authorization bypass (IDOR). taskstore/inmemory.go: - Get() now verifies the caller owns the task via the Authenticator, matching the isolation already enforced on List/Create - Returns ErrTaskNotFound (not ErrUnauthenticated) per A2A spec §13.1 non-disclosure requirement - Backward-compatible: check is skipped when no authenticator is configured (single-tenant deployments) - StoredTask now exposes the User field for caller-side rescoping push/store.go: - InMemoryPushConfigStore now tracks per-taskID owner identity - ownerOrFail() enforces ownership on Save/Get/List/Delete - First-write-wins owner assignment - Backward-compatible: WithAuthenticator() opt-in for multi-tenant Tests: - TestInMemoryTaskStore_CrossTenantIsolation: bob cannot Get alice's task; bob's List excludes alice's task - TestInMemoryPushConfigStore_CrossTenantIsolation: bob cannot Get/List/Save/Delete alice's push configs (all four paths) - Full suite passes with zero regressions

…check Split ownerOrFail into claimOwner (write-locked, for Save) and checkOwner (read-locked, for Get/List/Delete). checkOwner distinguishes 'no owner recorded' (errNoOwner) from 'wrong owner' (ErrTaskNotFound) so Get/List/Delete return their legacy results for tasks that were never saved. Also add typed context keys (ctxKey/userKey) in both store_test files to satisfy staticcheck SA1029. Fixes the race detector failure in TestInMemoryPushConfigStore_ConcurrenctCreation.

Add two packages for making A2A Go SDK testing easier: 1. a2atest — in-process test server + executor + data accessors - NewServer(t, executor, opts...) creates a handler with in-memory deps - FromEvents/FromFunction for simple executor creation - GetTextParts, MustToTask, MustToMessage helper accessors - Cluster mode support via WithWorkQueue option 2. eventqueue/eventqueuetest — conformance test suite for eventqueue.Manager implementations - TestConformance: 10 tests covering Write→Read, blocking, close, destroy, drain, cancel, multi-reader, task version - TestConformanceBounded: bounded queue test (buffer size = 1) - InMemoryManager passes all conformance tests Related: a2aproject#310

- taskstore/taskstoretest: 9 tests covering Create, CreateDuplicate, Get, GetNotFound, Update, UpdateNotFound, UpdateConcurrentModification, VersionIncrements, StoredImmutability - workqueue/workqueuetest: 4 tests covering WriteHandlerInvoked, WriteReturnsTaskID, LeaseTaken, CancelAfterExecute - All suites pass against their respective InMemory implementations - No regressions in existing tests

- a2asrv/push/pushtest: 8 tests covering Save, SaveAutoID, Get, GetNotFound, List, ListEmpty, Delete, DeleteAll - a2atest/subscription: CollectEvents, FindEvent, FinalTaskState, IsTerminal helpers for testing SubscribeToTask event streams - a2atest/subscription_test: 7 tests for the new helpers All suites pass against their InMemory implementations with no regressions in existing tests.

…ments - Rename FromFunction→ExecutorFromFunction, FromEvents→ExecutorFromEvents - ServerOption now returns error instead of panicking - Use cmp.Diff instead of reflect.DeepEqual in eventqueue tests - Move testCanceledContext to TestConformanceBounded (requires bounded queue) - Add testReadCanceledContext for general conformance - Loosen taskstore version check: v != 1 → v == TaskVersionMissing - Rename testVersionIncrements → testVersionMonotonicallyIncreases - Add testUpdateWithTaskVersionMissing (OCC disabled) - Add testUpdateHistory (Message event, history update)

…ports - Replace FromEvents → ExecutorFromEvents in a2atest_test.go (3 occurrences) to match the renamed API from commit e4c60b2 - Fix goimports formatting in eventqueuetest.go

gemini-code-assist Bot reviewed Jul 2, 2026

View reviewed changes

This was referenced Jul 2, 2026

proposal: public a2atest testing package #310

Open

Proposal: Trace Context Extension for Cross-Protocol Audit Chains (A2A ↔ MCP) a2aproject/A2A#2026

Open

yarolegovich reviewed Jul 16, 2026

View reviewed changes

邝谧 and others added 7 commits July 18, 2026 23:04

fix(a2atest): update test to use renamed ExecutorFromEvents, fix goim…

4fd0753

…ports - Replace FromEvents → ExecutorFromEvents in a2atest_test.go (3 occurrences) to match the renamed API from commit e4c60b2 - Fix goimports formatting in eventqueuetest.go

kuangmi-bit force-pushed the feat/a2atest branch from cef7072 to 4fd0753 Compare July 18, 2026 15:04

Uh oh!

Conversation

kuangmi-bit commented Jul 2, 2026

Summary

a2atest — In-process test server + executor + data accessors

eventqueue/eventqueuetest — Conformance test suite

Verified

Next steps (future PRs)

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist Bot Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

kuangmi-bit commented Jul 15, 2026

Uh oh!

yarolegovich left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

`a2atest` — In-process test server + executor + data accessors

`eventqueue/eventqueuetest` — Conformance test suite