File: test_client_connection.py

package info (click to toggle)
python-aiohttp 3.5.1-1
  • links: PTS, VCS
  • area: main
  • in suites: bullseye, buster, sid
  • size: 5,612 kB
  • sloc: python: 36,917; ansic: 15,734; makefile: 365; sh: 83
file content (135 lines) | stat: -rw-r--r-- 3,420 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
import gc
from unittest import mock

import pytest

from aiohttp.connector import Connection


@pytest.fixture
def key():
    return object()


@pytest.fixture
def loop():
    return mock.Mock()


@pytest.fixture
def connector():
    return mock.Mock()


@pytest.fixture
def protocol():
    return mock.Mock(should_close=False)


def test_ctor(connector, key, protocol, loop) -> None:
    conn = Connection(connector, key, protocol, loop)
    with pytest.warns(DeprecationWarning):
        assert conn.loop is loop
    assert conn.protocol is protocol
    conn.close()


def test_callbacks_on_close(connector, key, protocol, loop) -> None:
    conn = Connection(connector, key, protocol, loop)
    notified = False

    def cb():
        nonlocal notified
        notified = True

    conn.add_callback(cb)
    conn.close()
    assert notified


def test_callbacks_on_release(connector, key, protocol, loop) -> None:
    conn = Connection(connector, key, protocol, loop)
    notified = False

    def cb():
        nonlocal notified
        notified = True

    conn.add_callback(cb)
    conn.release()
    assert notified


def test_callbacks_exception(connector, key, protocol, loop) -> None:
    conn = Connection(connector, key, protocol, loop)
    notified = False

    def cb1():
        raise Exception

    def cb2():
        nonlocal notified
        notified = True

    conn.add_callback(cb1)
    conn.add_callback(cb2)
    conn.close()
    assert notified


def test_del(connector, key, protocol, loop) -> None:
    loop.is_closed.return_value = False
    conn = Connection(connector, key, protocol, loop)
    exc_handler = mock.Mock()
    loop.set_exception_handler(exc_handler)

    with pytest.warns(ResourceWarning):
        del conn
        gc.collect()

    connector._release.assert_called_with(key, protocol, should_close=True)
    msg = {'client_connection': mock.ANY,  # conn was deleted
           'message': 'Unclosed connection'}
    if loop.get_debug():
        msg['source_traceback'] = mock.ANY
    loop.call_exception_handler.assert_called_with(msg)


def test_close(connector, key, protocol, loop) -> None:
    conn = Connection(connector, key, protocol, loop)
    assert not conn.closed
    conn.close()
    assert conn._protocol is None
    connector._release.assert_called_with(key, protocol, should_close=True)
    assert conn.closed


def test_release(connector, key, protocol, loop) -> None:
    conn = Connection(connector, key, protocol, loop)
    assert not conn.closed
    conn.release()
    assert not protocol.transport.close.called
    assert conn._protocol is None
    connector._release.assert_called_with(key, protocol, should_close=False)
    assert conn.closed


def test_release_proto_should_close(connector, key, protocol, loop) -> None:
    protocol.should_close = True
    conn = Connection(connector, key, protocol, loop)
    assert not conn.closed
    conn.release()
    assert not protocol.transport.close.called
    assert conn._protocol is None
    connector._release.assert_called_with(key, protocol, should_close=True)
    assert conn.closed


def test_release_released(connector, key, protocol, loop) -> None:
    conn = Connection(connector, key, protocol, loop)
    conn.release()
    connector._release.reset_mock()
    conn.release()
    assert not protocol.transport.close.called
    assert conn._protocol is None
    assert not connector._release.called