mirror of git://git.psyced.org/git/pypsyc
98 lines
2.8 KiB
Python
98 lines
2.8 KiB
Python
"""
|
|
:copyright: 2010 by Manuel Jacob
|
|
:license: MIT
|
|
"""
|
|
from nose.tools import assert_raises
|
|
from twisted.protocols.loopback import loopbackAsync as loopback_async
|
|
|
|
from pypsyc.core.mmp import Uni, Header, Circuit
|
|
|
|
|
|
UNI = 'psyc://server/foo/bar'
|
|
|
|
class TestUni(object):
|
|
def test_uni(self):
|
|
uni = Uni(UNI)
|
|
assert uni == UNI
|
|
assert uni.prefix == 'psyc'
|
|
|
|
def test_parts(self):
|
|
uni = Uni(UNI)
|
|
assert uni.into_parts() == ['server', 'foo', 'bar']
|
|
assert Uni.from_parts(uni.into_parts()) == uni
|
|
|
|
def test_descendant_ancestor(self):
|
|
uni = Uni(UNI)
|
|
assert uni.is_descendant_of('psyc://server/foo')
|
|
assert not uni.is_descendant_of('psyc://server/spam')
|
|
assert uni.is_ancestor_of('psyc://server/foo/bar/child')
|
|
assert not uni.is_ancestor_of('psyc://server/spam/eggs/child')
|
|
|
|
assert uni.is_descendant_of('psyc://server/')
|
|
assert Uni('psyc://server/').is_ancestor_of('psyc://server/foo/bar')
|
|
|
|
def test_chain(self):
|
|
assert Uni(UNI).chain('foobar') == 'psyc://server/foo/bar/foobar'
|
|
assert Uni('psyc://server/').chain('foo') == 'psyc://server/foo'
|
|
|
|
|
|
class TestHeader(object):
|
|
def test_empty(self):
|
|
header = Header()
|
|
assert header == {}
|
|
header._init()
|
|
assert header.source is None
|
|
assert header.target is None
|
|
assert header.context is None
|
|
|
|
def test_header(self):
|
|
header_dict = {
|
|
'_source': 'psyc://server/uni1',
|
|
'_target': 'psyc://server/uni2',
|
|
'_context': 'psyc://server/@place'
|
|
}
|
|
header = Header(header_dict)
|
|
assert header == header_dict
|
|
header._init()
|
|
assert header.source == Uni('psyc://server/uni1')
|
|
assert header.target == Uni('psyc://server/uni2')
|
|
assert header.context == Uni('psyc://server/@place')
|
|
|
|
|
|
class TestCircuit(object):
|
|
def setup(self):
|
|
self.success = False
|
|
|
|
def test_render_empty(self):
|
|
self._test({}, [])
|
|
|
|
def test_onlyvars(self):
|
|
self._test({'_target': 'psyc://example.org/'}, [])
|
|
|
|
def test_onlycontent(self):
|
|
self._test({}, ['content'])
|
|
|
|
def test_packet(self):
|
|
self._test({'_target': 'psyc://example.org/'}, ['content'])
|
|
|
|
def _test(self, header, content):
|
|
def received(header_, content_):
|
|
assert header_ == header
|
|
assert content_ == content
|
|
self.success = True
|
|
|
|
circuit1 = Circuit()
|
|
circuit1.packet_received = received
|
|
|
|
circuit2 = Circuit()
|
|
circuit2.initiator = True
|
|
circuit2.inited = lambda: circuit2.send(header, iter(content))
|
|
|
|
loopback_async(circuit1, circuit2)
|
|
assert self.success
|
|
|
|
def test_subclass(self):
|
|
circuit = Circuit()
|
|
circuit.lineReceived('|')
|
|
assert_raises(NotImplementedError, circuit.lineReceived, '|')
|