1# -*- coding: utf-8 -*- 2import hyper.common.connection 3 4from hyper.common.connection import HTTPConnection 5from hyper.common.exceptions import TLSUpgrade, HTTPUpgrade 6 7 8class TestHTTPConnection(object): 9 def test_h1_kwargs(self): 10 c = HTTPConnection( 11 'test', 443, secure=False, window_manager=True, enable_push=True, 12 ssl_context=False, proxy_host=False, proxy_port=False, 13 proxy_headers=False, other_kwarg=True, timeout=5 14 ) 15 16 assert c._h1_kwargs == { 17 'secure': False, 18 'ssl_context': False, 19 'proxy_host': False, 20 'proxy_port': False, 21 'proxy_headers': False, 22 'other_kwarg': True, 23 'enable_push': True, 24 'timeout': 5, 25 } 26 27 def test_h2_kwargs(self): 28 c = HTTPConnection( 29 'test', 443, secure=False, window_manager=True, enable_push=True, 30 ssl_context=True, proxy_host=False, proxy_port=False, 31 proxy_headers=False, other_kwarg=True, timeout=(10, 30) 32 ) 33 34 assert c._h2_kwargs == { 35 'window_manager': True, 36 'enable_push': True, 37 'secure': False, 38 'ssl_context': True, 39 'proxy_host': False, 40 'proxy_port': False, 41 'proxy_headers': False, 42 'other_kwarg': True, 43 'timeout': (10, 30), 44 } 45 46 def test_tls_upgrade(self, monkeypatch): 47 monkeypatch.setattr( 48 hyper.common.connection, 'HTTP11Connection', DummyH1Connection 49 ) 50 monkeypatch.setattr( 51 hyper.common.connection, 'HTTP20Connection', DummyH2Connection 52 ) 53 c = HTTPConnection('test', 443) 54 55 assert isinstance(c._conn, DummyH1Connection) 56 57 r = c.request('GET', '/') 58 59 assert r == 'h2' 60 assert isinstance(c._conn, DummyH2Connection) 61 assert c._conn._sock == 'totally a secure socket' 62 63 def test_http_upgrade(self, monkeypatch): 64 monkeypatch.setattr( 65 hyper.common.connection, 'HTTP11Connection', DummyH1Connection 66 ) 67 monkeypatch.setattr( 68 hyper.common.connection, 'HTTP20Connection', DummyH2Connection 69 ) 70 c = HTTPConnection('test', 80) 71 72 assert isinstance(c._conn, DummyH1Connection) 73 74 c.request('GET', '/') 75 resp = c.get_response() 76 77 assert resp == 'h2c' 78 assert isinstance(c._conn, DummyH2Connection) 79 assert c._conn._sock == 'totally a non-secure socket' 80 81 82class DummyH1Connection(object): 83 def __init__(self, host, port=None, secure=None, **kwargs): 84 self.host = host 85 self.port = port 86 87 if secure is not None: 88 self.secure = secure 89 elif self.port == 443: 90 self.secure = True 91 else: 92 self.secure = False 93 94 def request(self, *args, **kwargs): 95 if self.secure: 96 raise TLSUpgrade('h2', 'totally a secure socket') 97 98 def get_response(self): 99 if not self.secure: 100 raise HTTPUpgrade('h2c', 'totally a non-secure socket') 101 102 103class DummyH2Connection(object): 104 def __init__(self, host, port=None, secure=None, **kwargs): 105 self.host = host 106 self.port = port 107 108 if secure is not None: 109 self.secure = secure 110 elif self.port == 443: 111 self.secure = True 112 else: 113 self.secure = False 114 115 def _send_preamble(self): 116 pass 117 118 def _connect_upgrade(self, sock): 119 self._sock = sock 120 121 def _new_stream(self, *args, **kwargs): 122 pass 123 124 def request(self, *args, **kwargs): 125 if self.secure: 126 return 'h2' 127 128 def get_response(self, *args, **kwargs): 129 if not self.secure: 130 return 'h2c' 131