LogDog: Add project namespace to Butler/Collector.

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

Index: server/internal/logdog/collector/coordinator/cache_test.go
diff --git a/server/internal/logdog/collector/coordinator/cache_test.go b/server/internal/logdog/collector/coordinator/cache_test.go
index a5d2eb718406c9f18857a9f00f8e044d9226db12..1748042ee996462632cbd44a7b72a05b8f6b655a 100644
--- a/server/internal/logdog/collector/coordinator/cache_test.go
+++ b/server/internal/logdog/collector/coordinator/cache_test.go
@@ -12,6 +12,7 @@ import (
 	"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"
@@ -76,6 +77,7 @@ func TestStreamStateCache(t *testing.T) {
 		tcc := testCoordinator{}
 
 		st := LogStreamState{
+			Project:       "test-project",
 			Path:          "foo/+/bar",
 			TerminalIndex: -1,
 		}
@@ -166,6 +168,7 @@ func TestStreamStateCache(t *testing.T) {
 				terminalErrC := make(chan error)
 				go func() {
 					terminalErrC <- ssc.TerminateStream(c, &LogStreamState{
+						Project:       st.Project,
 						Path:          st.Path,
 						TerminalIndex: 1337,
 					})
@@ -245,6 +248,30 @@ func TestStreamStateCache(t *testing.T) {
 					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() {