LogDog: Update client/bootstrap to generate URLs.

logdog/client/butler/butler_test.go

 // Copyright 2015 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package butler
 
 import (
         "bytes"
         "errors"
         "fmt"
@@ skipping 111 matching lines @@
         err  error
 }
 
 type testStream struct {
         inC     chan *testStreamData
         closedC chan struct{}
 
         properties *streamproto.Properties
 }
 
-func newTestStream(p streamproto.Properties) *testStream {
-        return &testStream{
-                inC:        make(chan *testStreamData, 16),
-                closedC:    make(chan struct{}),
-                properties: &p,
-        }
-}
-
 func (ts *testStream) data(d []byte, err error) {
         ts.inC <- &testStreamData{
                 data: d,
                 err:  err,
         }
         if err == io.EOF {
                 // If EOF is hit, continue reporting EOF.
                 close(ts.inC)
         }
 }
@@ skipping 160 matching lines @@
                 })
 
                 Convey(`Will retain the first error and ignore duplicate shutdowns.`, func() {
                         b := mkb(c, conf)
                         b.shutdown(errors.New("first error"))
                         b.shutdown(errors.New("second error"))
                         So(b.Wait(), ShouldErrLike, "first error")
                 })
 
                 Convey(`Using a generic stream Properties`, func() {
-                        props := streamproto.Properties{
-                                LogStreamDescriptor: logpb.LogStreamDescriptor{
-                                        Name:        "test",
-                                        StreamType:  logpb.StreamType_TEXT,
-                                        ContentType: string(types.ContentTypeText),
-                                },
+                        newTestStream := func(setup func(p *streamproto.Properties)) *testStream {
+                                props := streamproto.Properties{
+                                        LogStreamDescriptor: &logpb.LogStreamDescriptor{
+                                                Name:        "test",
+                                                StreamType:  logpb.StreamType_TEXT,
+                                                ContentType: string(types.ContentTypeText),
+                                        },
+                                }
+                                if setup != nil {
+                                        setup(&props)
+                                }
+
+                                return &testStream{
+                                        inC:        make(chan *testStreamData, 16),
+                                        closedC:    make(chan struct{}),
+                                        properties: &props,
+                                }
                         }
 
                         Convey(`Will not add a stream with an invalid configuration.`, func() {
                                 // No content type.
-                                props.ContentType = ""
-
-                                s := newTestStream(props)
+                                s := newTestStream(func(p *streamproto.Properties) {
+                                        p.ContentType = ""
+                                })
                                 b := mkb(c, conf)
-                                So(b.AddStream(s, *s.properties), ShouldNotBeNil)
+                                So(b.AddStream(s, s.properties), ShouldNotBeNil)
                         })
 
                         Convey(`Will not add a stream with a duplicate stream name.`, func() {
                                 b := mkb(c, conf)
 
-                                s0 := newTestStream(props)
-                                So(b.AddStream(s0, *s0.properties), ShouldBeNil)
+                                s0 := newTestStream(nil)
+                                So(b.AddStream(s0, s0.properties), ShouldBeNil)
 
-                                s1 := newTestStream(props)
-                                So(b.AddStream(s1, *s1.properties), ShouldErrLike, "a stream has already been registered")
+                                s1 := newTestStream(nil)
+                                So(b.AddStream(s1, s1.properties), ShouldErrLike, "a stream has already been registered")
                         })
 
                         Convey(`Will not accept invalid tee configuration`, func() {
                                 conf.TeeStdout = nil
                                 conf.TeeStderr = nil
 
                                 for _, tc := range []struct {
                                         tee streamproto.TeeType
                                         err string
                                 }{
                                         {streamproto.TeeStdout, "no STDOUT is configured"},
                                         {streamproto.TeeStderr, "no STDERR is configured"},
                                         {streamproto.TeeType(0xFFFFFFFF), "invalid tee value"},
                                 } {
                                         Convey(fmt.Sprintf(`Rejects stream with TeeType [%v], when no tee outputs are configured.`, tc.tee), func() {
-                                                props.Tee = tc.tee
-
                                                 b := mkb(c, conf)
-                                                s := newTestStream(props)
-                                                So(b.AddStream(s, *s.properties), ShouldErrLike, tc.err)
+                                                s := newTestStream(func(p *streamproto.Properties) {
+                                                        p.Tee = tc.tee
+                                                })
+                                                So(b.AddStream(s, s.properties), ShouldErrLike, tc.err)
                                         })
                                 }
                         })
 
                         Convey(`When adding a stream configured to tee through STDOUT/STDERR, tees.`, func() {
-                                props.Name = "stdout"
-                                props.Tee = streamproto.TeeStdout
-                                stdout := newTestStream(props)
+                                stdout := newTestStream(func(p *streamproto.Properties) {
+                                        p.Name = "stdout"
+                                        p.Tee = streamproto.TeeStdout
+                                })
 
-                                props.Name = "stderr"
-                                props.Tee = streamproto.TeeStderr
-                                stderr := newTestStream(props)
+                                stderr := newTestStream(func(p *streamproto.Properties) {
+                                        p.Name = "stderr"
+                                        p.Tee = streamproto.TeeStderr
+                                })
 
                                 b := mkb(c, conf)
-                                So(b.AddStream(stdout, *stdout.properties), ShouldBeNil)
-                                So(b.AddStream(stderr, *stderr.properties), ShouldBeNil)
+                                So(b.AddStream(stdout, stdout.properties), ShouldBeNil)
+                                So(b.AddStream(stderr, stderr.properties), ShouldBeNil)
 
                                 stdout.data([]byte("Hello, STDOUT"), io.EOF)
                                 stderr.data([]byte("Hello, STDERR"), io.EOF)
 
                                 b.Activate()
                                 So(b.Wait(), ShouldBeNil)
 
                                 So(teeStdout.String(), ShouldEqual, "Hello, STDOUT")
                                 So(to.logs("stdout"), shouldHaveTextLogs, "Hello, STDOUT")
                                 So(to.isTerminal("stdout"), ShouldBeTrue)
 
                                 So(teeStderr.String(), ShouldEqual, "Hello, STDERR")
                                 So(to.logs("stderr"), shouldHaveTextLogs, "Hello, STDERR")
                                 So(to.isTerminal("stderr"), ShouldBeTrue)
                         })
 
                         Convey(`Run with 256 streams, stream{0..256} will deplete and finish.`, func() {
                                 b := mkb(c, conf)
                                 streams := make([]*testStream, 256)
                                 for i := range streams {
-                                        props.Name = fmt.Sprintf("stream%d", i)
-                                        streams[i] = newTestStream(props)
+                                        streams[i] = newTestStream(func(p *streamproto.Properties) {
+                                                p.Name = fmt.Sprintf("stream%d", i)
+                                        })
                                 }
 
                                 for _, s := range streams {
-                                        So(b.AddStream(s, *s.properties), ShouldBeNil)
+                                        So(b.AddStream(s, s.properties), ShouldBeNil)
                                         s.data([]byte("stream data 0!\n"), nil)
                                         s.data([]byte("stream data 1!\n"), nil)
                                 }
 
                                 // Add data to the streams after shutdown.
                                 for _, s := range streams {
                                         s.data([]byte("stream data 2!\n"), io.EOF)
                                 }
 
                                 b.Activate()
                                 So(b.Wait(), ShouldBeNil)
 
                                 for _, s := range streams {
                                         name := string(s.properties.Name)
 
                                         So(to.logs(name), shouldHaveTextLogs, "stream data 0!", "stream data 1!", "stream data 2!")
                                         So(to.isTerminal(name), ShouldBeTrue)
                                 }
                         })
 
                         Convey(`Shutdown with 256 in-progress streams, stream{0..256} will terminate if they emitted logs.`, func() {
                                 b := mkb(c, conf)
                                 streams := make([]*testStream, 256)
                                 for i := range streams {
-                                        props.Name = fmt.Sprintf("stream%d", i)
-                                        streams[i] = newTestStream(props)
+                                        streams[i] = newTestStream(func(p *streamproto.Properties) {
+                                                p.Name = fmt.Sprintf("stream%d", i)
+                                        })
                                 }
 
                                 for _, s := range streams {
-                                        So(b.AddStream(s, *s.properties), ShouldBeNil)
+                                        So(b.AddStream(s, s.properties), ShouldBeNil)
                                         s.data([]byte("stream data!\n"), nil)
                                 }
 
                                 b.shutdown(errors.New("test shutdown"))
                                 So(b.Wait(), ShouldErrLike, "test shutdown")
 
                                 for _, s := range streams {
                                         if len(to.logs(s.properties.Name)) > 0 {
                                                 So(to.isTerminal(string(s.properties.Name)), ShouldBeTrue)
                                         } else {
                                                 So(to.isTerminal(string(s.properties.Name)), ShouldBeFalse)
                                         }
                                 }
                         })
 
                         Convey(`Using ten test stream servers`, func() {
                                 servers := make([]*testStreamServer, 10)
                                 for i := range servers {
                                         servers[i] = newTestStreamServer()
                                 }
                                 streams := []*testStream(nil)
 
                                 Convey(`Can register both before Run and will retain streams.`, func() {
                                         b := mkb(c, conf)
                                         for i, tss := range servers {
                                                 b.AddStreamServer(tss)
 
-                                                props.Name = fmt.Sprintf("stream%d", i)
-                                                s := newTestStream(props)
+                                                s := newTestStream(func(p *streamproto.Properties) {
+                                                        p.Name = fmt.Sprintf("stream%d", i)
+                                                })
                                                 streams = append(streams, s)
                                                 s.data([]byte("test data"), io.EOF)
                                                 tss.enqueue(s)
                                         }
 
                                         b.Activate()
                                         So(b.Wait(), ShouldBeNil)
 
                                         for _, s := range streams {
                                                 So(to.logs(s.properties.Name), shouldHaveTextLogs, "test data")
                                                 So(to.isTerminal(s.properties.Name), ShouldBeTrue)
                                         }
                                 })
 
                                 Convey(`Can register both during Run and will retain streams.`, func() {
                                         b := mkb(c, conf)
                                         for i, tss := range servers {
                                                 b.AddStreamServer(tss)
 
-                                                props.Name = fmt.Sprintf("stream%d", i)
-                                                s := newTestStream(props)
+                                                s := newTestStream(func(p *streamproto.Properties) {
+                                                        p.Name = fmt.Sprintf("stream%d", i)
+                                                })
                                                 streams = append(streams, s)
                                                 s.data([]byte("test data"), io.EOF)
                                                 tss.enqueue(s)
                                         }
 
                                         b.Activate()
                                         So(b.Wait(), ShouldBeNil)
 
                                         for _, s := range streams {
                                                 So(to.logs(s.properties.Name), shouldHaveTextLogs, "test data")
                                                 So(to.isTerminal(s.properties.Name), ShouldBeTrue)
                                         }
                                 })
                         })
 
                         Convey(`Will ignore stream registration errors, allowing re-registration.`, func() {
                                 tss := newTestStreamServer()
 
                                 // Generate an invalid stream for "tss" to register.
-                                sGood := newTestStream(props)
+                                sGood := newTestStream(nil)
                                 sGood.data([]byte("good test data"), io.EOF)
 
-                                props.ContentType = ""
-                                sBad := newTestStream(props)
+                                sBad := newTestStream(func(p *streamproto.Properties) {
+                                        p.ContentType = ""
+                                })
                                 sBad.data([]byte("bad test data"), io.EOF)
 
                                 b := mkb(c, conf)
                                 b.AddStreamServer(tss)
                                 tss.enqueue(sBad)
                                 tss.enqueue(sGood)
                                 b.Activate()
                                 So(b.Wait(), ShouldBeNil)
 
                                 So(sBad.isClosed(), ShouldBeTrue)
                                 So(sGood.isClosed(), ShouldBeTrue)
-                                So(to.logs(props.Name), shouldHaveTextLogs, "good test data")
-                                So(to.isTerminal(props.Name), ShouldBeTrue)
+                                So(to.logs("test"), shouldHaveTextLogs, "good test data")
+                                So(to.isTerminal("test"), ShouldBeTrue)
                         })
                 })
 
                 Convey(`Will terminate if the stream server panics.`, func() {
                         tss := newTestStreamServer()
                         tss.onNext = func() {
                                 panic("test panic")
                         }
 
                         b := mkb(c, conf)
                         b.AddStreamServer(tss)
                         So(b.Wait(), ShouldErrLike, "test panic")
                 })
         })
 }