mirror of
https://github.com/google/ax.git
synced 2026-10-21 05:49:14 +08:00
Remove the Gateway resource across the API, controller, server, CLI, store, and documentation. Gateway introduced an unnecessary layer and architectural bifurcation between Agent Substrate and AX. By removing it, AX aligns directly with Substrate's native primitives and networking rather than maintaining an overlapping ingress/routing abstraction. - Reserved protobuf field 8 and removed Gateway RPCs and messages. - Cleaned up controller reconciler, server endpoints, and store models. - Updated CLI, documentation, and example manifests.
279 lines
9.7 KiB
Go
279 lines
9.7 KiB
Go
// Copyright 2026 Google LLC
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package server_test
|
|
|
|
import (
|
|
"context"
|
|
"net"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/ax/internal/server"
|
|
"github.com/google/ax/internal/store/memory"
|
|
"github.com/google/ax/pkg/apis/v1alpha1"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
"google.golang.org/grpc/status"
|
|
)
|
|
|
|
func TestServerHealthzHTTP(t *testing.T) {
|
|
memStore := memory.NewStore()
|
|
srv := server.NewServer(memStore)
|
|
|
|
req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
|
|
rec := httptest.NewRecorder()
|
|
srv.Handler().ServeHTTP(rec, req)
|
|
|
|
if rec.Code != http.StatusOK {
|
|
t.Fatalf("expected 200 OK from /healthz, got %d", rec.Code)
|
|
}
|
|
if rec.Body.String() != "ok\n" {
|
|
t.Fatalf("expected 'ok\\n' from /healthz, got %q", rec.Body.String())
|
|
}
|
|
|
|
// Verify non-healthz HTTP returns 404
|
|
req404 := httptest.NewRequest(http.MethodGet, "/api/v1/tasks", nil)
|
|
rec404 := httptest.NewRecorder()
|
|
srv.Handler().ServeHTTP(rec404, req404)
|
|
if rec404.Code != http.StatusNotFound {
|
|
t.Fatalf("expected 404 Not Found for /api/v1/tasks, got %d", rec404.Code)
|
|
}
|
|
}
|
|
|
|
func TestServerGRPC(t *testing.T) {
|
|
memStore := memory.NewStore()
|
|
srv := server.NewServer(memStore)
|
|
|
|
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatalf("failed to listen: %v", err)
|
|
}
|
|
defer ln.Close()
|
|
|
|
httpServer := &http.Server{
|
|
Handler: srv.Handler(),
|
|
}
|
|
httpServer.Protocols = new(http.Protocols)
|
|
httpServer.Protocols.SetHTTP1(true)
|
|
httpServer.Protocols.SetUnencryptedHTTP2(true)
|
|
|
|
go func() {
|
|
_ = httpServer.Serve(ln)
|
|
}()
|
|
defer httpServer.Close()
|
|
|
|
conn, err := grpc.NewClient(ln.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
|
|
if err != nil {
|
|
t.Fatalf("failed to dial gRPC: %v", err)
|
|
}
|
|
defer conn.Close()
|
|
|
|
client := v1alpha1.NewAXClient(conn)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
// 1. Create one resource of each kind through the typed RPCs. Metadata is left
|
|
// partially empty to exercise server-side defaulting.
|
|
if _, err := client.UpdateWorkspace(ctx, &v1alpha1.UpdateWorkspaceRequest{Workspace: &v1alpha1.Workspace{
|
|
Metadata: &v1alpha1.ObjectMeta{Name: "grpc-ws"},
|
|
Spec: &v1alpha1.WorkspaceSpec{},
|
|
}}); err != nil {
|
|
t.Fatalf("UpdateWorkspace failed: %v", err)
|
|
}
|
|
if _, err := client.UpdateModel(ctx, &v1alpha1.UpdateModelRequest{Model: &v1alpha1.Model{
|
|
Metadata: &v1alpha1.ObjectMeta{Name: "grpc-model"},
|
|
Spec: &v1alpha1.ModelSpec{Provider: "google", Model: "gemini-3.8-flash"},
|
|
}}); err != nil {
|
|
t.Fatalf("UpdateModel failed: %v", err)
|
|
}
|
|
if _, err := client.UpdateTask(ctx, &v1alpha1.UpdateTaskRequest{Task: &v1alpha1.Task{
|
|
Metadata: &v1alpha1.ObjectMeta{Name: "grpc-task"},
|
|
Spec: &v1alpha1.TaskSpec{Image: "alpine"},
|
|
}}); err != nil {
|
|
t.Fatalf("UpdateTask failed: %v", err)
|
|
}
|
|
|
|
// 2. Defaulting applies to every kind: atespace and creation timestamp are filled in.
|
|
ws, err := client.GetWorkspace(ctx, &v1alpha1.GetWorkspaceRequest{Atespace: "default", Name: "grpc-ws"})
|
|
if err != nil {
|
|
t.Fatalf("GetWorkspace failed: %v", err)
|
|
}
|
|
if ws.GetMetadata().GetAtespace() != "default" || ws.GetMetadata().GetCreationTimestamp() == nil {
|
|
t.Errorf("expected workspace metadata to be defaulted, got %v", ws.GetMetadata())
|
|
}
|
|
|
|
// 3. GetTask & ListTasks
|
|
task, err := client.GetTask(ctx, &v1alpha1.GetTaskRequest{Atespace: "default", Name: "grpc-task"})
|
|
if err != nil {
|
|
t.Fatalf("GetTask failed: %v", err)
|
|
}
|
|
if task.Metadata.Name != "grpc-task" {
|
|
t.Errorf("expected name 'grpc-task', got %s", task.Metadata.Name)
|
|
}
|
|
if task.Metadata.CreationTimestamp == nil {
|
|
t.Errorf("expected creation timestamp to be populated on applied task")
|
|
}
|
|
|
|
listTasksResp, err := client.ListTasks(ctx, &v1alpha1.ListTasksRequest{Atespace: "default"})
|
|
if err != nil {
|
|
t.Fatalf("ListTasks failed: %v", err)
|
|
}
|
|
if len(listTasksResp.Tasks) != 1 {
|
|
t.Fatalf("expected 1 task in list, got %d", len(listTasksResp.Tasks))
|
|
}
|
|
if listTasksResp.Tasks[0].Metadata.CreationTimestamp == nil {
|
|
t.Errorf("expected creation timestamp on listed task")
|
|
}
|
|
|
|
// Test UpdateTask
|
|
task.Spec.Image = "ghcr.io/test/updated-image"
|
|
updatedTask, err := client.UpdateTask(ctx, &v1alpha1.UpdateTaskRequest{Task: task})
|
|
if err != nil {
|
|
t.Fatalf("UpdateTask failed: %v", err)
|
|
}
|
|
if updatedTask.Spec.Image != "ghcr.io/test/updated-image" {
|
|
t.Errorf("expected image 'ghcr.io/test/updated-image', got %s", updatedTask.Spec.Image)
|
|
}
|
|
|
|
// 4. Suspend & Resume Task
|
|
suspTask, err := client.SuspendTask(ctx, &v1alpha1.SuspendTaskRequest{Atespace: "default", Name: "grpc-task"})
|
|
if err != nil {
|
|
t.Fatalf("SuspendTask failed: %v", err)
|
|
}
|
|
if !suspTask.Spec.Suspend {
|
|
t.Errorf("expected task to be suspended")
|
|
}
|
|
|
|
resTask, err := client.ResumeTask(ctx, &v1alpha1.ResumeTaskRequest{Atespace: "default", Name: "grpc-task"})
|
|
if err != nil {
|
|
t.Fatalf("ResumeTask failed: %v", err)
|
|
}
|
|
if resTask.Spec.Suspend {
|
|
t.Errorf("expected task to be resumed")
|
|
}
|
|
|
|
|
|
// 5. Workspaces
|
|
ws, err = client.GetWorkspace(ctx, &v1alpha1.GetWorkspaceRequest{Atespace: "default", Name: "grpc-ws"})
|
|
if err != nil {
|
|
t.Fatalf("GetWorkspace failed: %v", err)
|
|
}
|
|
if ws.Metadata.Name != "grpc-ws" {
|
|
t.Errorf("expected workspace 'grpc-ws', got %s", ws.Metadata.Name)
|
|
}
|
|
|
|
listWorkspacesResp, err := client.ListWorkspaces(ctx, &v1alpha1.ListWorkspacesRequest{Atespace: "default"})
|
|
if err != nil {
|
|
t.Fatalf("ListWorkspaces failed: %v", err)
|
|
}
|
|
if len(listWorkspacesResp.Workspaces) != 1 {
|
|
t.Fatalf("expected 1 workspace in list, got %d", len(listWorkspacesResp.Workspaces))
|
|
}
|
|
|
|
// 7. Models
|
|
model, err := client.GetModel(ctx, &v1alpha1.GetModelRequest{Atespace: "default", Name: "grpc-model"})
|
|
if err != nil {
|
|
t.Fatalf("GetModel failed: %v", err)
|
|
}
|
|
if model.Metadata.Name != "grpc-model" {
|
|
t.Errorf("expected model 'grpc-model', got %s", model.Metadata.Name)
|
|
}
|
|
|
|
listModelsResp, err := client.ListModels(ctx, &v1alpha1.ListModelsRequest{Atespace: "default"})
|
|
if err != nil {
|
|
t.Fatalf("ListModels failed: %v", err)
|
|
}
|
|
if len(listModelsResp.Models) != 1 {
|
|
t.Fatalf("expected 1 model in list, got %d", len(listModelsResp.Models))
|
|
}
|
|
|
|
// 7b. WatchTask (should receive INITIAL state)
|
|
stream, err := client.WatchTask(ctx, &v1alpha1.WatchTaskRequest{Atespace: "default", Name: "grpc-task"})
|
|
if err != nil {
|
|
t.Fatalf("WatchTask failed: %v", err)
|
|
}
|
|
watchMsg, err := stream.Recv()
|
|
if err != nil {
|
|
t.Fatalf("WatchTask Recv failed: %v", err)
|
|
}
|
|
if watchMsg.Action != "INITIAL" {
|
|
t.Errorf("expected INITIAL action, got %s", watchMsg.Action)
|
|
}
|
|
if watchMsg.Task == nil || watchMsg.Task.Metadata == nil || watchMsg.Task.Metadata.Name != "grpc-task" {
|
|
t.Errorf("expected task 'grpc-task' in watch, got %+v", watchMsg.Task)
|
|
}
|
|
|
|
// 8. Delete operations
|
|
// Task deletion is two-phase: the RPC marks the task Terminating and the
|
|
// controller removes the record after tearing down the actor.
|
|
if _, err := client.DeleteTask(ctx, &v1alpha1.DeleteTaskRequest{Atespace: "default", Name: "grpc-task"}); err != nil {
|
|
t.Fatalf("DeleteTask failed: %v", err)
|
|
}
|
|
terminating, err := client.GetTask(ctx, &v1alpha1.GetTaskRequest{Atespace: "default", Name: "grpc-task"})
|
|
if err != nil {
|
|
t.Fatalf("GetTask after DeleteTask failed: %v", err)
|
|
}
|
|
if terminating.GetStatus().GetPhase() != v1alpha1.PhaseTerminating {
|
|
t.Errorf("expected phase %q after DeleteTask, got %q", v1alpha1.PhaseTerminating, terminating.GetStatus().GetPhase())
|
|
}
|
|
if _, err := client.DeleteTask(ctx, &v1alpha1.DeleteTaskRequest{Atespace: "default", Name: "no-such-task"}); status.Code(err) != codes.NotFound {
|
|
t.Errorf("expected NotFound deleting a missing task, got %v", err)
|
|
}
|
|
// Stand in for the controller finishing cleanup.
|
|
if err := memStore.DeleteTask(ctx, "default", "grpc-task"); err != nil {
|
|
t.Fatalf("removing task record failed: %v", err)
|
|
}
|
|
if _, err := client.DeleteWorkspace(ctx, &v1alpha1.DeleteWorkspaceRequest{Atespace: "default", Name: "grpc-ws"}); err != nil {
|
|
t.Fatalf("DeleteWorkspace failed: %v", err)
|
|
}
|
|
if _, err := client.DeleteModel(ctx, &v1alpha1.DeleteModelRequest{Atespace: "default", Name: "grpc-model"}); err != nil {
|
|
t.Fatalf("DeleteModel failed: %v", err)
|
|
}
|
|
|
|
// Check 404 after delete
|
|
_, err = client.GetTask(ctx, &v1alpha1.GetTaskRequest{Atespace: "default", Name: "grpc-task"})
|
|
if status.Code(err) != codes.NotFound {
|
|
t.Fatalf("expected NotFound code, got %v", err)
|
|
}
|
|
}
|
|
|
|
func TestUpdateTask_ValidatesWorkspaceBindings(t *testing.T) {
|
|
srv := server.NewServer(memory.NewStore())
|
|
ctx := context.Background()
|
|
|
|
_, err := srv.UpdateTask(ctx, &v1alpha1.UpdateTaskRequest{Task: &v1alpha1.Task{
|
|
Metadata: &v1alpha1.ObjectMeta{Name: "bad"},
|
|
Spec: &v1alpha1.TaskSpec{
|
|
Workspaces: []*v1alpha1.WorkspaceRef{{Name: "a", Path: "/same"}, {Name: "b", Path: "/same"}},
|
|
},
|
|
}})
|
|
if status.Code(err) != codes.InvalidArgument {
|
|
t.Fatalf("expected InvalidArgument for colliding workspace paths, got %v", err)
|
|
}
|
|
|
|
_, err = srv.UpdateTask(ctx, &v1alpha1.UpdateTaskRequest{Task: &v1alpha1.Task{
|
|
Metadata: &v1alpha1.ObjectMeta{Name: "good"},
|
|
Spec: &v1alpha1.TaskSpec{
|
|
Workspaces: []*v1alpha1.WorkspaceRef{{Name: "a"}, {Name: "b"}},
|
|
},
|
|
}})
|
|
if err != nil {
|
|
t.Fatalf("expected a valid multi-workspace task to be accepted, got %v", err)
|
|
}
|
|
}
|