[logdog] Replace Tumble with push queues.

logdog/appengine/coordinator/coordinatorTest/context.go

 // Copyright 2015 The LUCI Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //      http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
 package coordinatorTest
 
 import (
         "fmt"
+        "path/filepath"
         "strings"
         "time"
 
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/clock/testclock"
         "github.com/luci/luci-go/common/config/impl/memory"
         "github.com/luci/luci-go/common/data/caching/cacheContext"
         "github.com/luci/luci-go/common/gcloud/gs"
         "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/logging/gologger"
         configPB "github.com/luci/luci-go/common/proto/config"
         "github.com/luci/luci-go/common/proto/google"
         "github.com/luci/luci-go/logdog/api/config/svcconfig"
         "github.com/luci/luci-go/logdog/appengine/coordinator"
         "github.com/luci/luci-go/logdog/appengine/coordinator/config"
+        "github.com/luci/luci-go/logdog/appengine/coordinator/tasks"
         "github.com/luci/luci-go/logdog/common/storage/archive"
         "github.com/luci/luci-go/logdog/common/storage/bigtable"
         "github.com/luci/luci-go/luci_config/common/cfgtypes"
         "github.com/luci/luci-go/luci_config/server/cfgclient"
         "github.com/luci/luci-go/luci_config/server/cfgclient/backend/testconfig"
         "github.com/luci/luci-go/luci_config/server/cfgclient/textproto"
         "github.com/luci/luci-go/server/auth"
         "github.com/luci/luci-go/server/auth/authtest"
         "github.com/luci/luci-go/server/auth/identity"
+        "github.com/luci/luci-go/server/router"
         "github.com/luci/luci-go/server/settings"
         "github.com/luci/luci-go/tumble"
 
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/gae/service/info"
+        tq "github.com/luci/gae/service/taskqueue"
 
         "github.com/golang/protobuf/proto"
         "golang.org/x/net/context"
 )
 
 // Environment contains all of the testing facilities that are installed into
 // the Context.
 type Environment struct {
         // Tumble is the Tumble testing instance.
         Tumble tumble.Testing
@@ skipping 92 matching lines @@
 
 func (e *Environment) addConfigEntry(configSet cfgtypes.ConfigSet, path, content string) {
         cset := e.Config[string(configSet)]
         if cset == nil {
                 cset = make(map[string]string)
                 e.Config[string(configSet)] = cset
         }
         cset[path] = content
 }
 
+// RunTaskQueues processes all tasks in all task queues if they are available.
+func (e *Environment) RunTaskQueues(c context.Context, tls *TestStream) {
+        r := router.New()
+
+        tls.WithProjectNamespace(c, func(c context.Context) {
+                tasks.InstallHandlers(r, router.NewMiddlewareChain(func(ctx *router.Context, next router.Handler) {
+                        ctx.Context = c
+                        next(ctx)
+                }))
+
+                drainTaskQueues(c, r)
+        })
+}
+
 // Install creates a testing Context and installs common test facilities into
 // it, returning the Environment to which they're bound.
 func Install() (context.Context, *Environment) {
         e := Environment{
                 Config:   make(map[string]memory.ConfigSet),
                 GSClient: GSClient{},
                 StorageCache: StorageCache{
                         Base: &coordinator.StorageCache{},
                 },
         }
 
         // Get our starting context. This installs, among other things, in-memory
         // gae, settings, and logger.
         c := e.Tumble.Context()
         if *testGoLogger {
                 c = logging.SetLevel(gologger.StdConfig.Use(c), logging.Debug)
         }
 
         // Create/install our BigTable memory instance.
         e.BigTable = bigtable.NewMemoryInstance(c, bigtable.Options{
                 Cache: &e.StorageCache,
         })
 
-        // Add indexes. These should match the indexes defined in the application's
-        // "index.yaml".
-        indexDefs := [][]string{
-                {"Prefix", "-Created"},
-                {"Name", "-Created"},
-                {"State", "-Created"},
-                {"Purged", "-Created"},
-                {"ProtoVersion", "-Created"},
-                {"ContentType", "-Created"},
-                {"StreamType", "-Created"},
-                {"Timestamp", "-Created"},
-                {"_C", "-Created"},
-                {"_Tags", "-Created"},
-                {"_Terminated", "-Created"},
-                {"_Archived", "-Created"},
+        // Register our task queues.
+        tq.GetTestable(c).CreateQueue(tasks.ArchivalTaskQueue)
+
+        // Load indexes from "index.yaml".
+        mainServicePath := filepath.Join("..", "..", "..", "cmd", "coordinator", "vmuser")
+        indexDefs, err := ds.FindAndParseIndexYAML(mainServicePath)
+        if err != nil {
+                panic(fmt.Errorf("failed to load 'index.yaml': %s", err))
         }
-        indexes := make([]*ds.IndexDefinition, len(indexDefs))
-        for i, id := range indexDefs {
-                cols := make([]ds.IndexColumn, len(id))
-                for j, ic := range id {
-                        var err error
-                        cols[j], err = ds.ParseIndexColumn(ic)
-                        if err != nil {
-                                panic(fmt.Errorf("failed to parse index %q: %s", ic, err))
-                        }
-                }
-                indexes[i] = &ds.IndexDefinition{Kind: "LogStream", SortBy: cols}
-        }
-        ds.GetTestable(c).AddIndexes(indexes...)
+        ds.GetTestable(c).AddIndexes(indexDefs...)
 
         // Setup clock.
         e.Clock = clock.Get(c).(testclock.TestClock)
 
         // Install GAE config service settings.
         c = settings.Use(c, settings.New(&settings.MemoryStorage{}))
 
         // Setup luci-config configuration.
         c = testconfig.WithCommonClient(c, memory.New(e.Config))
 
@@ skipping 108 matching lines @@
 }
 
 // WithProjectNamespace runs f in proj's namespace, bypassing authentication
 // checks.
 func WithProjectNamespace(c context.Context, proj cfgtypes.ProjectName, f func(context.Context)) {
         if err := coordinator.WithProjectNamespace(&c, proj, coordinator.NamespaceAccessAllTesting); err != nil {
                 panic(err)
         }
         f(c)
 }