+class ClientSeqnums(unittest.TestCase):
+ def test_client(self):
+ basedir = "introducer/ClientSeqnums/test_client"
+ fileutil.make_dirs(basedir)
+ f = open(os.path.join(basedir, "tahoe.cfg"), "w")
+ f.write("[client]\n")
+ f.write("introducer.furl = nope\n")
+ f.close()
+ c = TahoeClient(basedir)
+ ic = c.introducer_client
+ outbound = ic._outbound_announcements
+ published = ic._published_announcements
+ def read_seqnum():
+ f = open(os.path.join(basedir, "announcement-seqnum"))
+ seqnum = f.read().strip()
+ f.close()
+ return int(seqnum)
+
+ ic.publish("sA", {"key": "value1"}, c._server_key)
+ self.failUnlessEqual(read_seqnum(), 1)
+ self.failUnless("sA" in outbound)
+ self.failUnlessEqual(outbound["sA"]["seqnum"], 1)
+ nonce1 = outbound["sA"]["nonce"]
+ self.failUnless(isinstance(nonce1, str))
+ self.failUnlessEqual(simplejson.loads(published["sA"][0]),
+ outbound["sA"])
+ # [1] is the signature, [2] is the pubkey
+
+ # publishing a second service causes both services to be
+ # re-published, with the next higher sequence number
+ ic.publish("sB", {"key": "value2"}, c._server_key)
+ self.failUnlessEqual(read_seqnum(), 2)
+ self.failUnless("sB" in outbound)
+ self.failUnlessEqual(outbound["sB"]["seqnum"], 2)
+ self.failUnless("sA" in outbound)
+ self.failUnlessEqual(outbound["sA"]["seqnum"], 2)
+ nonce2 = outbound["sA"]["nonce"]
+ self.failUnless(isinstance(nonce2, str))
+ self.failIfEqual(nonce1, nonce2)
+ self.failUnlessEqual(simplejson.loads(published["sA"][0]),
+ outbound["sA"])
+ self.failUnlessEqual(simplejson.loads(published["sB"][0]),
+ outbound["sB"])
+
+