LogDog: Add log rendering view.

appengine/logdog/coordinator/logView/view_test.go

Index: appengine/logdog/coordinator/logView/view_test.go
diff --git a/appengine/logdog/coordinator/logView/view_test.go b/appengine/logdog/coordinator/logView/view_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..5dd81a639b1c2bc593f69aa2ff0ce5519a5093b5
--- /dev/null
+++ b/appengine/logdog/coordinator/logView/view_test.go
@@ -0,0 +1,285 @@
+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+package logView
+
+import (
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/golang/protobuf/proto"
+	"github.com/julienschmidt/httprouter"
+	"github.com/luci/gae/impl/memory"
+	ds "github.com/luci/gae/service/datastore"
+	"github.com/luci/luci-go/appengine/logdog/coordinator"
+	ct "github.com/luci/luci-go/appengine/logdog/coordinator/coordinatorTest"
+	"github.com/luci/luci-go/common/clock"
+	"github.com/luci/luci-go/common/clock/testclock"
+	"github.com/luci/luci-go/common/logdog/types"
+	"github.com/luci/luci-go/common/logging"
+	"github.com/luci/luci-go/common/logging/gologger"
+	"github.com/luci/luci-go/common/proto/logdog/logpb"
+	"github.com/luci/luci-go/common/proto/logdog/svcconfig"
+	"github.com/luci/luci-go/server/auth"
+	"github.com/luci/luci-go/server/auth/authtest"
+	"github.com/luci/luci-go/server/logdog/storage"
+	"github.com/luci/luci-go/server/middleware"
+	"golang.org/x/net/context"
+
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+type simulatedStorageInsn struct {
+	idx int64
+	d   [][]byte
+	err error
+}
+
+type simulatedStorage struct {
+	storage.Storage
+
+	insnC   chan *simulatedStorageInsn
+	gotC    chan struct{}
+	closedC chan struct{}
+}
+
+func (s *simulatedStorage) Get(req *storage.GetRequest, cb storage.GetCallback) error {
+	insn := <-s.insnC
+	if insn.err != nil {
+		return insn.err
+	}
+
+	defer func() {
+		if s.gotC != nil {
+			s.gotC <- struct{}{}
+		}
+	}()
+
+	for i, d := range insn.d {
+		if !cb(types.MessageIndex(insn.idx+int64(i)), d) {
+			return nil
+		}
+	}
+	return nil
+}
+
+func (s *simulatedStorage) Close() {
+	close(s.insnC)
+	close(s.closedC)
+}
+
+func (s *simulatedStorage) err(err error) {
+	s.insnC <- &simulatedStorageInsn{err: err}
+}
+
+func (s *simulatedStorage) text(startIdx int64, lines ...string) {
+	d := make([][]byte, len(lines))
+	for i, v := range lines {
+		line := logpb.Text_Line{
+			Value: v,
+		}
+		if strings.HasSuffix(line.Value, "\n") {
+			line.Value = strings.TrimSuffix(line.Value, "\n")
+			line.Delimiter = "\n"
+		}
+
+		pb := logpb.LogEntry{
+			StreamIndex: uint64(startIdx + int64(i)),
+			Content: &logpb.LogEntry_Text{
+				Text: &logpb.Text{
+					Lines: []*logpb.Text_Line{
+						&line,
+					},
+				},
+			},
+		}
+
+		data, err := proto.Marshal(&pb)
+		require(err)
+		d[i] = data
+	}
+	s.data(startIdx, d...)
+}
+
+func (s *simulatedStorage) data(startIdx int64, d ...[]byte) {
+	s.insnC <- &simulatedStorageInsn{idx: startIdx, d: d}
+}
+
+// testBase is a middleware.Base which uses its current Context as the base
+// context.
+type testBase struct {
+	context.Context
+}
+
+func (t *testBase) base(h middleware.Handler) httprouter.Handle {
+	return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
+		h(t.Context, w, r, p)
+	}
+}
+
+func TestLogView(t *testing.T) {
+	Convey(`With a testing configuration, a LogView handler`, t, func() {
+		c, tc := testclock.UseTime(context.Background(), testclock.TestTimeLocal)
+		c = memory.Use(c)
+		c = gologger.Use(c) // XXX: Remove me.
+		c = logging.SetLevel(c, logging.Debug)
+
+		fs := authtest.FakeState{}
+		c = auth.WithState(c, &fs)
+
+		c = ct.UseConfig(c, &svcconfig.Coordinator{
+			AdminAuthGroup: "test-administrators",
+		})
+
+		st := simulatedStorage{
+			insnC:   make(chan *simulatedStorageInsn, 1),
+			closedC: make(chan struct{}),
+		}
+
+		desc := ct.TestLogStreamDescriptor(c, "foo/bar")
+		ls, err := ct.TestLogStream(c, desc)
+		require(err)
+
+		err = ls.Put(ds.Get(c))
+		require(err)
+
+		h := Handler{
+			Service: coordinator.Service{
+				IntermediateStorageFunc: func(c context.Context) (storage.Storage, error) {
+					return &st, nil
+				},
+			},
+		}
+
+		r := httprouter.New()
+		h.InstallHandlers(r, func(h middleware.Handler) httprouter.Handle {
+			return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
+				h(c, w, r, p)
+			}
+		})
+
+		srv := httptest.NewServer(r)
+		defer srv.Close()
+
+		// Executes a view request for the named stream/hash against our server.
+		view := func(v string) (string, int) {
+			u, err := url.Parse(srv.URL)
+			require(err)
+
+			u.Path = fmt.Sprintf("/logs/view/%s", v)
+			resp, err := http.Get(u.String())
+			require(err)
+			defer resp.Body.Close()
+
+			b, err := ioutil.ReadAll(resp.Body)
+			require(err)
+			return string(b), resp.StatusCode
+		}
+
+		Convey(`Will return Bad Request for an invalid log stream path/hash.`, func() {
+			_, err := view("!!!invalid!!!")
+			So(err, ShouldEqual, http.StatusBadRequest)
+		})
+
+		Convey(`Will return Not Found for a non-existent log stream.`, func() {
+			_, err := view("does/not/+/exist")
+			So(err, ShouldEqual, http.StatusNotFound)
+		})
+
+		Convey(`When Storage has a finished log with "Line0\nLine1\nLine2\n"`, func() {
+			st.text(0, "Line0\n", "Line1\n", "Line2\n")
+			ls.TerminalIndex = 2
+			require(ls.Put(ds.Get(c)))
+
+			Convey(`Can read the log stream.`, func() {
+				txt, err := view(string(ls.Path()))
+				So(err, ShouldEqual, http.StatusOK)
+				So(txt, ShouldEqual, "Line0\nLine1\nLine2\n")
+			})
+
+			Convey(`When a log stream has been purged`, func() {
+				ls.Purged = true
+				require(ls.Put(ds.Get(c)))
+
+				Convey(`Will return Not Found if the user is not admin.`, func() {
+					_, err := view(string(ls.Path()))
+					So(err, ShouldEqual, http.StatusNotFound)
+				})
+
+				Convey(`Will return the log stream if the user is admin.`, func() {
+					fs.IdentityGroups = []string{"test-administrators"}
+
+					txt, err := view(string(ls.Path()))
+					So(err, ShouldEqual, http.StatusOK)
+					So(txt, ShouldEqual, "Line0\nLine1\nLine2\n")
+				})
+			})
+		})
+
+		Convey(`Can stream a non-terminal log.`, func() {
+			st.gotC = make(chan struct{})
+
+			go func() {
+				// Put initial log record.
+				st.text(0, "Line0\n")
+				<-st.gotC
+
+				// Skip a log record. This should be ignored since it's non-contiguous.
+				st.text(2, "Line2\n")
+
+				// When we sleep pending a new log, add log line #1. This will happen
+				// AFTER the following "gotC" block. We do this to ensure that the
+				// callback is in place when the timer needs it.
+				tc.SetTimerCallback(func(d time.Duration, t clock.Timer) {
+					st.text(1, "Line1\n")
+					tc.SetTimerCallback(nil)
+					tc.Add(d)
+				})
+
+				<-st.gotC
+
+				// The log stream should sleep pending new logs. When it does, our timer
+				// callback will add the missing log (once).
+				<-st.gotC
+
+				// Mark the log stream as terminal @3.
+				ls.TerminalIndex = 3
+				require(ls.Put(ds.Get(c)))
+
+				// Add the missing logs.
+				st.text(2, "Line2\n", "Line3\n")
+				<-st.gotC
+			}()
+
+			txt, err := view(string(ls.Path()))
+			So(err, ShouldEqual, http.StatusOK)
+			So(txt, ShouldEqual, "Line0\nLine1\nLine2\nLine3\n")
+		})
+
+		Convey(`Will return InternalServerError if our storage returns an error.`, func() {
+			st.err(errors.New("test error"))
+			_, err := view(string(ls.Path()))
+			So(err, ShouldEqual, http.StatusInternalServerError)
+		})
+
+		Convey(`Will return InternalServerError if our storage returns junk log data.`, func() {
+			st.data(0, []byte{0x00})
+			_, err := view(string(ls.Path()))
+			So(err, ShouldEqual, http.StatusInternalServerError)
+		})
+	})
+}
+
+func require(err error) {
+	if err != nil {
+		panic(err)
+	}
+}