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
|
# -*- test-case-name: twisted.internet.test.test_endpoints -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Fake client and server endpoint string parser plugins for testing purposes.
"""
from zope.interface.declarations import implements
from twisted.plugin import IPlugin
from twisted.internet.interfaces import (IStreamClientEndpoint,
IStreamServerEndpoint,
IStreamClientEndpointStringParser,
IStreamServerEndpointStringParser)
class PluginBase(object):
implements(IPlugin)
def __init__(self, pfx):
self.prefix = pfx
class FakeClientParser(PluginBase):
implements(IStreamClientEndpointStringParser)
def parseStreamClient(self, *a, **kw):
return StreamClient(self, a, kw)
class FakeParser(PluginBase):
implements(IStreamServerEndpointStringParser)
def parseStreamServer(self, *a, **kw):
return StreamServer(self, a, kw)
class EndpointBase(object):
def __init__(self, parser, args, kwargs):
self.parser = parser
self.args = args
self.kwargs = kwargs
class StreamClient(EndpointBase):
implements(IStreamClientEndpoint)
class StreamServer(EndpointBase):
implements(IStreamServerEndpoint)
# Instantiate plugin interface providers to register them.
fake = FakeParser('fake')
fakeClient = FakeClientParser('cfake')
|
