"Fossies" - the Fresh Open Source Software Archive 
Member "zorp-7.0.4/tests/python/test_session.py" (28 Oct 2019, 4960 Bytes) of package /linux/privat/zorp-7.0.4.tar.gz:
As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style:
standard) with prefixed line numbers.
Alternatively you can here
view or
download the uninterpreted source code file.
See also the last
Fossies "Diffs" side-by-side code changes report for "test_session.py":
6.0.11_vs_7.0.1.
1 # vim: ts=8 sts=4 expandtab autoindent
2
3 ############################################################################
4 ##
5 ## Copyright (c) 2000-2015 BalaBit IT Ltd, Budapest, Hungary
6 ## Copyright (c) 2015-2018 BalaSys IT Ltd, Budapest, Hungary
7 ##
8 ##
9 ## This program is free software; you can redistribute it and/or modify
10 ## it under the terms of the GNU General Public License as published by
11 ## the Free Software Foundation; either version 2 of the License, or
12 ## (at your option) any later version.
13 ##
14 ## This program is distributed in the hope that it will be useful,
15 ## but WITHOUT ANY WARRANTY; without even the implied warranty of
16 ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 ## GNU General Public License for more details.
18 ##
19 ## You should have received a copy of the GNU General Public License along
20 ## with this program; if not, write to the Free Software Foundation, Inc.,
21 ## 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
22 ##
23 ############################################################################
24
25 from Zorp.Core import *
26 from Zorp.Zorp import quit
27 from Zorp.Proxy import Proxy
28 from Zorp.Http import HttpProxy
29 from Zorp.Session import MasterSession, StackedSession
30 import Zorp.Globals
31 import Zorp.Resolver
32 import socket
33 import unittest
34
35 config.options.kzorp_enabled = FALSE
36
37 InetZone("internet", "0.0.0.0/0")
38
39
40 class DummyResolver(Zorp.Resolver.AbstractResolver):
41
42 def resolve(self, host, port):
43 return SockAddrInet("127.0.0.1", port)
44
45 ResolverPolicy("test_resolver", DummyResolver())
46
47
48 class TestSessionBase(unittest.TestCase):
49
50 def setUp(self):
51 self.service = Service("testservice", HttpProxy, resolver_policy="test_resolver")
52 self.address = SockAddrInet("127.0.0.1", 80)
53 self.mastersession = MasterSession(self.service, client_stream=None, client_local=None, client_listen=DBSockAddr(self.address, ZD_PROTO_TCP), client_address=self.address, instance_id=1)
54 self.session = StackedSession(self.mastersession)
55
56 def tearDown(self):
57 del self.session
58 del self.mastersession
59 del self.service
60 del self.address
61 Zorp.Globals.services.clear()
62
63
64 class TestTopLevelProxySession(TestSessionBase):
65
66 def test_construction(self):
67 self.assertTrue(isinstance(self.mastersession, MasterSession))
68 self.assertTrue(isinstance(self.session, StackedSession))
69
70 def test_construction_initializes_protocol(self):
71 master = MasterSession(self.service, None, None, DBSockAddr(SockAddrInet('127.0.0.1', 1234), ZD_PROTO_TCP), None, instance_id=1)
72 self.assertEqual(master.protocol, ZD_PROTO_TCP)
73 self.assertEqual(master.protocol_name, "TCP")
74
75 def test_client_zone_lookup(self):
76 self.assertEqual(self.session.client_address, self.address)
77 self.assertEqual(self.session.client_zone.name, "internet")
78
79 def test_setServerAddress(self):
80 self.session.setServerAddress(self.address)
81 self.assertEqual(self.session.server_address, self.address)
82 self.assertEqual(self.session.server_zone.name, "internet")
83
84 def test_setTargetAddress(self):
85 self.session.setTargetAddress(self.address)
86 self.assertEqual(self.session.target_address, (self.address, ))
87 self.assertEqual(len(self.session.target_zone), 1)
88 self.assertEqual(self.session.target_zone[0].name, "internet")
89
90 def test_setProxy(self):
91 """Check the final session ID constructed when an associated proxy was created and the proxy is properly set in the session"""
92 (client_stream, server_stream) = streamPair(socket.AF_UNIX, socket.SOCK_STREAM)
93 self.session.client_stream = client_stream
94 proxy = HttpProxy(self.session)
95 self.assertEqual(self.session.session_id, "svc//" + self.service.name + ":1" + "//http")
96 self.assertEqual(self.session.proxy, proxy)
97 self.assertEqual(self.session.http, proxy)
98 self.assertEqual(proxy.session, self.session)
99
100
101 class TestStackedProxySession(TestSessionBase):
102
103 def test_stacking_inherit_target_address(self):
104 """Stacked proxies should inherit the target address of their parent"""
105 self.session.setTargetAddress(self.address)
106 ss = StackedSession(self.session)
107 self.assertEqual(ss.target_address, (self.address, ))
108
109 def test_stacking_override_target_address(self):
110 """Stacked proxies can override their target own address without affecting their parent session"""
111 self.session.setTargetAddress(self.address)
112 ss = StackedSession(self.session)
113 address2 = SockAddrInet("10.0.0.1", 90)
114 ss.setTargetAddress(address2)
115 self.assertEqual(ss.target_address, (address2, ))
116 self.assertEqual(self.session.target_address, (self.address, ))
117
118
119 def init(name, virtual_name, is_master):
120 unittest.main(argv=('/',))
121
122 # Local Variables:
123 # mode: python
124 # indent-tabs-mode: nil
125 # python-indent: 4
126 # End: