package runnerfacade import ( "context" "crypto/tls" "crypto/x509" "net/http" "net/http/httptest" "net/url" "strings" "testing" "connectrpc.com/connect" "gitea.dev/actionslib/pkg/protocol" runnerv1 "gitea.dev/actionslib/runner/v1" "gitea.dev/actionslib/runner/v1/runnerv1connect" "google.golang.org/protobuf/types/known/structpb" "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment" ) type fakeUpstream struct { taskUpdates int logUpdates int } func (u *fakeUpstream) UpdateTask(_ context.Context, request *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) { u.taskUpdates++ return connect.NewResponse(&runnerv1.UpdateTaskResponse{State: request.Msg.State}), nil } func (u *fakeUpstream) UpdateLog(_ context.Context, request *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) { u.logUpdates++ return connect.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: request.Msg.Index + int64(len(request.Msg.Rows))}), nil } func facadeAssignment(t *testing.T) taskassignment.Assignment { t.Helper() fields, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"}) if err != nil { t.Fatal(err) } assignment, err := taskassignment.New(&runnerv1.Task{ Id: 42, Context: fields, WorkflowPayload: []byte("jobs:\n publish:\n runs-on: [self-hosted, pod]\n steps: []\n"), }, "ddupan.top") if err != nil { t.Fatal(err) } return assignment } func testFacade(t *testing.T) (*Facade, taskassignment.Assignment, string) { t.Helper() capabilities, err := NewCapabilities([]byte("0123456789abcdef0123456789abcdef")) if err != nil { t.Fatal(err) } assignment := facadeAssignment(t) registry := NewRegistry() if _, err := registry.Offer(assignment); err != nil { t.Fatal(err) } return &Facade{Registry: registry, Capabilities: capabilities, Upstream: &fakeUpstream{}}, assignment, capabilities.Issue(assignment.ID) } func authenticatedRequest[T any](message *T, assignmentID, token string) *connect.Request[T] { request := connect.NewRequest(message) request.Header().Set(protocol.UUIDHeader, assignmentID) request.Header().Set(protocol.TokenHeader, token) return request } func TestFacadeReturnsOnlyPreassignedTaskAndSignalsClaim(t *testing.T) { facade, assignment, token := testFacade(t) ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID) response, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)) if err != nil { t.Fatal(err) } if response.Msg.GetTask().GetId() != assignment.Task.GetId() { t.Fatalf("task = %#v", response.Msg.GetTask()) } if err := facade.Registry.WaitClaimed(context.Background(), assignment.ID); err != nil { t.Fatal(err) } if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); connect.CodeOf(err) != connect.CodeFailedPrecondition { t.Fatalf("second FetchTask error = %v", err) } } func TestFacadeDeclareReturnsOfficialRunnerMetadata(t *testing.T) { facade, assignment, token := testFacade(t) ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID) response, err := facade.Declare(ctx, authenticatedRequest(&runnerv1.DeclareRequest{ Version: "v3.5.0", Labels: []string{"self-hosted", "pod"}, }, assignment.ID, token)) if err != nil { t.Fatal(err) } runner := response.Msg.GetRunner() if runner.GetUuid() != assignment.ID || runner.GetName() != assignment.ID || runner.GetVersion() != "v3.5.0" || !runner.GetEphemeral() { t.Fatalf("runner = %#v", runner) } if len(runner.GetLabels()) != 2 || runner.GetLabels()[0] != "self-hosted" || runner.GetLabels()[1] != "pod" { t.Fatalf("labels = %#v", runner.GetLabels()) } } func TestAPIHandlerMatchesOfficialRunnerBasePath(t *testing.T) { facade, assignment, token := testFacade(t) identityURL, err := url.Parse(assignment.Identity.SPIFFEID) if err != nil { t.Fatal(err) } handler := APIHandler(facade) server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { request.TLS = &tls.ConnectionState{PeerCertificates: []*x509.Certificate{{URIs: []*url.URL{identityURL}}}} handler.ServeHTTP(response, request) })) defer server.Close() client := runnerv1connect.NewRunnerServiceClient(server.Client(), server.URL+APIBasePath) request := authenticatedRequest(&runnerv1.DeclareRequest{ Version: "v3.5.0", Labels: []string{"self-hosted", "pod"}, }, assignment.ID, token) response, err := client.Declare(context.Background(), request) if err != nil { t.Fatal(err) } if response.Msg.GetRunner().GetUuid() != assignment.ID { t.Fatalf("runner = %#v", response.Msg.GetRunner()) } } func TestHandlerProxiesRepositoryTrafficToGitea(t *testing.T) { var upstream *httptest.Server upstream = httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { upstreamHost := strings.TrimPrefix(upstream.URL, "http://") if request.URL.Path != "/owner/repo/info/refs" || request.Header.Get("Authorization") != "Basic checkout-token" || request.Host != upstreamHost { t.Errorf("request path=%q authorization=%q host=%q", request.URL.Path, request.Header.Get("Authorization"), request.Host) response.WriteHeader(http.StatusBadRequest) return } response.WriteHeader(http.StatusOK) })) defer upstream.Close() facade, _, _ := testFacade(t) handler, err := Handler(facade, upstream.URL) if err != nil { t.Fatal(err) } request := httptest.NewRequest(http.MethodGet, "http://facade/owner/repo/info/refs", nil) request.Header.Set("Authorization", "Basic checkout-token") response := httptest.NewRecorder() handler.ServeHTTP(response, request) if response.Code != http.StatusOK { t.Fatalf("status = %d", response.Code) } } func TestFacadeRejectsWrongIdentityOrCapability(t *testing.T) { facade, assignment, token := testFacade(t) wrongIdentity := WithSPIFFEID(context.Background(), "spiffe://ddupan.top/ci/owner/repo/other") if _, err := facade.FetchTask(wrongIdentity, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); connect.CodeOf(err) != connect.CodeFailedPrecondition { t.Fatalf("wrong identity error = %v", err) } ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID) if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, "wrong")); connect.CodeOf(err) != connect.CodeUnauthenticated { t.Fatalf("wrong capability error = %v", err) } } func TestFacadeForwardsOnlyMatchingTaskAndLogUpdates(t *testing.T) { facade, assignment, token := testFacade(t) ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID) if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); err != nil { t.Fatal(err) } if _, err := facade.UpdateTask(ctx, authenticatedRequest(&runnerv1.UpdateTaskRequest{State: &runnerv1.TaskState{Id: 42}}, assignment.ID, token)); err != nil { t.Fatal(err) } if _, err := facade.UpdateLog(ctx, authenticatedRequest(&runnerv1.UpdateLogRequest{TaskId: 42}, assignment.ID, token)); err != nil { t.Fatal(err) } upstream := facade.Upstream.(*fakeUpstream) if upstream.taskUpdates != 1 || upstream.logUpdates != 1 { t.Fatalf("task updates=%d log updates=%d", upstream.taskUpdates, upstream.logUpdates) } if _, err := facade.UpdateLog(ctx, authenticatedRequest(&runnerv1.UpdateLogRequest{TaskId: 99}, assignment.ID, token)); connect.CodeOf(err) != connect.CodePermissionDenied { t.Fatalf("mismatched log error = %v", err) } } func TestFacadeSignalsTerminalTaskAfterUpstreamAcceptsIt(t *testing.T) { facade, assignment, token := testFacade(t) ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID) if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); err != nil { t.Fatal(err) } completed := 0 facade.OnTerminal = func(_ context.Context, got taskassignment.Assignment) error { if got.ID != assignment.ID { t.Fatalf("terminal assignment = %s", got.ID) } completed++ return nil } if _, err := facade.UpdateTask(ctx, authenticatedRequest(&runnerv1.UpdateTaskRequest{ State: &runnerv1.TaskState{Id: 42, Result: runnerv1.Result_RESULT_SUCCESS}, }, assignment.ID, token)); err != nil { t.Fatal(err) } if completed != 1 { t.Fatalf("terminal notifications = %d", completed) } } func TestCapabilitiesAreDeterministicAndAssignmentScoped(t *testing.T) { capabilities, err := NewCapabilities([]byte("0123456789abcdef0123456789abcdef")) if err != nil { t.Fatal(err) } token := capabilities.Issue("gitea-task-42") if !capabilities.Verify("gitea-task-42", token) || capabilities.Verify("gitea-task-43", token) { t.Fatal("capability scope is invalid") } } func TestRegistryRecoversClaimedAssignment(t *testing.T) { registry := NewRegistry() assignment := facadeAssignment(t) if err := registry.RecoverClaimed(assignment); err != nil { t.Fatal(err) } resolved, err := registry.Resolve(assignment.ID, assignment.Identity.SPIFFEID) if err != nil || resolved.Task.GetId() != assignment.Task.GetId() { t.Fatalf("resolved=%#v err=%v", resolved, err) } }