Index: third_party/typ/typ/fakes/test_result_server_fake.py |
diff --git a/third_party/typ/typ/fakes/test_result_server_fake.py b/third_party/typ/typ/fakes/test_result_server_fake.py |
new file mode 100644 |
index 0000000000000000000000000000000000000000..e7b81e92cc1177ff5882c86a4174f02f6ea3b123 |
--- /dev/null |
+++ b/third_party/typ/typ/fakes/test_result_server_fake.py |
@@ -0,0 +1,78 @@ |
+# Copyright 2014 Google Inc. All rights reserved. |
+# |
+# Licensed under the Apache License, Version 2.0 (the "License"); |
+# you may not use this file except in compliance with the License. |
+# You may obtain a copy of the License at |
+# |
+# http://www.apache.org/licenses/LICENSE-2.0 |
+# |
+# Unless required by applicable law or agreed to in writing, software |
+# distributed under the License is distributed on an "AS IS" BASIS, |
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
+# See the License for the specific language governing permissions and |
+# limitations under the License. |
+ |
+"""A fake implementation of test-results.appspot.com.""" |
+ |
+import io |
+import sys |
+import threading |
+ |
+ |
+if sys.version_info.major == 2: # pragma: python2 |
+ from SimpleHTTPServer import SimpleHTTPRequestHandler as HTTPRequestHandler |
+ from SocketServer import TCPServer |
+else: # pragma: python3 |
+ assert sys.version_info.major == 3 |
+ unicode = str # pylint: disable=W0622 |
+ from http.server import BaseHTTPRequestHandler # pylint: disable=F0401 |
+ HTTPRequestHandler = BaseHTTPRequestHandler |
+ from socketserver import TCPServer # pylint: disable=F0401 |
+ |
+ |
+def start(code=200): |
+ server = _Server(code=code) |
+ thread = threading.Thread(target=_run, args=(server,)) |
+ server.main_thread = thread |
+ thread.daemon = True |
+ thread.start() |
+ return server |
+ |
+ |
+def _run(server): |
+ server.serve_forever(0.05) |
+ |
+ |
+class _Server(TCPServer): |
+ |
+ def __init__(self, code): |
+ self.allow_reuse_address = True |
+ TCPServer.__init__(self, ('localhost', 0), _RequestHandler) |
+ self.log = io.StringIO() |
+ self.requests = [] |
+ self.main_thread = None |
+ self.code = code |
+ |
+ def stop(self): |
+ self.shutdown() |
+ self.main_thread.join() |
+ return self.requests |
+ |
+ |
+class _RequestHandler(HTTPRequestHandler): |
+ |
+ def __init__(self, *args, **kwargs): |
+ HTTPRequestHandler.__init__(self, *args, **kwargs) |
+ |
+ # 'Invalid Name' pylint: disable=C0103 |
+ def do_POST(self): |
+ path = self.path |
+ length = int(self.headers['content-length']) |
+ payload = self.rfile.read(length) |
+ self.server.requests.append(('post', path, payload)) |
+ self.send_response(self.server.code) |
+ self.end_headers() |
+ |
+ # 'Redefining built-in' pylint: disable=W0622 |
+ def log_message(self, format, *args): |
+ self.server.log.write(unicode('%s\n' % (format % args))) |