How to use PostAbort method of parallel_support Package

1package interrupt_handler_test
2
3import (
4	"time"
5
6	. "github.com/onsi/ginkgo/v2"
7	"github.com/onsi/ginkgo/v2/internal/interrupt_handler"
8	"github.com/onsi/ginkgo/v2/internal/parallel_support"
9	. "github.com/onsi/ginkgo/v2/internal/test_helpers"
10	. "github.com/onsi/gomega"
11)
12
13var _ = Describe("InterruptHandler", func() {
14	var interruptHandler *interrupt_handler.InterruptHandler
15	Describe("Timeout interrupts", func() {
16		BeforeEach(func() {
17			interruptHandler = interrupt_handler.NewInterruptHandler(500*time.Millisecond, nil)
18			DeferCleanup(interruptHandler.Stop)
19		})
20
21		It("eventually closes the interrupt channel to signal an interrupt has occurred", func() {
22			status := interruptHandler.Status()
23			Ω(status.Interrupted).Should(BeFalse())
24			Eventually(status.Channel).Should(BeClosed())
25
26			Ω(interruptHandler.Status().Interrupted).Should(BeTrue())
27		})
28
29		It("notes the cause as 'Interrupted By Timeout'", func() {
30			status := interruptHandler.Status()
31			Eventually(status.Channel).Should(BeClosed())
32			cause := interruptHandler.Status().Cause
33			Ω(cause).Should(Equal(interrupt_handler.InterruptCauseTimeout))
34			Ω(interruptHandler.InterruptMessageWithStackTraces()).Should(HavePrefix("Interrupted by Timeout\n\n"))
35			Ω(interruptHandler.InterruptMessageWithStackTraces()).Should(ContainSubstring("Here's a stack trace"))
36		})
37
38		It("repeatedly triggers an interrupt every 1/10th of the registered timeout", func() {
39			status := interruptHandler.Status()
40			Ω(status.Interrupted).Should(BeFalse())
41			Eventually(status.Channel).Should(BeClosed())
42
43			status = interruptHandler.Status()
44			Ω(status.Channel).ShouldNot(BeClosed())
45			Eventually(status.Channel).Should(BeClosed())
46		})
47	})
48
49	Describe("Interrupting when another Ginkgo process has aborted", func() {
50		var client parallel_support.Client
51		BeforeEach(func() {
52			_, client, _ = SetUpServerAndClient(2)
53			interruptHandler = interrupt_handler.NewInterruptHandler(0, client)
54			DeferCleanup(interruptHandler.Stop)
55		})
56
57		It("interrupts when the server is told to abort", func() {
58			status := interruptHandler.Status()
59			Consistently(status.Channel).ShouldNot(BeClosed())
60			client.PostAbort()
61			Eventually(status.Channel).Should(BeClosed())
62		})
63
64		It("notes the correct cause and returns an interrupt message that does not include the stacktrace ", func() {
65			status := interruptHandler.Status()
66			client.PostAbort()
67			Eventually(status.Channel).Should(BeClosed())
68			status = interruptHandler.Status()
69			Ω(status.Cause).Should(Equal(interrupt_handler.InterruptCauseAbortByOtherProcess))
70			Ω(interruptHandler.InterruptMessageWithStackTraces()).Should(HavePrefix("Interrupted by Other Ginkgo Process"))
71			Ω(interruptHandler.InterruptMessageWithStackTraces()).ShouldNot(ContainSubstring("Here's a stack trace"))
72		})
73	})
74})
75

Source: interrupt_handler_test.go

package interrupt_handler_test

import (
	"time"

	. "github.com/onsi/ginkgo/v2"
	"github.com/onsi/ginkgo/v2/internal/interrupt_handler"
	"github.com/onsi/ginkgo/v2/internal/parallel_support"
	. "github.com/onsi/ginkgo/v2/internal/test_helpers"
	. "github.com/onsi/gomega"
)

var _ = Describe("InterruptHandler", func() {
	var interruptHandler *interrupt_handler.InterruptHandler
	Describe("Timeout interrupts", func() {
		BeforeEach(func() {
			interruptHandler = interrupt_handler.NewInterruptHandler(500*time.Millisecond, nil)
			DeferCleanup(interruptHandler.Stop)
		})

		It("eventually closes the interrupt channel to signal an interrupt has occurred", func() {
			status := interruptHandler.Status()
			Ω(status.Interrupted).Should(BeFalse())
			Eventually(status.Channel).Should(BeClosed())

			Ω(interruptHandler.Status().Interrupted).Should(BeTrue())
		})

		It("notes the cause as 'Interrupted By Timeout'", func() {
			status := interruptHandler.Status()
			Eventually(status.Channel).Should(BeClosed())
			cause := interruptHandler.Status().Cause
			Ω(cause).Should(Equal(interrupt_handler.InterruptCauseTimeout))
			Ω(interruptHandler.InterruptMessageWithStackTraces()).Should(HavePrefix("Interrupted by Timeout\n\n"))
			Ω(interruptHandler.InterruptMessageWithStackTraces()).Should(ContainSubstring("Here's a stack trace"))
		})

		It("repeatedly triggers an interrupt every 1/10th of the registered timeout", func() {
			status := interruptHandler.Status()
			Ω(status.Interrupted).Should(BeFalse())
			Eventually(status.Channel).Should(BeClosed())

			status = interruptHandler.Status()
			Ω(status.Channel).ShouldNot(BeClosed())
			Eventually(status.Channel).Should(BeClosed())
		})
	})

	Describe("Interrupting when another Ginkgo process has aborted", func() {
		var client parallel_support.Client
		BeforeEach(func() {
			_, client, _ = SetUpServerAndClient(2)
			interruptHandler = interrupt_handler.NewInterruptHandler(0, client)
			DeferCleanup(interruptHandler.Stop)
		})

		It("interrupts when the server is told to abort", func() {
			status := interruptHandler.Status()
			Consistently(status.Channel).ShouldNot(BeClosed())
			client.PostAbort()
			Eventually(status.Channel).Should(BeClosed())
		})

		It("notes the correct cause and returns an interrupt message that does not include the stacktrace ", func() {
			status := interruptHandler.Status()
			client.PostAbort()
			Eventually(status.Channel).Should(BeClosed())
			status = interruptHandler.Status()
			Ω(status.Cause).Should(Equal(interrupt_handler.InterruptCauseAbortByOtherProcess))
			Ω(interruptHandler.InterruptMessageWithStackTraces()).Should(HavePrefix("Interrupted by Other Ginkgo Process"))
			Ω(interruptHandler.InterruptMessageWithStackTraces()).ShouldNot(ContainSubstring("Here's a stack trace"))
		})
	})
})

1package parallel_support
2
3import (
4	"fmt"
5	"io"
6	"os"
7	"time"
8
9	"github.com/onsi/ginkgo/v2/reporters"
10	"github.com/onsi/ginkgo/v2/types"
11)
12
13type BeforeSuiteState struct {
14	Data  []byte
15	State types.SpecState
16}
17
18type ParallelIndexCounter struct {
19	Index int
20}
21
22var ErrorGone = fmt.Errorf("gone")
23var ErrorFailed = fmt.Errorf("failed")
24var ErrorEarly = fmt.Errorf("early")
25
26var POLLING_INTERVAL = 50 * time.Millisecond
27
28type Server interface {
29	Start()
30	Close()
31	Address() string
32	RegisterAlive(node int, alive func() bool)
33	GetSuiteDone() chan interface{}
34	GetOutputDestination() io.Writer
35	SetOutputDestination(io.Writer)
36}
37
38type Client interface {
39	Connect() bool
40	Close() error
41
42	PostSuiteWillBegin(report types.Report) error
43	PostDidRun(report types.SpecReport) error
44	PostSuiteDidEnd(report types.Report) error
45	PostSynchronizedBeforeSuiteCompleted(state types.SpecState, data []byte) error
46	BlockUntilSynchronizedBeforeSuiteData() (types.SpecState, []byte, error)
47	BlockUntilNonprimaryProcsHaveFinished() error
48	BlockUntilAggregatedNonprimaryProcsReport() (types.Report, error)
49	FetchNextCounter() (int, error)
50	PostAbort() error
51	ShouldAbort() bool
52	Write(p []byte) (int, error)
53}
54
55func NewServer(parallelTotal int, reporter reporters.Reporter) (Server, error) {
56	if os.Getenv("GINKGO_PARALLEL_PROTOCOL") == "HTTP" {
57		return newHttpServer(parallelTotal, reporter)
58	} else {
59		return newRPCServer(parallelTotal, reporter)
60	}
61}
62
63func NewClient(serverHost string) Client {
64	if os.Getenv("GINKGO_PARALLEL_PROTOCOL") == "HTTP" {
65		return newHttpClient(serverHost)
66	} else {
67		return newRPCClient(serverHost)
68	}
69}
70

Source: client_server.go

package parallel_support

import (
	"fmt"
	"io"
	"os"
	"time"

	"github.com/onsi/ginkgo/v2/reporters"
	"github.com/onsi/ginkgo/v2/types"
)

type BeforeSuiteState struct {
	Data  []byte
	State types.SpecState
}

type ParallelIndexCounter struct {
	Index int
}

var ErrorGone = fmt.Errorf("gone")
var ErrorFailed = fmt.Errorf("failed")
var ErrorEarly = fmt.Errorf("early")

var POLLING_INTERVAL = 50 * time.Millisecond

type Server interface {
	Start()
	Close()
	Address() string
	RegisterAlive(node int, alive func() bool)
	GetSuiteDone() chan interface{}
	GetOutputDestination() io.Writer
	SetOutputDestination(io.Writer)
}

type Client interface {
	Connect() bool
	Close() error

	PostSuiteWillBegin(report types.Report) error
	PostDidRun(report types.SpecReport) error
	PostSuiteDidEnd(report types.Report) error
	PostSynchronizedBeforeSuiteCompleted(state types.SpecState, data []byte) error
	BlockUntilSynchronizedBeforeSuiteData() (types.SpecState, []byte, error)
	BlockUntilNonprimaryProcsHaveFinished() error
	BlockUntilAggregatedNonprimaryProcsReport() (types.Report, error)
	FetchNextCounter() (int, error)
	PostAbort() error
	ShouldAbort() bool
	Write(p []byte) (int, error)
}

func NewServer(parallelTotal int, reporter reporters.Reporter) (Server, error) {
	if os.Getenv("GINKGO_PARALLEL_PROTOCOL") == "HTTP" {
		return newHttpServer(parallelTotal, reporter)
	} else {
		return newRPCServer(parallelTotal, reporter)
	}
}

func NewClient(serverHost string) Client {
	if os.Getenv("GINKGO_PARALLEL_PROTOCOL") == "HTTP" {
		return newHttpClient(serverHost)
	} else {
		return newRPCClient(serverHost)
	}
}