<commit_before>package integration_test

import (
	"encoding/json"
	"fmt"
	"net/http"
	"os/exec"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/onsi/gomega/gbytes"
	"github.com/onsi/gomega/gexec"
	"github.com/onsi/gomega/ghttp"
	"github.com/vito/go-sse/sse"

	"github.com/concourse/atc"
	"github.com/concourse/atc/event"
)

var _ = Describe("Watching", func() {
	var atcServer *ghttp.Server
	var streaming chan struct{}
	var events chan atc.Event

	BeforeEach(func() {
		atcServer = ghttp.NewServer()
		streaming = make(chan struct{})
		events = make(chan atc.Event)
	})

	eventsHandler := func() http.HandlerFunc {
		return ghttp.CombineHandlers(
			ghttp.VerifyRequest("GET", "/api/v1/builds/3/events"),
			func(w http.ResponseWriter, r *http.Request) {
				flusher := w.(http.Flusher)

				w.Header().Add("Content-Type", "text/event-stream; charset=utf-8")
				w.Header().Add("Cache-Control", "no-cache, no-store, must-revalidate")
				w.Header().Add("Connection", "keep-alive")

				w.WriteHeader(http.StatusOK)

				flusher.Flush()

				close(streaming)

				id := 0

				for e := range events {
					payload, err := json.Marshal(event.Message{e})
					Expect(err).NotTo(HaveOccurred())

					event := sse.Event{
						ID:   fmt.Sprintf("%d", id),
						Name: "event",
						Data: payload,
					}

					err = event.Write(w)
					Expect(err).NotTo(HaveOccurred())

					flusher.Flush()

					id++
				}

				err := sse.Event{
					Name: "end",
				}.Write(w)
				Expect(err).NotTo(HaveOccurred())
			},
		)
	}

	watch := func(args ...string) {
		watchWithArgs := append([]string{"watch"}, args...)

		flyCmd := exec.Command(flyPath, append([]string{"-t", atcServer.URL()}, watchWithArgs...)...)

		sess, err := gexec.Start(flyCmd, GinkgoWriter, GinkgoWriter)
		Expect(err).NotTo(HaveOccurred())

		Eventually(streaming).Should(BeClosed())

		events <- event.Log{Payload: "sup"}

		Eventually(sess.Out).Should(gbytes.Say("sup"))

		close(events)

		<-sess.Exited
		Expect(sess.ExitCode()).To(Equal(0))
	}

	Context("with no arguments", func() {
		BeforeEach(func() {
			atcServer.AppendHandlers(
				ghttp.CombineHandlers(
					ghttp.VerifyRequest("GET", "/api/v1/builds"),
					ghttp.RespondWithJSONEncoded(200, []atc.Build{
						{ID: 4, Name: "1", Status: "started", JobName: "some-job"},
						{ID: 3, Name: "3", Status: "started"},
						{ID: 2, Name: "2", Status: "started"},
						{ID: 1, Name: "1", Status: "finished"},
					}),
				),
				eventsHandler(),
			)
		})

		It("watches the most recent one-off build", func() {
			watch()
		})
	})

	Context("with a specific job and pipeline", func() {
		Context("when the job has no builds", func() {
			BeforeEach(func() {
				didStream := make(chan struct{})
				streaming = didStream

				atcServer.AppendHandlers(
					ghttp.CombineHandlers(
						ghttp.VerifyRequest("GET", "/api/v1/pipelines/some-pipeline/jobs/some-job"),
						ghttp.RespondWithJSONEncoded(200, atc.Job{}),
					),
					eventsHandler(),
				)
			})

			It("returns an error and exits", func() {
				flyCmd := exec.Command(flyPath, "-t", atcServer.URL(), "watch", "--job", "some-pipeline/some-job")
				sess, err := gexec.Start(flyCmd, GinkgoWriter, GinkgoWriter)
				Expect(err).NotTo(HaveOccurred())

				Eventually(sess.Err).Should(gbytes.Say("job has no builds"))
				Expect(sess.ExitCode()).To(Equal(1))
			})
		})

		Context("when the job has a next build", func() {
			BeforeEach(func() {
				didStream := make(chan struct{})
				streaming = didStream

				atcServer.AppendHandlers(
					ghttp.CombineHandlers(
						ghttp.VerifyRequest("GET", "/api/v1/pipelines/some-pipeline/jobs/some-job"),
						ghttp.RespondWithJSONEncoded(200, atc.Job{
							NextBuild: &atc.Build{
								ID:      3,
								Name:    "3",
								Status:  "started",
								JobName: "some-job",
							},
							FinishedBuild: &atc.Build{
								ID:      2,
								Name:    "2",<commit_msg>fix race where command had not exited yet<commit_after>package integration_test

import (
	"encoding/json"
	"fmt"
	"net/http"
	"os/exec"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/onsi/gomega/gbytes"
	"github.com/onsi/gomega/gexec"
	"github.com/onsi/gomega/ghttp"
	"github.com/vito/go-sse/sse"

	"github.com/concourse/atc"
	"github.com/concourse/atc/event"
)

var _ = Describe("Watching", func() {
	var atcServer *ghttp.Server
	var streaming chan struct{}
	var events chan atc.Event

	BeforeEach(func() {
		atcServer = ghttp.NewServer()
		streaming = make(chan struct{})
		events = make(chan atc.Event)
	})

	eventsHandler := func() http.HandlerFunc {
		return ghttp.CombineHandlers(
			ghttp.VerifyRequest("GET", "/api/v1/builds/3/events"),
			func(w http.ResponseWriter, r *http.Request) {
				flusher := w.(http.Flusher)

				w.Header().Add("Content-Type", "text/event-stream; charset=utf-8")
				w.Header().Add("Cache-Control", "no-cache, no-store, must-revalidate")
				w.Header().Add("Connection", "keep-alive")

				w.WriteHeader(http.StatusOK)

				flusher.Flush()

				close(streaming)

				id := 0

				for e := range events {
					payload, err := json.Marshal(event.Message{e})
					Expect(err).NotTo(HaveOccurred())

					event := sse.Event{
						ID:   fmt.Sprintf("%d", id),
						Name: "event",
						Data: payload,
					}

					err = event.Write(w)
					Expect(err).NotTo(HaveOccurred())

					flusher.Flush()

					id++
				}

				err := sse.Event{
					Name: "end",
				}.Write(w)
				Expect(err).NotTo(HaveOccurred())
			},
		)
	}

	watch := func(args ...string) {
		watchWithArgs := append([]string{"watch"}, args...)

		flyCmd := exec.Command(flyPath, append([]string{"-t", atcServer.URL()}, watchWithArgs...)...)

		sess, err := gexec.Start(flyCmd, GinkgoWriter, GinkgoWriter)
		Expect(err).NotTo(HaveOccurred())

		Eventually(streaming).Should(BeClosed())

		events <- event.Log{Payload: "sup"}

		Eventually(sess.Out).Should(gbytes.Say("sup"))

		close(events)

		<-sess.Exited
		Expect(sess.ExitCode()).To(Equal(0))
	}

	Context("with no arguments", func() {
		BeforeEach(func() {
			atcServer.AppendHandlers(
				ghttp.CombineHandlers(
					ghttp.VerifyRequest("GET", "/api/v1/builds"),
					ghttp.RespondWithJSONEncoded(200, []atc.Build{
						{ID: 4, Name: "1", Status: "started", JobName: "some-job"},
						{ID: 3, Name: "3", Status: "started"},
						{ID: 2, Name: "2", Status: "started"},
						{ID: 1, Name: "1", Status: "finished"},
					}),
				),
				eventsHandler(),
			)
		})

		It("watches the most recent one-off build", func() {
			watch()
		})
	})

	Context("with a specific job and pipeline", func() {
		Context("when the job has no builds", func() {
			BeforeEach(func() {
				didStream := make(chan struct{})
				streaming = didStream

				atcServer.AppendHandlers(
					ghttp.CombineHandlers(
						ghttp.VerifyRequest("GET", "/api/v1/pipelines/some-pipeline/jobs/some-job"),
						ghttp.RespondWithJSONEncoded(200, atc.Job{}),
					),
					eventsHandler(),
				)
			})

			It("returns an error and exits", func() {
				flyCmd := exec.Command(flyPath, "-t", atcServer.URL(), "watch", "--job", "some-pipeline/some-job")
				sess, err := gexec.Start(flyCmd, GinkgoWriter, GinkgoWriter)
				Expect(err).NotTo(HaveOccurred())

				Eventually(sess.Err).Should(gbytes.Say("job has no builds"))
				<-sess.Exited

				Expect(sess.ExitCode()).To(Equal(1))
			})
		})

		Context("when the job has a next build", func() {
			BeforeEach(func() {
				didStream := make(chan struct{})
				streaming = didStream

				atcServer.AppendHandlers(
					ghttp.CombineHandlers(
						ghttp.VerifyRequest("GET", "/api/v1/pipelines/some-pipeline/jobs/some-job"),
						ghttp.RespondWithJSONEncoded(200, atc.Job{
							NextBuild: &atc.Build{
								ID:      3,
								Name:    "3",
								Status:  "started",
								JobName: "some-job",
							},
							FinishedBuild: &atc.Build{
								ID:      2,
								Name:    "2",
{'fraction_non_alphanumeric': 0.17621758380771663, 'fraction_numerical': 0.007590132827324478, 'mean_word_length': 3.113423517169615, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7216256', 'n_tokens_mistral': 3572, 'n_tokens_neox': 2835, 'n_words': 509}