.

wcfs/wcfs_test.py

@@ -31,7 +31,7 @@ from wendelin import wcfs
 import transaction
 from persistent import Persistent
 from persistent.timestamp import TimeStamp
-from ZODB.utils import z64
+from ZODB.utils import z64, u64, p64
 
 import sys, os, os.path, subprocess, threading, inspect, traceback
 from errno import EINVAL
@@ -104,6 +104,14 @@ def _xdefer(f):
     defer(_)
 
 
+# many tests need to be run with some reasonable timeout to detect lack of wcfs
+# response. with_timout and timeout provides syntatic shortcuts to do so.
+def with_timeout(parent=context.background()):  # -> ctx, cancel
+    return context.with_timeout(parent, 3*time.second)
+
+def timeout(parent=context.background()):   # -> ctx
+    ctx, _ = with_timeout()
+    return ctx
 
 
 # ---- test join/autostart ----
@@ -754,7 +762,7 @@ def watch(twlink, zf, at):   # XXX -> ?
     # XXX detect not sent pins with timeout, or better via ack'ing previous
     # pins as they come in (not waiting for all of them) and then seeing that
     # we did not received expeced pin when wcfs sends final ok?
-    ctx, cancel = context.with_timeout(context.background(), 3*time.second)
+    ctx, cancel = with_timeout()
     wg = sync.WorkGroup(ctx)
 
     def _(ctx):
@@ -899,20 +907,29 @@ def test_wcfs():
     assert wl.fatalv == []
     wl._write(b'zzz hello\n')
     _, _rx = select(
-        time.after(3*time.second).recv,
+        timeout().done().recv,
         wl.rx_eof.recv,
     )
     if _ == 0:
         raise RuntimeError("%s: did not rx EOF after bad frame " % wl)
     assert wl.fatalv == [b'error: invalid frame: "zzz hello\\n" (invalid stream)']
+    wl.close()
 
+    # watch with @at < δtail.tail -> rejected
+    print('\n\nAAA\n\n')
+    wl = t.openwatch()
+    assert wl.sendReq(timeout(), b"watch %s @%s" % (h(zf._p_oid), h(p64(u64(at0)-1)))) == \
+            "XXX"
     wl.close()
 
+    return  # XXX
+
+
     # valid watch requests going at_i -> at_j -> ...
     for zf in t.zfiles():
         for revv in t.iter_revv():
             print('\n--------')
-            #print(' -> '.join([t.hat(_) for _ in revv]))
+            print(' -> '.join([t.hat(_) for _ in revv]))
             wl = t.openwatch()
             wl.watch(zf, revv[0])
             wl.watch(zf, revv[0])    # verify at_i -> at_i
@@ -932,7 +949,6 @@ def test_wcfs():
     # XXX watch for 2 files via single watch open
 
     # XXX watch with @at > head - must wait for head to become >= at
-    # XXX watch with @at < δtail.tail -> rejected
     # XXX watch with at="-" -> stop watching
 
     # XXX drop file[blk] from cache, access again -> no pin message sent the second time