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
|
#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vi:ts=4:et
from . import localhost
import pycurl
import pytest
import sys
import unittest
from . import appmanager
from . import util
setup_module, teardown_module = appmanager.setup(('app', 8380))
class MultiCallbackTest(unittest.TestCase):
def setUp(self):
self.easy = util.DefaultCurl()
self.easy.setopt(pycurl.URL, 'http://%s:8380/long_pause' % localhost)
self.multi = pycurl.CurlMulti()
self.multi.setopt(pycurl.M_SOCKETFUNCTION, self.socket_callback)
self.multi.setopt(pycurl.M_TIMERFUNCTION, self.timer_callback)
self.socket_result = None
self.timer_result = None
self.sockets = {}
self.handle_added = False
def tearDown(self):
if self.handle_added:
self.multi.remove_handle(self.easy)
self.multi.close()
self.easy.close()
def socket_callback(self, ev_bitmask, sock_fd, multi, data):
self.socket_result = (sock_fd, ev_bitmask)
if ev_bitmask & pycurl.POLL_REMOVE:
self.sockets.pop(sock_fd)
else:
self.sockets[sock_fd] = ev_bitmask | self.sockets.get(sock_fd, 0)
def timer_callback(self, timeout_ms):
self.timer_result = timeout_ms
def partial_transfer(self):
perform = True
def write_callback(data):
nonlocal perform
perform = False
self.easy.setopt(pycurl.WRITEFUNCTION, write_callback)
self.multi.add_handle(self.easy)
self.handle_added = True
self.multi.socket_action(pycurl.SOCKET_TIMEOUT, 0)
while self.sockets and perform:
for socket, action in tuple(self.sockets.items()):
self.multi.socket_action(socket, action)
# multi.socket_action must call both SOCKETFUNCTION and TIMERFUNCTION at
# various points during the transfer (at least at the start and end)
def test_multi_socket_action(self):
self.multi.add_handle(self.easy)
self.handle_added = True
self.timer_result = None
self.socket_result = None
self.multi.socket_action(pycurl.SOCKET_TIMEOUT, 0)
assert self.socket_result is not None
assert self.timer_result is not None
# multi.add_handle must call TIMERFUNCTION to schedule a kick-start
def test_multi_add_handle(self):
self.multi.add_handle(self.easy)
self.handle_added = True
assert self.timer_result is not None
# (mid-transfer) multi.remove_handle must call SOCKETFUNCTION to remove sockets
def test_multi_remove_handle(self):
self.multi.add_handle(self.easy)
self.handle_added = True
self.multi.socket_action(pycurl.SOCKET_TIMEOUT, 0)
self.socket_result = None
self.multi.remove_handle(self.easy)
self.handle_added = False
assert self.socket_result is not None
# (mid-transfer) easy.pause(PAUSE_ALL) must call SOCKETFUNCTION to remove sockets
# (mid-transfer) easy.pause(PAUSE_CONT) must call TIMERFUNCTION to resume
@pytest.mark.skipif(sys.platform == 'win32', reason='https://github.com/pycurl/pycurl/issues/819')
def test_easy_pause_unpause(self):
self.partial_transfer()
self.socket_result = None
# libcurl will now inform us that we should remove some sockets
self.easy.pause(pycurl.PAUSE_ALL)
assert self.socket_result is not None
self.socket_result = None
self.timer_result = None
# libcurl will now tell us to add those sockets and schedule a kickstart
self.easy.pause(pycurl.PAUSE_CONT)
assert self.socket_result is not None
assert self.timer_result is not None
# (mid-transfer) easy.close() must call SOCKETFUNCTION to remove sockets
@pytest.mark.skipif(sys.platform in ['win32', 'darwin'], reason='https://github.com/pycurl/pycurl/issues/819')
def test_easy_close(self):
self.partial_transfer()
self.socket_result = None
self.easy.close()
assert self.socket_result is not None
|
