4.1 server tests: add delegation_want tests

index 895916a..c64a7ef 100644
@@ -141,7 +141,7 @@ def testReadDeleg(t, env):
     check(res)
     fh = res.resarray[-1].object
     deleg = res.resarray[-2].delegation
-    if deleg.delegation_type == OPEN_DELEGATE_NONE:
+    if deleg.delegation_type == OPEN_DELEGATE_NONE or deleg.delegation_type == OPEN_DELEGATE_NONE_EXT:
         fail("Could not get delegation")
     # c2 - OPEN - WRITE
     sess2 = env.c1.new_client_session("%s_2" % env.testname(t))

index 2bedb57..9d260a4 100644
@@ -9,6 +9,7 @@ __all__ = ["st_exchange_id.py", # draft 21
            "st_sequence.py",
 	   "st_trunking.py",
            "st_open.py",
+           "st_delegation.py",
            "st_verify.py",
            "st_getdevicelist.py",
            "st_lookupp.py",

54 lines not on this page · raw message

51 lines not on this page · raw message

index b9f866f..1011cbc 100644
@@ -52,3 +52,48 @@ def testReadDeleg(t, env):
     # Now get OPEN reply
     res = sess2.listen(slot)
     checklist(res, [NFS4_OK, NFS4ERR_DELAY])
+
+def testWriteDeleg(t, env):
+    """Test write delgation handout and return
+
+    FLAGS: open deleg all
+    CODE: DELEG2
+    """
+    recall = threading.Event()
+    def pre_hook(arg, env):
+        recall.stateid = arg.stateid # NOTE this must be done before set()
+        env.notify = recall.set # This is called after compound sent to queue
+    def post_hook(arg, env, res):
+        return res
+    # c1 - OPEN - WRITE
+    c1 = env.c1.new_client("%s_1" % env.testname(t))
+    c1.cb_pre_hook(OP_CB_RECALL, pre_hook)
+    c1.cb_post_hook(OP_CB_RECALL, post_hook)
+    sess1 = c1.create_session()
+    sess1.compound([op.reclaim_complete(FALSE)])
+    res = create_file(sess1, env.testname(t),
+                      access=OPEN4_SHARE_ACCESS_BOTH |
+                      OPEN4_SHARE_ACCESS_WANT_WRITE_DELEG)
+    check(res)
+    fh = res.resarray[-1].object
+    deleg = res.resarray[-2].delegation
+    if deleg.delegation_type == OPEN_DELEGATE_NONE or deleg.delegation_type == OPEN_DELEGATE_NONE_EXT:
+        fail("Could not get delegation")
+    # c2 - OPEN - READ
+    sess2 = env.c1.new_client_session("%s_2" % env.testname(t))
+    claim = open_claim4(CLAIM_NULL, env.testname(t))
+    owner = open_owner4(0, "My Open Owner 2")
+    how = openflag4(OPEN4_NOCREATE)
+    open_op = op.open(0, OPEN4_SHARE_ACCESS_READ, OPEN4_SHARE_DENY_NONE,
+                      owner, how, claim)
+    slot = sess2.compound_async(env.home + [open_op])
+    # Wait for recall, and return delegation
+    recall.wait() # STUB - deal with timeout
+    # Getting here means CB_RECALL reply is in the send queue.
+    # Give it a moment to actually be sent
+    env.sleep(1)
+    res = sess1.compound([op.putfh(fh), op.delegreturn(recall.stateid)])
+    check(res)
+    # Now get OPEN reply
+    res = sess2.listen(slot)
+    checklist(res, [NFS4_OK, NFS4ERR_DELAY])

index 1011cbc..3b32c74 100644
@@ -97,3 +97,48 @@ def testWriteDeleg(t, env):
     # Now get OPEN reply
     res = sess2.listen(slot)
     checklist(res, [NFS4_OK, NFS4ERR_DELAY])
+
+def testAnyDeleg(t, env):
+    """Test any delgation handout and return
+
+    FLAGS: open deleg all
+    CODE: DELEG3
+    """
+    recall = threading.Event()
+    def pre_hook(arg, env):
+        recall.stateid = arg.stateid # NOTE this must be done before set()
+        env.notify = recall.set # This is called after compound sent to queue
+    def post_hook(arg, env, res):
+        return res
+    # c1 - OPEN - READ
+    c1 = env.c1.new_client("%s_1" % env.testname(t))
+    c1.cb_pre_hook(OP_CB_RECALL, pre_hook)
+    c1.cb_post_hook(OP_CB_RECALL, post_hook)
+    sess1 = c1.create_session()
+    sess1.compound([op.reclaim_complete(FALSE)])
+    res = create_file(sess1, env.testname(t),
+                      access=OPEN4_SHARE_ACCESS_READ |
+                      OPEN4_SHARE_ACCESS_WANT_ANY_DELEG)
+    check(res)
+    fh = res.resarray[-1].object
+    deleg = res.resarray[-2].delegation
+    if deleg.delegation_type == OPEN_DELEGATE_NONE or deleg.delegation_type == OPEN_DELEGATE_NONE_EXT:
+        fail("Could not get delegation")
+    # c2 - OPEN - WRITE
+    sess2 = env.c1.new_client_session("%s_2" % env.testname(t))
+    claim = open_claim4(CLAIM_NULL, env.testname(t))
+    owner = open_owner4(0, "My Open Owner 2")
+    how = openflag4(OPEN4_NOCREATE)
+    open_op = op.open(0, OPEN4_SHARE_ACCESS_BOTH, OPEN4_SHARE_DENY_NONE,
+                      owner, how, claim)
+    slot = sess2.compound_async(env.home + [open_op])
+    # Wait for recall, and return delegation
+    recall.wait() # STUB - deal with timeout
+    # Getting here means CB_RECALL reply is in the send queue.
+    # Give it a moment to actually be sent
+    env.sleep(1)
+    res = sess1.compound([op.putfh(fh), op.delegreturn(recall.stateid)])
+    check(res)
+    # Now get OPEN reply
+    res = sess2.listen(slot)
+    checklist(res, [NFS4_OK, NFS4ERR_DELAY])

index 3b32c74..ac425dc 100644
@@ -142,3 +142,25 @@ def testAnyDeleg(t, env):
     # Now get OPEN reply
     res = sess2.listen(slot)
     checklist(res, [NFS4_OK, NFS4ERR_DELAY])
+
+def testNoDeleg(t, env):
+    """Test no delgation handout
+
+    FLAGS: open deleg all
+    CODE: DELEG4
+    """
+    c1 = env.c1.new_client("%s_1" % env.testname(t))
+    sess1 = c1.create_session()
+    sess1.compound([op.reclaim_complete(FALSE)])
+    res = create_file(sess1, env.testname(t),
+                      access=OPEN4_SHARE_ACCESS_READ |
+                      OPEN4_SHARE_ACCESS_WANT_NO_DELEG)
+    check(res)
+    fh = res.resarray[-1].object
+    deleg = res.resarray[-2].delegation
+    if deleg.delegation_type == OPEN_DELEGATE_NONE:
+        fail("Got no delegation, expected OPEN_DELEGATE_NONE_EXT")
+    if deleg.delegation_type != OPEN_DELEGATE_NONE_EXT:
+        fail("Got a delegation (type "+str(deleg.delegation_type)+") despite asking for none")
+    if deleg.ond_why != WND4_NOT_WANTED:
+        fail("Wrong reason ("+str(deleg.ond_why)+") for giving no delegation")