forum_modules/oauthauth/lib/oauth2/httplib2/test/functional/test_proxies.py

   1 import unittest
   2 import errno
   3 import os
   4 import signal
   5 import subprocess
   6 import tempfile
   7
   8 import nose
   9
  10 import httplib2
  11 from httplib2 import socks
  12 from httplib2.test import miniserver
  13
  14 tinyproxy_cfg = """
  15 User "%(user)s"
  16 Port %(port)s
  17 Listen 127.0.0.1
  18 PidFile "%(pidfile)s"
  19 LogFile "%(logfile)s"
  20 MaxClients 2
  21 StartServers 1
  22 LogLevel Info
  23 """
  24
  25
  26 class FunctionalProxyHttpTest(unittest.TestCase):
  27     def setUp(self):
  28         if not socks:
  29             raise nose.SkipTest('socks module unavailable')
  30         if not subprocess:
  31             raise nose.SkipTest('subprocess module unavailable')
  32
  33         # start a short-lived miniserver so we can get a likely port
  34         # for the proxy
  35         self.httpd, self.proxyport = miniserver.start_server(
  36             miniserver.ThisDirHandler)
  37         self.httpd.shutdown()
  38         self.httpd, self.port = miniserver.start_server(
  39             miniserver.ThisDirHandler)
  40
  41         self.pidfile = tempfile.mktemp()
  42         self.logfile = tempfile.mktemp()
  43         fd, self.conffile = tempfile.mkstemp()
  44         f = os.fdopen(fd, 'w')
  45         our_cfg = tinyproxy_cfg % {'user': os.getlogin(),
  46                                    'pidfile': self.pidfile,
  47                                    'port': self.proxyport,
  48                                    'logfile': self.logfile}
  49         f.write(our_cfg)
  50         f.close()
  51         try:
  52             # TODO use subprocess.check_call when 2.4 is dropped
  53             ret = subprocess.call(['tinyproxy', '-c', self.conffile])
  54             self.assertEqual(0, ret)
  55         except OSError, e:
  56             if e.errno == errno.ENOENT:
  57                 raise nose.SkipTest('tinyproxy not available')
  58             raise
  59
  60     def tearDown(self):
  61         self.httpd.shutdown()
  62         try:
  63             pid = int(open(self.pidfile).read())
  64             os.kill(pid, signal.SIGTERM)
  65         except OSError, e:
  66             if e.errno == errno.ESRCH:
  67                 print '\n\n\nTinyProxy Failed to start, log follows:'
  68                 print open(self.logfile).read()
  69                 print 'end tinyproxy log\n\n\n'
  70             raise
  71         map(os.unlink, (self.pidfile,
  72                         self.logfile,
  73                         self.conffile))
  74
  75     def testSimpleProxy(self):
  76         proxy_info = httplib2.ProxyInfo(socks.PROXY_TYPE_HTTP,
  77                                         'localhost', self.proxyport)
  78         client = httplib2.Http(proxy_info=proxy_info)
  79         src = 'miniserver.py'
  80         response, body = client.request('http://localhost:%d/%s' %
  81                                         (self.port, src))
  82         self.assertEqual(response.status, 200)
  83         self.assertEqual(body, open(os.path.join(miniserver.HERE, src)).read())
  84         lf = open(self.logfile).read()
  85         expect = ('Established connection to host "127.0.0.1" '
  86                   'using file descriptor')
  87         self.assertTrue(expect in lf,
  88                         'tinyproxy did not proxy a request for miniserver')