.

go/zodb/zodbtools/watch.go

@@ -63,7 +63,7 @@ func Watch(ctx context.Context, stor zodb.IStorage, w io.Writer, verbose bool) (
 	defer xerr.Contextf(&err, "%s: watch", stor.URL())
 
 	emitf := func(format string, argv ...interface{}) error {
-		_, err := fmt.Fprintf(w, format, argv)
+		_, err := fmt.Fprintf(w, format, argv...)
 		return err
 	}
 
@@ -150,7 +150,7 @@ func watchMain(argv []string) {
 	// TODO defer stor.Close()
 
 	err = Watch(ctx, stor, os.Stdout, verbose)
-	if err != nil { // XXX & not canceled
+	if err != nil {
 		prog.Fatal(err)
 	}
 }

go/zodb/zodbtools/watch_test.go

@@ -20,10 +20,101 @@
 package zodbtools
 
 import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
 	"testing"
-)
 
+	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
+
+	"lab.nexedi.com/kirr/go123/exc"
+	"lab.nexedi.com/kirr/neo/go/internal/xtesting"
+	"lab.nexedi.com/kirr/neo/go/zodb"
+)
 
 func TestWatch(t *testing.T) {
-	// XXX
+	X := exc.Raiseif
+
+	xtesting.NeedPy(t, "zodbtools")
+	work, err := ioutil.TempDir("", "t-zodbwatch"); X(err)
+	defer os.RemoveAll(work)
+	tfs := work + "/t.fs"
+
+	// force tfs creation
+	at := zodb.Tid(0)
+	xcommit := func(objv ...xtesting.ZRawObject) {
+		t.Helper()
+		var err error
+		at, err = xtesting.ZPyCommitRaw(tfs, at, objv...)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	xcommit(xtesting.ZRawObject{0, []byte("data0")})
+
+	// spawn plain and verbose watchers
+	ctx0, cancel := context.WithCancel(context.Background())
+	wg, ctx := errgroup.WithContext(ctx0)
+
+	// gowatch spawns Watch(verbose) and returns expect(line) func that is
+	// connected to Watch output.
+	gowatch := func(verbose bool) /*expectf*/func(format string, argv ...interface{}) {
+		pr, pw := io.Pipe()
+		wg.Go(func() error {
+			stor, err := zodb.OpenStorage(ctx, tfs, &zodb.OpenOptions{ReadOnly: true})
+			if err != nil {
+				return err
+			}
+			return Watch(ctx, stor, pw, verbose)
+		})
+
+		r := bufio.NewReader(pr)
+		expectf := func(format string, argv ...interface{}) {
+			t.Helper()
+			l, err := r.ReadString('\n')
+			if err != nil {
+				t.Fatalf("expect: %s", err)
+			}
+			l = l[:len(l)-1] // trim trailing \n
+			line := fmt.Sprintf(format, argv...)
+			if l != line {
+				t.Fatalf("expect\nhave: %q\nwant: %q", l, line)
+			}
+		}
+		return expectf
+	}
+
+	pexpect := gowatch(false)
+	vexpect := gowatch(true)
+
+	pexpect("# at %s", at)
+	vexpect("# at %s", at)
+
+	xcommit(xtesting.ZRawObject{0, []byte("data01")})
+
+	pexpect("txn %s", at)
+	vexpect("txn %s", at)
+	vexpect("obj 0000000000000000")
+	vexpect("")
+
+	xcommit(xtesting.ZRawObject{1, []byte("data1")}, xtesting.ZRawObject{2, []byte("data2")})
+
+	pexpect("txn %s", at)
+	vexpect("txn %s", at)
+	vexpect("obj 0000000000000001")
+	vexpect("obj 0000000000000002")
+	vexpect("")
+
+	cancel()
+
+	err = wg.Wait()
+	ecause := errors.Cause(err)
+	if ecause != context.Canceled {
+		t.Fatalf("finished: err: expected 'canceled' cause; got %q", err)
+	}
 }