test_tagValues.py 文件源码

def testMultipleSimultaneousUpdates(self):
        path = 'fluiddb/testing/test1'
        objectId = yield self.createObject()
        yield self.setTagValue(path, objectId, 5)
        deferreds = []
        try:
            for _ in range(10):
                for value in (7, True, None, 4.8, False, 'hi', ['aa', 'bb']):
                    deferreds.append(self.setTagValue(path, objectId, value))
            results = yield defer.DeferredList(deferreds, consumeErrors=True)
            failed = False
            for result in results:
                if not result[0]:
                    failed = True
                    log.err(result[1])
            if failed:
                self.fail()
        finally:
            yield self.deleteTagValue(path, objectId)