Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(21)

Unified Diff: bin/au_test_harness/cros_test_proxy.py

Issue 6717011: Remove au_test_harness code and change symlinks to point to new location (Closed) Base URL: http://git.chromium.org/git/crosutils.git@master
Patch Set: Created 9 years, 9 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
« no previous file with comments | « bin/au_test_harness/cros_au_test_harness.py ('k') | bin/au_test_harness/dev_server_wrapper.py » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
Index: bin/au_test_harness/cros_test_proxy.py
diff --git a/bin/au_test_harness/cros_test_proxy.py b/bin/au_test_harness/cros_test_proxy.py
deleted file mode 100644
index 0b3728638bf0c4a835562c341482100035400d73..0000000000000000000000000000000000000000
--- a/bin/au_test_harness/cros_test_proxy.py
+++ /dev/null
@@ -1,121 +0,0 @@
-# Copyright (c) 2010 The Chromium OS Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-"""Module containing various classes pertaining to inserting a proxy in a test.
-"""
-
-import os
-import select
-import socket
-import SocketServer
-import threading
-
-class Filter(object):
- """Base class for data filters.
-
- Pass subclass of this to CrosTestProxy which will perform whatever
- connection manipulation you prefer.
- """
-
- def setup(self):
- """This setup method is called once per connection."""
- pass
-
- def InBound(self, data):
- """This method is called once per packet of incoming data.
-
- The value returned is what is sent through the proxy. If
- None is returned, the connection will be closed.
- """
- return data
-
- def OutBound(self, data):
- """This method is called once per packet of outgoing data.
-
- The value returned is what is sent through the proxy. If
- None is returned, the connection will be closed.
- """
- return data
-
-
-class CrosTestProxy(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
- """A transparent proxy for simulating network errors"""
-
- class _Handler(SocketServer.BaseRequestHandler):
- """Proxy connection handler that passes data though a filter"""
-
- def setup(self):
- """Setup is called once for each connection proxied."""
- self.server.filter.setup()
-
- def handle(self):
- """Handles each incoming connection.
-
- Opens a new connection to the port we are proxing to, then
- passes each packet along in both directions after passing
- them through the filter object passed in.
- """
- # Open outgoing socket
- s_in = self.request
- s_out = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- s_out.connect((self.server.address_out, self.server.port_out))
-
- while True:
- rlist, wlist, xlist = select.select([s_in, s_out], [], [])
-
- if s_in in rlist:
- data = s_in.recv(1024)
- data = self.server.filter.InBound(data)
- if not data: break
- try:
- # If there is any error sending data, close both connections.
- s_out.sendall(data)
- except socket.error:
- break
-
- if s_out in rlist:
- data = s_out.recv(1024)
- data = self.server.filter.OutBound(data)
- if not data: break
- try:
- # If there is any error sending data, close both connections.
- s_in.sendall(data)
- except socket.error:
- break
-
- s_in.close()
- s_out.close()
-
- def __init__(self,
- filter,
- port_in=8081,
- address_out='127.0.0.1', port_out=8080):
- """Configures the proxy object.
-
- Args:
- filter: An instance of a subclass of Filter.
- port_in: Port on which to listen for incoming connections.
- address_out: Address to which outgoing connections will go.
- address_port: Port to which outgoing connections will go.
- """
- self.port_in = port_in
- self.address_out = address_out
- self.port_out = port_out
- self.filter = filter
-
- try:
- SocketServer.TCPServer.__init__(self,
- ('', port_in),
- self._Handler)
- except socket.error:
- os.system('sudo netstat -l --tcp -n -p')
- raise
-
- def serve_forever_in_thread(self):
- """Helper method to start the server in a new background thread."""
- server_thread = threading.Thread(target=self.serve_forever)
- server_thread.setDaemon(True)
- server_thread.start()
-
- return server_thread
« no previous file with comments | « bin/au_test_harness/cros_au_test_harness.py ('k') | bin/au_test_harness/dev_server_wrapper.py » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698