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