LogDog: Add project namespace to Butler/Collector.

server/internal/logdog/collector/coordinator/cache_test.go

 // 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 coordinator
 
 import (
         "fmt"
         "sync"
         "sync/atomic"
         "testing"
         "time"
 
         "github.com/luci/luci-go/common/clock/testclock"
+        "github.com/luci/luci-go/common/config"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/logdog/types"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
         "golang.org/x/net/context"
 
         . "github.com/smartystreets/goconvey/convey"
 )
 
 // testCoordinator is an implementation of Coordinator that can be used for
 // testing.
@@ skipping 44 matching lines @@
 }
 
 func TestStreamStateCache(t *testing.T) {
         t.Parallel()
 
         Convey(`Using a test configuration`, t, func() {
                 c, tc := testclock.UseTime(context.Background(), testclock.TestTimeLocal)
                 tcc := testCoordinator{}
 
                 st := LogStreamState{
+                        Project:       "test-project",
                         Path:          "foo/+/bar",
                         TerminalIndex: -1,
                 }
 
                 // Note: In all of these tests, we check if "proto" field (ProtoVersion)
                 // is "remote". We use ProtoVersion as a channel between our fake remote
                 // service. When our fake remote service returns a LogStreamState, it sets
                 // "remote" to true to differentiate it from the local pushed state.
                 //
                 // If a LogStreamState has "remote" set to true, that implies that it was
@@ skipping 70 matching lines @@
                                 <-tcc.callC
 
                                 // Set the terminal index. We will use a minimal LogStreamState. We know
                                 // that this will happen after the streamStateCacheEntry is registered
                                 // because both block on the LRU cache's Mutate, which is atomic, and
                                 // RegisterStream must have added the streamStateCacheEntry in order for
                                 // the lock to be available for TerminateStream to proceed.
                                 terminalErrC := make(chan error)
                                 go func() {
                                         terminalErrC <- ssc.TerminateStream(c, &LogStreamState{
+                                                Project:       st.Project,
                                                 Path:          st.Path,
                                                 TerminalIndex: 1337,
                                         })
                                 }()
 
                                 // Let both requests succeed.
                                 <-tcc.callC
                                 tcc.errC <- nil
                                 tcc.errC <- nil
 
@@ skipping 59 matching lines @@
                                 So(tcc.calls, ShouldEqual, 1)
 
                                 Convey(`A second registration without error will make a new request.`, func() {
                                         tcc.errC = nil
 
                                         _, err := ssc.RegisterStream(c, &st, nil)
                                         So(err, ShouldBeNil)
                                         So(tcc.calls, ShouldEqual, 2)
                                 })
                         })
+
+                        Convey(`Different projects with the sme stream name will not conflict.`, func() {
+                                var projects = []config.ProjectName{"", "foo", "bar"}
+
+                                for i, p := range projects {
+                                        st.Project = p
+                                        s, err := ssc.RegisterStream(c, &st, nil)
+                                        So(err, ShouldBeNil)
+
+                                        s.TerminalIndex = types.MessageIndex(i)
+                                        So(ssc.TerminateStream(c, s), ShouldBeNil)
+                                }
+                                So(tcc.calls, ShouldEqual, len(projects)*2)
+
+                                for i, p := range projects {
+                                        st.Project = p
+                                        st.TerminalIndex = -1
+
+                                        s, err := ssc.RegisterStream(c, &st, nil)
+                                        So(err, ShouldBeNil)
+                                        So(s.TerminalIndex, ShouldEqual, types.MessageIndex(i))
+                                }
+                                So(tcc.calls, ShouldEqual, len(projects)*2)
+                        })
                 })
 
                 Convey(`A streamStateCache can register multiple streams at once.`, func() {
                         ssc := NewCache(&tcc, 0, 0)
                         tcc.callC = make(chan struct{})
                         tcc.errC = make(chan error)
 
                         count := 2048
                         wg := sync.WaitGroup{}
                         errs := make(errors.MultiError, count)
@@ skipping 28 matching lines @@
                         remotes := 0
                         for i := 0; i < count; i++ {
                                 if state[i].ProtoVersion == "remote" {
                                         remotes++
                                 }
                         }
                         So(remotes, ShouldEqual, count)
                 })
         })
 }