<commit_before>package queue_test

import (
	"context"
	"cred-alert/metrics/metricsfakes"
	"cred-alert/pubsubrunner"
	"cred-alert/queue"
	"cred-alert/queue/queuefakes"
	"os"

	"cloud.google.com/go/pubsub"

	"code.cloudfoundry.org/lager"
	"code.cloudfoundry.org/lager/lagertest"
	"github.com/tedsuo/ifrit"
	"github.com/tedsuo/ifrit/ginkgomon"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
)

var _ = Describe("PubSubSubscriber", func() {
	var (
		logger        *lagertest.TestLogger
		firstMessage  *pubsub.Message
		secondMessage *pubsub.Message
		processor     *queuefakes.FakePubSubProcessor
		subscription  *pubsub.Subscription
		topic         *pubsub.Topic
		client        *pubsub.Client
		emitter       *metricsfakes.FakeEmitter

		psRunner *pubsubrunner.Runner
		runner   ifrit.Runner
		process  ifrit.Process
	)

	BeforeEach(func() {
		psRunner = &pubsubrunner.Runner{}
		psRunner.Setup()

		logger = lagertest.NewTestLogger("processor")

		firstMessage = &pubsub.Message{
			Attributes: map[string]string{
				"id": "some-id",
			},
		}

		secondMessage = &pubsub.Message{
			Attributes: map[string]string{
				"id": "some-other-id",
			},
		}

		var err error
		ctx := context.Background()
		client, err = pubsub.NewClient(ctx, "a-project-id")
		Expect(err).NotTo(HaveOccurred())

		topic, err = client.CreateTopic(ctx, "a-topic-id")
		Expect(err).NotTo(HaveOccurred())

		subscription, err = client.CreateSubscription(ctx, "a-subscription-id", topic, 0, nil)
		Expect(err).NotTo(HaveOccurred())

		res := topic.Publish(ctx, firstMessage)
		_, err = res.Get(ctx)
		Expect(err).NotTo(HaveOccurred())

		res = topic.Publish(ctx, secondMessage)
		_, err = res.Get(ctx)
		Expect(err).NotTo(HaveOccurred())

		processor = &queuefakes.FakePubSubProcessor{}

		emitter = &metricsfakes.FakeEmitter{}
		emitter.CounterReturns(&metricsfakes.FakeCounter{})

		fakeTimer := &metricsfakes.FakeTimer{}
		fakeTimer.TimeStub = func(logger lager.Logger, f func(), tags ...string) {
			f()
		}
		emitter.TimerReturns(fakeTimer)
	})

	AfterEach(func() {
		ginkgomon.Interrupt(process)
		client.Close()
		psRunner.Teardown()
	})

	JustBeforeEach(func() {
		runner = queue.NewPubSubSubscriber(logger, subscription, processor, emitter)
		process = ginkgomon.Invoke(runner)
	})

	Context("when the runner is signaled", func() {
		It("exits gracefully", func() {
			process.Signal(os.Interrupt)
			Eventually(process.Wait()).Should(Receive())
		})

		It("does not process any more messages", func() {
			Eventually(processor.ProcessCallCount).Should(Equal(2))
			process.Signal(os.Interrupt)
			Eventually(process.Wait()).Should(Receive())

			res := topic.Publish(context.Background(), firstMessage)
			_, err := res.Get(context.Background())
			Expect(err).NotTo(HaveOccurred())

			Consistently(processor.ProcessCallCount).Should(Equal(2))
		})
	})

	It("tries to process the messages", func() {
		Eventually(processor.ProcessCallCount).Should(Equal(2))

		_, message1 := processor.ProcessArgsForCall(0)
		_, message2 := processor.ProcessArgsForCall(1)

		attributes := []map[string]string{
			message1.Attributes,
			message2.Attributes,
		}

		Expect(attributes).To(ConsistOf(firstMessage.Attributes, secondMessage.Attributes))
	})
})
<commit_msg>missed one usage<commit_after>package queue_test

import (
	"context"
	"os"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"

	"cloud.google.com/go/pubsub"
	"code.cloudfoundry.org/lager"
	"code.cloudfoundry.org/lager/lagertest"
	"github.com/tedsuo/ifrit"
	"github.com/tedsuo/ifrit/ginkgomon"

	"cred-alert/metrics/metricsfakes"
	"cred-alert/pubsubrunner"
	"cred-alert/queue"
	"cred-alert/queue/queuefakes"
)

var _ = Describe("PubSubSubscriber", func() {
	var (
		logger        *lagertest.TestLogger
		firstMessage  *pubsub.Message
		secondMessage *pubsub.Message
		processor     *queuefakes.FakePubSubProcessor
		subscription  *pubsub.Subscription
		topic         *pubsub.Topic
		client        *pubsub.Client
		emitter       *metricsfakes.FakeEmitter

		psRunner *pubsubrunner.Runner
		runner   ifrit.Runner
		process  ifrit.Process
	)

	BeforeEach(func() {
		psRunner = &pubsubrunner.Runner{}
		psRunner.Setup()

		logger = lagertest.NewTestLogger("processor")

		firstMessage = &pubsub.Message{
			Attributes: map[string]string{
				"id": "some-id",
			},
		}

		secondMessage = &pubsub.Message{
			Attributes: map[string]string{
				"id": "some-other-id",
			},
		}

		var err error
		ctx := context.Background()
		client, err = pubsub.NewClient(ctx, "a-project-id")
		Expect(err).NotTo(HaveOccurred())

		topic, err = client.CreateTopic(ctx, "a-topic-id")
		Expect(err).NotTo(HaveOccurred())

		subscription, err = client.CreateSubscription(ctx, "a-subscription-id", pubsub.SubscriptionConfig{
			Topic: topic,
		})
		Expect(err).NotTo(HaveOccurred())

		res := topic.Publish(ctx, firstMessage)
		_, err = res.Get(ctx)
		Expect(err).NotTo(HaveOccurred())

		res = topic.Publish(ctx, secondMessage)
		_, err = res.Get(ctx)
		Expect(err).NotTo(HaveOccurred())

		processor = &queuefakes.FakePubSubProcessor{}

		emitter = &metricsfakes.FakeEmitter{}
		emitter.CounterReturns(&metricsfakes.FakeCounter{})

		fakeTimer := &metricsfakes.FakeTimer{}
		fakeTimer.TimeStub = func(logger lager.Logger, f func(), tags ...string) {
			f()
		}
		emitter.TimerReturns(fakeTimer)
	})

	AfterEach(func() {
		ginkgomon.Interrupt(process)
		client.Close()
		psRunner.Teardown()
	})

	JustBeforeEach(func() {
		runner = queue.NewPubSubSubscriber(logger, subscription, processor, emitter)
		process = ginkgomon.Invoke(runner)
	})

	Context("when the runner is signaled", func() {
		It("exits gracefully", func() {
			process.Signal(os.Interrupt)
			Eventually(process.Wait()).Should(Receive())
		})

		It("does not process any more messages", func() {
			Eventually(processor.ProcessCallCount).Should(Equal(2))
			process.Signal(os.Interrupt)
			Eventually(process.Wait()).Should(Receive())

			res := topic.Publish(context.Background(), firstMessage)
			_, err := res.Get(context.Background())
			Expect(err).NotTo(HaveOccurred())

			Consistently(processor.ProcessCallCount).Should(Equal(2))
		})
	})

	It("tries to process the messages", func() {
		Eventually(processor.ProcessCallCount).Should(Equal(2))

		_, message1 := processor.ProcessArgsForCall(0)
		_, message2 := processor.ProcessArgsForCall(1)

		attributes := []map[string]string{
			message1.Attributes,
			message2.Attributes,
		}

		Expect(attributes).To(ConsistOf(firstMessage.Attributes, secondMessage.Attributes))
	})
})

{'fraction_non_alphanumeric': 0.1446502372570029, 'fraction_numerical': 0.0029083116485534976, 'mean_word_length': 4.426910299003322, 'pattern_counts': {'":': 4, '<': 3, '<?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': '927046', 'n_tokens_mistral': 2467, 'n_tokens_neox': 2209, 'n_words': 338}