LogDog: Update to archival V2.

appengine/logdog/coordinator/endpoints/logs/query_test.go

 // Copyright 2015 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 logs
 
 import (
         "fmt"
         "sort"
         "testing"
         "time"
 
         "github.com/luci/gae/filter/featureBreaker"
         "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/api/logdog_coordinator/logs/v1"
         "github.com/luci/luci-go/common/clock/testclock"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/logdog/types"
         "github.com/luci/luci-go/common/proto/google"
         "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"
         "golang.org/x/net/context"
 
+        "github.com/luci/luci-go/common/logging/gologger"
+
         . "github.com/luci/luci-go/common/testing/assertions"
         . "github.com/smartystreets/goconvey/convey"
 )
 
 func shouldHaveLogPaths(actual interface{}, expected ...interface{}) string {
         resp := actual.(*logdog.QueryResponse)
         var paths []string
         if len(resp.Streams) > 0 {
                 paths = make([]string, len(resp.Streams))
                 for i, s := range resp.Streams {
@@ skipping 20 matching lines @@
 
         return ShouldResemble(paths, exp)
 }
 
 func TestQuery(t *testing.T) {
         t.Parallel()
 
         Convey(`With a testing configuration, a Query request`, t, func() {
                 c, tc := testclock.UseTime(context.Background(), testclock.TestTimeLocal)
                 c = memory.Use(c)
+                c = gologger.Use(c)
                 c, fb := featureBreaker.FilterRDS(c, nil)
 
                 di := ds.Get(c)
 
                 // Add LogStream indexes.
                 //
                 // These should be kept in sync with the "query endpoint" indexes in the
                 // vmuser module's "index.yaml" file.
                 indexDefs := [][]string{
                         {"Prefix", "-Created"},
@@ skipping 20 matching lines @@
                                 }
                         }
                         indexes[i] = &ds.IndexDefinition{Kind: "LogStream", SortBy: cols}
                 }
                 di.Testable().AddIndexes(indexes...)
                 di.Testable().Consistent(true)
 
                 fs := authtest.FakeState{}
                 c = auth.WithState(c, &fs)
 
-                c = ct.UseConfig(c, &svcconfig.Coordinator{
-                        AdminAuthGroup: "test-administrators",
-                })
+                svcStub := ct.Services{}
+                svcStub.InitConfig()
+                svcStub.ServiceConfig.Coordinator.AdminAuthGroup = "test-administrators"
 
-                s := Server{}
+                s := Server{
+                        ServiceBase: coordinator.ServiceBase{&svcStub},
+                }
 
                 req := logdog.QueryRequest{
                         Tags: map[string]string{},
                 }
 
                 // Install a set of stock log streams to query against.
                 var streamPaths []string
                 var purgedStreamPaths []string
                 streams := map[string]*coordinator.LogStream{}
                 descs := map[string]*logpb.LogStreamDescriptor{}
@@ skipping 22 matching lines @@
                                 "name":   string(name),
                         }
 
                         // Set an empty tag for each name segment.
                         for _, p := range name.Segments() {
                                 desc.Tags[p] = ""
                         }
 
                         ls := ct.TestLogStream(c, desc)
 
+                        now := tc.Now().UTC()
                         psegs := prefix.Segments()
                         if psegs[0] == "meta" {
                                 for _, p := range psegs[1:] {
                                         switch p {
                                         case "purged":
                                                 ls.Purged = true
 
                                         case "terminated":
-                                                ls.State = coordinator.LSTerminated
                                                 ls.TerminalIndex = 1337
+                                                ls.TerminatedTime = now
 
                                         case "archived":
+                                                ls.State = coordinator.LSArchived
+
+                                                ls.TerminalIndex = 1337
+                                                ls.TerminatedTime = now
+
                                                 ls.ArchiveStreamURL = "http://example.com"
-                                                ls.State = coordinator.LSArchived
+                                                ls.ArchivedTime = now
 
                                         case "datagram":
                                                 ls.StreamType = logpb.StreamType_DATAGRAM
 
                                         case "binary":
                                                 ls.StreamType = logpb.StreamType_BINARY
                                         }
                                 }
                         }
 
-                        if err := ls.Put(ds.Get(c)); err != nil {
+                        if err := ds.Get(c).Put(ls); err != nil {
                                 panic(fmt.Errorf("failed to put log stream %d: %v", i, err))
                         }
 
                         descs[string(v)] = desc
                         streams[string(v)] = ls
                         if !ls.Purged {
                                 streamPaths = append(streamPaths, string(v))
                         }
                         purgedStreamPaths = append(purgedStreamPaths, string(v))
                         tc.Add(time.Second)
@@ skipping 72 matching lines @@
                                         So(err, ShouldBeRPCOK)
                                         So(resp, shouldHaveLogPaths, "testing/+/baz")
 
                                         So(resp.Streams, ShouldHaveLength, 1)
                                         So(resp.Streams[0].State, ShouldResemble, loadLogStreamState(stream))
                                         So(resp.Streams[0].Desc, ShouldResemble, desc)
                                         So(resp.Streams[0].DescProto, ShouldBeNil)
                                 })
 
                                 Convey(`When not requesting protobufs, and with a corrupt descriptor, returns InternalServer error.`, func() {
-                                        // We can't use "stream.Put" here because it validates the protobuf!
+                                        stream.SetDSValidate(false)
                                         stream.Descriptor = []byte{0x00} // Invalid protobuf, zero tag.
                                         So(di.Put(stream), ShouldBeNil)
                                         di.Testable().CatchupIndexes()
 
                                         _, err := s.Query(c, &req)
                                         So(err, ShouldBeRPCInternal)
                                 })
 
                                 Convey(`When requesting protobufs, returns the raw protobuf descriptor.`, func() {
                                         req.Proto = true
@@ skipping 220 matching lines @@
 
                         Convey(`When an invalid tag is specified, returns BadRequest error`, func() {
                                 req.Tags["+++not a valid tag+++"] = ""
 
                                 _, err := s.Query(c, &req)
                                 So(err, ShouldBeRPCInvalidArgument, "invalid tag constraint")
                         })
                 })
         })
 }