1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
# -*- test-case-name: tubes.test.test_routing -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Tests for L{tubes.routing}.
"""
from unittest import TestCase
from tubes.routing import Router, to, Routed
from tubes.tube import series, receiver
from .util import FakeFount, FakeDrain, IFakeOutput, IFakeInput, FakeInput
if 0:
# Names used by PyDoctor.
from tubes.itube import IFount
IFount
class RouterTests(TestCase):
"""
Tests for L{Router}.
"""
def test_twoRoutes(self):
"""
The L{IFount} feeding into a L{Router} may yield L{to} each route
returned from L{Router.newRoute}.
"""
@receiver()
def chooser(item):
if item % 2:
yield to(odd, item)
else:
yield to(even, item)
router = Router()
even = router.newRoute("even")
evens = FakeDrain()
even.flowTo(evens)
odd = router.newRoute("odd")
odds = FakeDrain()
odd.flowTo(odds)
ff = FakeFount()
routeDrain = series(chooser, router.drain)
ff.flowTo(routeDrain)
for x in range(10):
ff.drain.receive(x)
self.assertEqual(odds.received, [1, 3, 5, 7, 9])
self.assertEqual(evens.received, [0, 2, 4, 6, 8])
def test_routeRepr(self):
"""
It's useful to C{repr} a route for debugging purposes; if we give it a
name, its C{repr} will contain that name.
"""
router = Router()
route = router.newRoute("hello")
self.assertTrue("hello" in repr(route))
def test_defaultTypeChecking(self):
"""
L{Router}'s drain accepts only L{Routed} objects; if no other type is
specified, L{Routed}C{(None)}.
"""
router = Router()
ff = FakeFount(IFakeOutput)
self.assertEqual(router.drain.inputType, Routed(None))
self.assertRaises(TypeError, ff.flowTo, router.drain)
self.assertEqual(router.newRoute().outputType, None)
def test_specifiedTypeChecking(self):
"""
The C{outputType} argument to L{Router}'s constructor specifies the
type of output that its routes will provide, and also the routed type
required as an input.
"""
router = Router(IFakeInput)
incorrect = FakeDrain(IFakeOutput)
correct = FakeDrain(IFakeInput)
self.assertEqual(router.drain.inputType, Routed(IFakeInput))
self.assertEqual(router.newRoute().outputType, IFakeInput)
self.assertRaises(TypeError, router.newRoute().flowTo, incorrect)
self.assertEqual(router.newRoute().flowTo(correct), None)
correctFount = FakeFount(Routed(IFakeInput))
incorrectFount = FakeFount(Routed(IFakeOutput))
self.assertRaises(TypeError, incorrectFount.flowTo, router.drain)
self.assertEqual(None, correctFount.flowTo(router.drain))
class RoutedTests(TestCase):
"""
Tests for L{Routed}.
"""
def test_eq(self):
"""
C{==} on L{Routed} is L{True} for equivalent ones, L{False} otherwise.
"""
self.assertEqual(True, Routed(IFakeInput) == Routed(IFakeInput))
self.assertEqual(False, Routed(IFakeInput) == Routed(IFakeOutput))
self.assertEqual(False, Routed() == 7)
def test_ne(self):
"""
C{==} on L{Routed} is L{False} for equivalent ones, L{True} otherwise.
"""
self.assertEqual(False, Routed(IFakeInput) != Routed(IFakeInput))
self.assertEqual(True, Routed(IFakeInput) != Routed(IFakeOutput))
self.assertEqual(True, Routed() != 7)
def test_providedBy(self):
"""
L{Routed.providedBy} ensures that the given object is a L{to} and that
its payload provides the proper specification.
"""
router = Router()
route = router.newRoute()
self.assertEqual(False,
Routed(IFakeInput).providedBy(FakeInput()))
self.assertEqual(False,
Routed(IFakeInput).providedBy(to(route, object())))
self.assertEqual(True,
Routed(IFakeInput).providedBy(to(route, FakeInput())))
def test_providedByNone(self):
"""
L{Routed.providedBy} ensures that the given object is L{to} but makes
no assertions about its payload if given L{Routed} is given no
sub-specification.
"""
router = Router()
route = router.newRoute()
self.assertEqual(False, Routed().providedBy(object()))
self.assertEqual(True, Routed().providedBy(to(route, object())))
|
