| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| # |
| |
| # setup, usage, teardown, errors(sync), errors(async), stress, soak, |
| # boundary-conditions, config |
| |
| from qpid.tests import Test |
| from qpid.framing import * |
| |
| class Base(Test): |
| |
| def cmp_frames(self, frm1, frm2): |
| assert frm1.flags == frm2.flags, "expected: %r, got %r" % (frm1, frm2) |
| assert frm1.type == frm2.type, "expected: %r, got %r" % (frm1, frm2) |
| assert frm1.track == frm2.track, "expected: %r, got %r" % (frm1, frm2) |
| assert frm1.channel == frm2.channel, "expected: %r, got %r" % (frm1, frm2) |
| assert frm1.payload == frm2.payload, "expected: %r, got %r" % (frm1, frm2) |
| |
| def cmp_segments(self, seg1, seg2): |
| assert seg1.first == seg2.first, "expected: %r, got %r" % (seg1, seg2) |
| assert seg1.last == seg2.last, "expected: %r, got %r" % (seg1, seg2) |
| assert seg1.type == seg2.type, "expected: %r, got %r" % (seg1, seg2) |
| assert seg1.track == seg2.track, "expected: %r, got %r" % (seg1, seg2) |
| assert seg1.channel == seg2.channel, "expected: %r, got %r" % (seg1, seg2) |
| assert seg1.payload == seg2.payload, "expected: %r, got %r" % (seg1, seg2) |
| |
| def cmp_list(self, l1, l2): |
| if l1 is None: |
| assert l2 is None |
| return |
| |
| assert len(l1) == len(l2) |
| for v1, v2 in zip(l1, l2): |
| if isinstance(v1, Compound): |
| self.cmp_ops(v1, v2) |
| else: |
| assert v1 == v2 |
| |
| def cmp_ops(self, op1, op2): |
| if op1 is None: |
| assert op2 is None |
| return |
| |
| assert op1.__class__ == op2.__class__ |
| cls = op1.__class__ |
| assert op1.NAME == op2.NAME |
| assert op1.CODE == op2.CODE |
| assert op1.FIELDS == op2.FIELDS |
| for f in cls.FIELDS: |
| v1 = getattr(op1, f.name) |
| v2 = getattr(op2, f.name) |
| if COMPOUND.has_key(f.type) or f.type == "struct32": |
| self.cmp_ops(v1, v2) |
| elif f.type in ("list", "array"): |
| self.cmp_list(v1, v2) |
| else: |
| assert v1 == v2, "expected: %r, got %r" % (v1, v2) |
| |
| if issubclass(cls, Command) or issubclass(cls, Control): |
| assert op1.channel == op2.channel |
| |
| if issubclass(cls, Command): |
| assert op1.sync == op2.sync, "expected: %r, got %r" % (op1.sync, op2.sync) |
| assert (op1.headers is None and op2.headers is None) or \ |
| (op1.headers is not None and op2.headers is not None) |
| if op1.headers is not None: |
| assert len(op1.headers) == len(op2.headers) |
| for h1, h2 in zip(op1.headers, op2.headers): |
| self.cmp_ops(h1, h2) |
| |
| class FrameTest(Base): |
| |
| def enc_dec(self, frames, encoded=None): |
| enc = FrameEncoder() |
| dec = FrameDecoder() |
| |
| enc.write(*frames) |
| bytes = enc.read() |
| if encoded is not None: |
| assert bytes == encoded, "expected %r, got %r" % (encoded, bytes) |
| dec.write(bytes) |
| dframes = dec.read() |
| |
| assert len(frames) == len(dframes) |
| for f, df, in zip(frames, dframes): |
| self.cmp_frames(f, df) |
| |
| def testEmpty(self): |
| self.enc_dec([Frame(0, 0, 0, 0, "")], |
| "\x00\x00\x00\x0c\x00\x00\x00\x00\x00\x00\x00\x00") |
| |
| def testSingle(self): |
| self.enc_dec([Frame(0, 0, 0, 1, "payload")], |
| "\x00\x00\x00\x13\x00\x00\x00\x01\x00\x00\x00\x00payload") |
| |
| def testMaxChannel(self): |
| self.enc_dec([Frame(0, 0, 0, 65535, "max-channel")], |
| "\x00\x00\x00\x17\x00\x00\xff\xff\x00\x00\x00\x00max-channel") |
| |
| def testMaxType(self): |
| self.enc_dec([Frame(0, 255, 0, 0, "max-type")], |
| "\x00\xff\x00\x14\x00\x00\x00\x00\x00\x00\x00\x00max-type") |
| |
| def testMaxTrack(self): |
| self.enc_dec([Frame(0, 0, 15, 0, "max-track")], |
| "\x00\x00\x00\x15\x00\x0f\x00\x00\x00\x00\x00\x00max-track") |
| |
| def testSequence(self): |
| self.enc_dec([Frame(0, 0, 0, 0, "zero"), |
| Frame(0, 0, 0, 1, "one"), |
| Frame(0, 0, 1, 0, "two"), |
| Frame(0, 0, 1, 1, "three"), |
| Frame(0, 1, 0, 0, "four"), |
| Frame(0, 1, 0, 1, "five"), |
| Frame(0, 1, 1, 0, "six"), |
| Frame(0, 1, 1, 1, "seven"), |
| Frame(1, 0, 0, 0, "eight"), |
| Frame(1, 0, 0, 1, "nine"), |
| Frame(1, 0, 1, 0, "ten"), |
| Frame(1, 0, 1, 1, "eleven"), |
| Frame(1, 1, 0, 0, "twelve"), |
| Frame(1, 1, 0, 1, "thirteen"), |
| Frame(1, 1, 1, 0, "fourteen"), |
| Frame(1, 1, 1, 1, "fifteen")]) |
| |
| class SegmentTest(Base): |
| |
| def enc_dec(self, segments, frames=None, interleave=None, max_payload=Frame.MAX_PAYLOAD): |
| enc = SegmentEncoder(max_payload) |
| dec = SegmentDecoder() |
| |
| enc.write(*segments) |
| frms = enc.read() |
| if frames is not None: |
| assert len(frames) == len(frms), "expected %s, got %s" % (frames, frms) |
| for f1, f2 in zip(frames, frms): |
| self.cmp_frames(f1, f2) |
| if interleave is not None: |
| ilvd = [] |
| for f in frms: |
| ilvd.append(f) |
| if interleave: |
| ilvd.append(interleave.pop(0)) |
| ilvd.extend(interleave) |
| dec.write(*ilvd) |
| else: |
| dec.write(*frms) |
| segs = dec.read() |
| assert len(segments) == len(segs) |
| for s1, s2 in zip(segments, segs): |
| self.cmp_segments(s1, s2) |
| |
| def testEmpty(self): |
| self.enc_dec([Segment(True, True, 0, 0, 0, "")], |
| [Frame(FIRST_FRM | LAST_FRM | FIRST_SEG | LAST_SEG, 0, 0, 0, |
| "")]) |
| |
| def testSingle(self): |
| self.enc_dec([Segment(True, True, 0, 0, 0, "payload")], |
| [Frame(FIRST_FRM | LAST_FRM | FIRST_SEG | LAST_SEG, 0, 0, 0, |
| "payload")]) |
| |
| def testMaxChannel(self): |
| self.enc_dec([Segment(False, False, 0, 0, 65535, "max-channel")], |
| [Frame(FIRST_FRM | LAST_FRM, 0, 0, 65535, "max-channel")]) |
| |
| def testMaxType(self): |
| self.enc_dec([Segment(False, False, 255, 0, 0, "max-type")], |
| [Frame(FIRST_FRM | LAST_FRM, 255, 0, 0, "max-type")]) |
| |
| def testMaxTrack(self): |
| self.enc_dec([Segment(False, False, 0, 15, 0, "max-track")], |
| [Frame(FIRST_FRM | LAST_FRM, 0, 15, 0, "max-track")]) |
| |
| def testSequence(self): |
| self.enc_dec([Segment(True, False, 0, 0, 0, "one"), |
| Segment(False, False, 0, 0, 0, "two"), |
| Segment(False, True, 0, 0, 0, "three")], |
| [Frame(FIRST_FRM | LAST_FRM | FIRST_SEG, 0, 0, 0, "one"), |
| Frame(FIRST_FRM | LAST_FRM, 0, 0, 0, "two"), |
| Frame(FIRST_FRM | LAST_FRM | LAST_SEG, 0, 0, 0, "three")]) |
| |
| def testInterleaveChannel(self): |
| frames = [Frame(0, 0, 0, 0, chr(ord("a") + i)) for i in range(7)] |
| frames[0].flags |= FIRST_FRM |
| frames[-1].flags |= LAST_FRM |
| |
| ilvd = [Frame(0, 0, 0, 1, chr(ord("a") + i)) for i in range(7)] |
| |
| self.enc_dec([Segment(False, False, 0, 0, 0, "abcdefg")], frames, ilvd, max_payload=1) |
| |
| def testInterleaveTrack(self): |
| frames = [Frame(0, 0, 0, 0, "%c%c" % (ord("a") + i, ord("a") + i + 1)) |
| for i in range(0, 8, 2)] |
| frames[0].flags |= FIRST_FRM |
| frames[-1].flags |= LAST_FRM |
| |
| ilvd = [Frame(0, 0, 1, 0, "%c%c" % (ord("a") + i, ord("a") + i + 1)) |
| for i in range(0, 8, 2)] |
| |
| self.enc_dec([Segment(False, False, 0, 0, 0, "abcdefgh")], frames, ilvd, max_payload=2) |
| |
| from qpid.ops import * |
| |
| class OpTest(Base): |
| |
| def enc_dec(self, ops): |
| enc = OpEncoder() |
| dec = OpDecoder() |
| enc.write(*ops) |
| segs = enc.read() |
| dec.write(*segs) |
| dops = dec.read() |
| assert len(ops) == len(dops) |
| for op1, op2 in zip(ops, dops): |
| self.cmp_ops(op1, op2) |
| |
| def testEmtpyMT(self): |
| self.enc_dec([MessageTransfer()]) |
| |
| def testEmptyMTSync(self): |
| self.enc_dec([MessageTransfer(sync=True)]) |
| |
| def testMT(self): |
| self.enc_dec([MessageTransfer(destination="asdf")]) |
| |
| def testSyncMT(self): |
| self.enc_dec([MessageTransfer(destination="asdf", sync=True)]) |
| |
| def testEmptyPayloadMT(self): |
| self.enc_dec([MessageTransfer(payload="")]) |
| |
| def testPayloadMT(self): |
| self.enc_dec([MessageTransfer(payload="test payload")]) |
| |
| def testHeadersEmptyPayloadMT(self): |
| self.enc_dec([MessageTransfer(headers=[DeliveryProperties()])]) |
| |
| def testHeadersPayloadMT(self): |
| self.enc_dec([MessageTransfer(headers=[DeliveryProperties()], payload="test payload")]) |
| |
| def testMultiHeadersEmptyPayloadMT(self): |
| self.enc_dec([MessageTransfer(headers=[DeliveryProperties(), MessageProperties()])]) |
| |
| def testMultiHeadersPayloadMT(self): |
| self.enc_dec([MessageTransfer(headers=[MessageProperties(), DeliveryProperties()], payload="test payload")]) |
| |
| def testContentTypeHeadersPayloadMT(self): |
| self.enc_dec([MessageTransfer(headers=[MessageProperties(content_type="text/plain")], payload="test payload")]) |
| |
| def testMulti(self): |
| self.enc_dec([MessageTransfer(), |
| MessageTransfer(sync=True), |
| MessageTransfer(destination="one"), |
| MessageTransfer(destination="two", sync=True), |
| MessageTransfer(destination="three", payload="test payload")]) |
| |
| def testControl(self): |
| self.enc_dec([SessionAttach(name="asdf")]) |
| |
| def testMixed(self): |
| self.enc_dec([SessionAttach(name="fdsa"), MessageTransfer(destination="test")]) |
| |
| def testChannel(self): |
| self.enc_dec([SessionAttach(name="asdf", channel=3), MessageTransfer(destination="test", channel=1)]) |
| |
| def testCompound(self): |
| self.enc_dec([MessageTransfer(headers=[MessageProperties(reply_to=ReplyTo(exchange="exch", routing_key="rk"))])]) |
| |
| def testListCompound(self): |
| self.enc_dec([ExecutionResult(value=RecoverResult(in_doubt=[Xid(global_id="one"), |
| Xid(global_id="two"), |
| Xid(global_id="three")]))]) |
