<commit_before>package rabbithole

import (
	"encoding/json"
	"fmt"
	"net/url"
	"strings"
	"time"

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

// TODO: extract duplication between these
func FindQueueByName(sl []QueueInfo, name string) (q QueueInfo) {
	for _, i := range sl {
		if name == i.Name {
			q = i
			break
		}
	}
	return q
}

func FindUserByName(sl []UserInfo, name string) (u UserInfo) {
	for _, i := range sl {
		if name == i.Name {
			u = i
			break
		}
	}
	return u
}

func openConnection(vhost string) *amqp.Connection {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/" + url.QueryEscape(vhost))
	Ω(err).Should(BeNil())

	if err != nil {
		panic("failed to connect")
	}

	return conn
}

func ensureNonZeroMessageRate(ch *amqp.Channel) {
	for i := 0; i < 2000; i++ {
		q, _ := ch.QueueDeclare(
			"",    // name
			false, // durable
			false, // auto delete
			true,  // exclusive
			false,
			nil)
		ch.Publish("", q.Name, false, false, amqp.Publishing{Body: []byte("")})
	}
}

// Wait for the list of connections to reach the expected length
func listConnectionsUntil(c *Client, i int) {
	xs, _ := c.ListConnections()
	// Avoid infinity loops by breaking it after 30s
	breakLoop := 0
	for i != len(xs) {
		if breakLoop == 300 {
			fmt.Printf("Stopping listConnectionsUntil loop: expected %v obtained %v", i, len(xs))
			break
		}
		breakLoop += 1
		// Wait between calls
		time.Sleep(100 * time.Millisecond)
		xs, _ = c.ListConnections()
	}
}

func awaitEventPropagation() {
	time.Sleep(1150 * time.Millisecond)
}

type portTestStruct struct {
	Port Port `json:"port"`
}

var _ = Describe("Rabbithole", func() {
	var (
		rmqc *Client
	)

	BeforeEach(func() {
		rmqc, _ = NewClient("http://127.0.0.1:15672", "guest", "guest")
	})

	Context("PUT /parameters/shovel/{vhost}/{name}", func() {
		It("declares a shovel", func() {
			vh := "rabbit/hole"
			sn := "temporary"

			shovelDefinition := ShovelDefinition{
				SourceURI:         "amqp://sourceURI",
				SourceQueue:       "mySourceQueue",
				DestinationURI:    "amqp://destinationURI",
				DestinationQueue:  "myDestQueue",
				AddForwardHeaders: true,
				AckMode:           "on-confirm",
				DeleteAfter:       "never"}

			_, err := rmqc.DeclareShovel(vh, sn, shovelDefinition)
			Ω(err).Should(BeNil())

			awaitEventPropagation()
			x, err := rmqc.GetShovel(vh, sn)
			Ω(err).Should(BeNil())
			Ω(x.Name).Should(Equal(sn))
			Ω(x.Vhost).Should(Equal(vh))
			Ω(x.Component).Should(Equal("shovel"))
			Ω(x.Definition.SourceURI).Should(Equal("amqp://sourceURI"))
			Ω(x.Definition.SourceQueue).Should(Equal("mySourceQueue"))
			Ω(x.Definition.DestinationURI).Should(Equal("amqp://destinationURI"))
			Ω(x.Definition.DestinationQueue).Should(Equal("myDestQueue"))
			Ω(x.Definition.AddForwardHeaders).Should(Equal(true))
			Ω(x.Definition.AckMode).Should(Equal("on-confirm"))
			Ω(x.Definition.DeleteAfter).Should(Equal("never"))

			rmqc.DeleteShovel(vh, sn)
		})
	})

	Context("GET /overview", func() {
		It("returns decoded response", func() {
			conn := openConnection("/")
			defer conn.Close()

			ch, err := conn.Channel()
			Ω(err).Should(BeNil())

			ensureNonZeroMessageRate(ch)

			res, err := rmqc.Overview()
			Ω(err).Should(BeNil())

			Ω(res.Node).ShouldNot(BeNil())
			Ω(res.StatisticsDBNode).ShouldNot(BeNil())<commit_msg>Use reachable endpoints for this shovel<commit_after>package rabbithole

import (
	"encoding/json"
	"fmt"
	"net/url"
	"strings"
	"time"

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

// TODO: extract duplication between these
func FindQueueByName(sl []QueueInfo, name string) (q QueueInfo) {
	for _, i := range sl {
		if name == i.Name {
			q = i
			break
		}
	}
	return q
}

func FindUserByName(sl []UserInfo, name string) (u UserInfo) {
	for _, i := range sl {
		if name == i.Name {
			u = i
			break
		}
	}
	return u
}

func openConnection(vhost string) *amqp.Connection {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/" + url.QueryEscape(vhost))
	Ω(err).Should(BeNil())

	if err != nil {
		panic("failed to connect")
	}

	return conn
}

func ensureNonZeroMessageRate(ch *amqp.Channel) {
	for i := 0; i < 2000; i++ {
		q, _ := ch.QueueDeclare(
			"",    // name
			false, // durable
			false, // auto delete
			true,  // exclusive
			false,
			nil)
		ch.Publish("", q.Name, false, false, amqp.Publishing{Body: []byte("")})
	}
}

// Wait for the list of connections to reach the expected length
func listConnectionsUntil(c *Client, i int) {
	xs, _ := c.ListConnections()
	// Avoid infinity loops by breaking it after 30s
	breakLoop := 0
	for i != len(xs) {
		if breakLoop == 300 {
			fmt.Printf("Stopping listConnectionsUntil loop: expected %v obtained %v", i, len(xs))
			break
		}
		breakLoop += 1
		// Wait between calls
		time.Sleep(100 * time.Millisecond)
		xs, _ = c.ListConnections()
	}
}

func awaitEventPropagation() {
	time.Sleep(1150 * time.Millisecond)
}

type portTestStruct struct {
	Port Port `json:"port"`
}

var _ = Describe("Rabbithole", func() {
	var (
		rmqc *Client
	)

	BeforeEach(func() {
		rmqc, _ = NewClient("http://127.0.0.1:15672", "guest", "guest")
	})

	Context("PUT /parameters/shovel/{vhost}/{name}", func() {
		It("declares a shovel", func() {
			vh := "rabbit/hole"
			sn := "temporary"

			ssu := "amqp://127.0.0.1/%2f"
			sdu := "amqp://127.0.0.1/%2f"

			shovelDefinition := ShovelDefinition{
				SourceURI:         ssu,
				SourceQueue:       "mySourceQueue",
				DestinationURI:    sdu,
				DestinationQueue:  "myDestQueue",
				AddForwardHeaders: true,
				AckMode:           "on-confirm",
				DeleteAfter:       "never"}

			_, err := rmqc.DeclareShovel(vh, sn, shovelDefinition)
			Ω(err).Should(BeNil())

			awaitEventPropagation()
			x, err := rmqc.GetShovel(vh, sn)
			Ω(err).Should(BeNil())
			Ω(x.Name).Should(Equal(sn))
			Ω(x.Vhost).Should(Equal(vh))
			Ω(x.Component).Should(Equal("shovel"))
			Ω(x.Definition.SourceURI).Should(Equal(ssu))
			Ω(x.Definition.SourceQueue).Should(Equal("mySourceQueue"))
			Ω(x.Definition.DestinationURI).Should(Equal(sdu))
			Ω(x.Definition.DestinationQueue).Should(Equal("myDestQueue"))
			Ω(x.Definition.AddForwardHeaders).Should(Equal(true))
			Ω(x.Definition.AckMode).Should(Equal("on-confirm"))
			Ω(x.Definition.DeleteAfter).Should(Equal("never"))

			rmqc.DeleteShovel(vh, sn)
			awaitEventPropagation()
			x, err = rmqc.GetShovel(vh, sn)
			Ω(x).Should(BeNil())
		})
	})

	Context("GET /overview", func() {
		It("returns decoded response", func() {
			conn := openConnection("/")
			defer conn.Close()

			ch, err := conn.Channel()
			Ω(err).Should(BeNil())

			ensureNonZeroMessageRate(ch)

			res, err := rmqc.Overview()
			Ω(err).Should(BeNil())

			Ω(res.Node).ShouldNot(BeNil())
			Ω(res.StatisticsDBNode).ShouldNot(BeNil())
{'fraction_non_alphanumeric': 0.16703200350723366, 'fraction_numerical': 0.011983048370597691, 'mean_word_length': 3.571810287241149, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, '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': '873199', 'n_tokens_mistral': 2869, 'n_tokens_neox': 2543, 'n_words': 502}