<commit_before>package server

import (
	"bytes"
	"fmt"
	"github.com/heroku/busl/broker"
	"github.com/heroku/busl/util"
	. "gopkg.in/check.v1"
	"io"
	"io/ioutil"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"
)

func Test(t *testing.T) { TestingT(t) }

type HttpServerSuite struct{}

var _ = Suite(&HttpServerSuite{})
var sf = fmt.Sprintf

func newRequest(method, url, body string) *http.Request {
	return newRequestFromReader(method, url, bytes.NewBufferString(body))
}

func newRequestFromReader(method, url string, reader io.Reader) *http.Request {
	request, _ := http.NewRequest(method, url, reader)
	urlParts := strings.Split(url, "/")
	if method == "POST" {
		request.TransferEncoding = []string{"chunked"}
		request.Header.Add("Transfer-Encoding", "chunked")
	}
	if len(urlParts) == 3 {
		streamId := urlParts[2]
		setStreamId(request, streamId)
	}
	return request
}

func setStreamId(req *http.Request, streamId string) {
	req.URL.RawQuery = "%3Auuid=" + streamId + "&"
}

func (s *HttpServerSuite) TestMkstream(c *C) {
	request := newRequest("POST", "/streams", "")
	response := httptest.NewRecorder()

	mkstream(response, request)

	c.Assert(response.Code, Equals, 200)
	c.Assert(response.Body.String(), HasLen, 32)
}

func (s *HttpServerSuite) Test410(c *C) {
	streamId, _ := util.NewUUID()
	request := newRequest("GET", "/streams/"+string(streamId), "")
	response := CloseNotifierRecorder{httptest.NewRecorder(), make(chan bool, 1)}

	sub(response, request)

	c.Assert(response.Code, Equals, http.StatusNotFound)
	c.Assert(response.Body.String(), Equals, "Channel is not registered.\n")
}

func (s *HttpServerSuite) TestPubNotRegistered(c *C) {
	streamId, _ := util.NewUUID()
	request := newRequest("POST", "/streams/"+string(streamId), "")
	response := httptest.NewRecorder()

	pub(response, request)

	c.Assert(response.Code, Equals, http.StatusNotFound)
}

func (s *HttpServerSuite) TestPubWithoutTransferEncoding(c *C) {
	request, _ := http.NewRequest("POST", "/streams/1234", nil)
	setStreamId(request, "1234")
	response := httptest.NewRecorder()

	pub(response, request)

	c.Assert(response.Code, Equals, http.StatusBadRequest)
	c.Assert(response.Body.String(), Equals, "A chunked Transfer-Encoding header is required.\n")
}

func (s *HttpServerSuite) TestSub(c *C) {
	streamId, _ := util.NewUUID()
	registrar := broker.NewRedisRegistrar()
	registrar.Register(streamId)
	writer, _ := broker.NewWriter(streamId)

	request := newRequest("GET", sf("/streams/%s", streamId), "")
	response := CloseNotifierRecorder{httptest.NewRecorder(), make(chan bool, 1)}

	waiter := util.TimeoutFunc(time.Millisecond*5, func() {
		sub(response, request)
	})

	writer.Write([]byte("busl1"))
	writer.Close()
	<-waiter

	c.Assert(response.Code, Equals, http.StatusOK)
	c.Assert(response.Body.String(), Equals, "busl1")
}

func (s *HttpServerSuite) TestPubSub(c *C) {
	streamId, _ := util.NewUUID()
	registrar := broker.NewRedisRegistrar()
	registrar.Register(streamId)

	body := new(bytes.Buffer)
	bodyCloser := ioutil.NopCloser(body)

	pubRequest := newRequestFromReader("POST", sf("/streams/%s", streamId), bodyCloser)
	pubResponse := CloseNotifierRecorder{httptest.NewRecorder(), make(chan bool, 1)}

	pubBlocker := util.TimeoutFunc(time.Millisecond*5, func() {
		pub(pubResponse, pubRequest)
	})

	subRequest := newRequest("GET", sf("/streams/%s", streamId), "")
	subResponse := CloseNotifierRecorder{httptest.NewRecorder(), make(chan bool, 1)}

	subBlocker := util.TimeoutFunc(time.Millisecond*5, func() {
		sub(subResponse, subRequest)
	})

	for _, m := range []string{"first", " ", "second", " ", "third"} {
		body.Write([]byte(m))
	}

	bodyCloser.Close()
	<-pubBlocker
	<-subBlocker

	c.Assert(subResponse.Code, Equals, http.StatusOK)
	c.Assert(subResponse.Body.String(), Equals, "first second third")
}

func (s *HttpServerSuite) TestBinaryPubSub(c *C) {
	streamId, _ := util.NewUUID()
	registrar := broker.NewRedisRegistrar()
	registrar.Register(streamId)

	body := new(bytes.Buffer)
	bodyCloser := ioutil.NopCloser(body)

	pubRequest := newRequestFromReader("POST", sf("/streams/%s", streamId), bodyCloser)
	pubResponse := CloseNotifierRecorder{httptest.NewRecorder(), make(chan bool, 1)}

	pubBlocker := util.TimeoutFunc(time.Millisecond*5, func() {
		pub(pubResponse, pubRequest)
	})

	subRequest := newRequest("GET", sf("/streams/%s", streamId), "")
	subResponse := CloseNotifierRecorder{httptest.NewRecorder(), make(chan bool, 1)}

	subBlocker := util.TimeoutFunc(time.Millisecond*5, func() {
		sub(subResponse, subRequest)
	})

	expected := []byte{0x1f, 0x8b, 0x08, 0x00, 0x3f, 0x6b, 0xe1, 0x53, 0x00, 0x03, 0xed, 0xce, 0xb1, 0x0a, 0xc2, 0x30}
	for _, m := range expected {
		body.Write([]byte{m})
	}

	bodyCloser.Close()
	<-pubBlocker
	<-subBlocker

	c.Assert(subResponse.Code, Equals, http.StatusOK)
	c.Assert(subResponse.Body.Bytes(), DeepEquals, expected)
}

func (s *HttpServerSuite) TestSubWaitingPub(c *C) {
	// Start the server in a randomly assigned port
	server := httptest.NewServer(app())
	defer server.Close()

	// uuid = curl -XPOST <url>/streams
	resp, err := http.Post(server.URL+"/streams", "", nil)
	defer resp.Body.Close()
	c.Assert(err, Equals, nil)

	body, err := ioutil.ReadAll(resp.Body)
	c.Assert(err, Equals, nil)

	// uuid extracted
	uuid := string(body)
	c.Assert(len(uuid), Equals, 32)

	done := make(chan bool)

	go func() {
		// curl <url>/streams/<uuid>
		// -- waiting for publish to arrive
		resp, err = http.Get(server.URL + "/streams/" + uuid)<commit_msg>Make the test less flappy<commit_after>package server

import (
	"bytes"
	"fmt"
	"github.com/heroku/busl/broker"
	"github.com/heroku/busl/util"
	. "gopkg.in/check.v1"
	"io"
	"io/ioutil"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"
)

func Test(t *testing.T) { TestingT(t) }

type HttpServerSuite struct{}

var _ = Suite(&HttpServerSuite{})
var sf = fmt.Sprintf

func newRequest(method, url, body string) *http.Request {
	return newRequestFromReader(method, url, bytes.NewBufferString(body))
}

func newRequestFromReader(method, url string, reader io.Reader) *http.Request {
	request, _ := http.NewRequest(method, url, reader)
	urlParts := strings.Split(url, "/")
	if method == "POST" {
		request.TransferEncoding = []string{"chunked"}
		request.Header.Add("Transfer-Encoding", "chunked")
	}
	if len(urlParts) == 3 {
		streamId := urlParts[2]
		setStreamId(request, streamId)
	}
	return request
}

func setStreamId(req *http.Request, streamId string) {
	req.URL.RawQuery = "%3Auuid=" + streamId + "&"
}

func (s *HttpServerSuite) TestMkstream(c *C) {
	request := newRequest("POST", "/streams", "")
	response := httptest.NewRecorder()

	mkstream(response, request)

	c.Assert(response.Code, Equals, 200)
	c.Assert(response.Body.String(), HasLen, 32)
}

func (s *HttpServerSuite) Test410(c *C) {
	streamId, _ := util.NewUUID()
	request := newRequest("GET", "/streams/"+string(streamId), "")
	response := CloseNotifierRecorder{httptest.NewRecorder(), make(chan bool, 1)}

	sub(response, request)

	c.Assert(response.Code, Equals, http.StatusNotFound)
	c.Assert(response.Body.String(), Equals, "Channel is not registered.\n")
}

func (s *HttpServerSuite) TestPubNotRegistered(c *C) {
	streamId, _ := util.NewUUID()
	request := newRequest("POST", "/streams/"+string(streamId), "")
	response := httptest.NewRecorder()

	pub(response, request)

	c.Assert(response.Code, Equals, http.StatusNotFound)
}

func (s *HttpServerSuite) TestPubWithoutTransferEncoding(c *C) {
	request, _ := http.NewRequest("POST", "/streams/1234", nil)
	setStreamId(request, "1234")
	response := httptest.NewRecorder()

	pub(response, request)

	c.Assert(response.Code, Equals, http.StatusBadRequest)
	c.Assert(response.Body.String(), Equals, "A chunked Transfer-Encoding header is required.\n")
}

func (s *HttpServerSuite) TestSub(c *C) {
	streamId, _ := util.NewUUID()
	registrar := broker.NewRedisRegistrar()
	registrar.Register(streamId)
	writer, _ := broker.NewWriter(streamId)

	request := newRequest("GET", sf("/streams/%s", streamId), "")
	response := CloseNotifierRecorder{httptest.NewRecorder(), make(chan bool, 1)}

	waiter := util.TimeoutFunc(time.Millisecond*5, func() {
		sub(response, request)
	})

	writer.Write([]byte("busl1"))
	writer.Close()
	<-waiter

	c.Assert(response.Code, Equals, http.StatusOK)
	c.Assert(response.Body.String(), Equals, "busl1")
}

func (s *HttpServerSuite) TestPubSub(c *C) {
	streamId, _ := util.NewUUID()
	registrar := broker.NewRedisRegistrar()
	registrar.Register(streamId)

	body := new(bytes.Buffer)
	bodyCloser := ioutil.NopCloser(body)

	pubRequest := newRequestFromReader("POST", sf("/streams/%s", streamId), bodyCloser)
	pubResponse := CloseNotifierRecorder{httptest.NewRecorder(), make(chan bool, 1)}

	pubBlocker := util.TimeoutFunc(time.Millisecond*5, func() {
		pub(pubResponse, pubRequest)
	})

	subRequest := newRequest("GET", sf("/streams/%s", streamId), "")
	subResponse := CloseNotifierRecorder{httptest.NewRecorder(), make(chan bool, 1)}

	subBlocker := util.TimeoutFunc(time.Millisecond*5, func() {
		sub(subResponse, subRequest)
	})

	for _, m := range []string{"first", " ", "second", " ", "third"} {
		body.Write([]byte(m))
	}

	bodyCloser.Close()
	<-pubBlocker
	<-subBlocker

	c.Assert(subResponse.Code, Equals, http.StatusOK)
	c.Assert(subResponse.Body.String(), Equals, "first second third")
}

func (s *HttpServerSuite) TestBinaryPubSub(c *C) {
	streamId, _ := util.NewUUID()
	registrar := broker.NewRedisRegistrar()
	registrar.Register(streamId)

	body := new(bytes.Buffer)
	bodyCloser := ioutil.NopCloser(body)

	pubRequest := newRequestFromReader("POST", sf("/streams/%s", streamId), bodyCloser)
	pubResponse := CloseNotifierRecorder{httptest.NewRecorder(), make(chan bool, 1)}

	pubDone := make(chan bool)
	subDone := make(chan bool)

	go func() {
		pub(pubResponse, pubRequest)
		pubDone <- true
	}()

	subRequest := newRequest("GET", sf("/streams/%s", streamId), "")
	subResponse := CloseNotifierRecorder{httptest.NewRecorder(), make(chan bool, 1)}

	go func() {
		sub(subResponse, subRequest)
		subDone <- true
	}()

	expected := []byte{0x1f, 0x8b, 0x08, 0x00, 0x3f, 0x6b, 0xe1, 0x53, 0x00, 0x03, 0xed, 0xce, 0xb1, 0x0a, 0xc2, 0x30}
	for _, m := range expected {
		body.Write([]byte{m})
	}

	bodyCloser.Close()
	<-pubDone
	<-subDone

	c.Assert(subResponse.Code, Equals, http.StatusOK)
	c.Assert(subResponse.Body.Bytes(), DeepEquals, expected)
}

func (s *HttpServerSuite) TestSubWaitingPub(c *C) {
	// Start the server in a randomly assigned port
	server := httptest.NewServer(app())
	defer server.Close()

	// uuid = curl -XPOST <url>/streams
	resp, err := http.Post(server.URL+"/streams", "", nil)
	defer resp.Body.Close()
	c.Assert(err, Equals, nil)

	body, err := ioutil.ReadAll(resp.Body)
	c.Assert(err, Equals, nil)

	// uuid extracted
	uuid := string(body)
	c.Assert(len(uuid), Equals, 32)

	done := make(chan bool)

	go func() {
		// curl <url>/streams/<uuid>
		// -- waiting for publish to arrive
		resp, err = http.Get(server.URL + "/streams/" + uuid)
{'fraction_non_alphanumeric': 0.16803983703033046, 'fraction_numerical': 0.012675418741511997, 'mean_word_length': 5.191704035874439, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 9, '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': '5286292', 'n_tokens_mistral': 4207, 'n_tokens_neox': 4024, 'n_words': 793}