FM / tests

/usr/share/doc/python3-pycurl/tests

Name	Size	Mode	Actions
certs/	-	0755	rm
ext/	-	0755	rm
fake-curl/	-	0755	rm
fixtures/	-	0755	rm
matrix/	-	0755	rm
tmp/	-	0755	rm
__pycache__/	-	0755	rm
app.py	4088	0644	edit dl rm
appmanager.py	1380	0644	edit dl rm
cadata_test.py	1452	0644	edit dl rm
certinfo_test.py	3468	0644	edit dl rm
close_socket_cb_test.py	2168	0644	edit dl rm
curl_object_test.py	4883	0644	edit dl rm
debug_test.py	2275	0644	edit dl rm
default_write_cb_test.py	3039	0644	edit dl rm
error_constants_test.py	662	0644	edit dl rm
error_test.py	2901	0644	edit dl rm
failonerror_test.py	3509	0644	edit dl rm
ftp_test.py	1537	0644	edit dl rm
global_init_test.py	1063	0644	edit dl rm
header_cb_test.py	1560	0644	edit dl rm
header_test.py	1848	0644	edit dl rm
high_level_curl_test.py	632	0644	edit dl rm
info_constants_test.py	387	0644	edit dl rm
internals_test.py	6674	0644	edit dl rm
matrix.py	6816	0644	edit dl rm
multi_option_constants_test.py	2503	0644	edit dl rm
multi_socket_test.py	3049	0644	edit dl rm
multi_test.py	12313	0644	edit dl rm
multi_timer_test.py	2695	0644	edit dl rm
open_socket_cb_test.py	4761	0644	edit dl rm
option_constants_test.py	16371	0644	edit dl rm
perform_test.py	2179	0644	edit dl rm
procmgr.py	2731	0644	edit dl rm
protocol_constants_test.py	1679	0644	edit dl rm
readdata_test.py	9623	0644	edit dl rm
read_cb_test.py	4340	0644	edit dl rm
relative_url_test.py	551	0644	edit dl rm
reload_test.py	430	0644	edit dl rm
reset_test.py	2699	0644	edit dl rm
resolve_test.py	780	0644	edit dl rm
run-quickstart.sh	768	0755	edit dl rm
run.sh	690	0755	edit dl rm
runwsgi.py	3733	0644	edit dl rm
seek_cb_constants_test.py	833	0644	edit dl rm
setopt_lifecycle_test.py	1797	0644	edit dl rm
setopt_string_test.py	816	0644	edit dl rm
setopt_test.py	3773	0644	edit dl rm
setopt_unicode_test.py	1092	0644	edit dl rm
setup_test.py	10035	0644	edit dl rm
share_test.py	1726	0644	edit dl rm
sockopt_cb_test.py	2453	0644	edit dl rm
unset_range_test.py	1482	0644	edit dl rm
user_agent_string_test.py	820	0644	edit dl rm
util.py	8291	0644	edit dl rm
version_comparison_test.py	514	0644	edit dl rm
version_constants_test.py	2056	0644	edit dl rm
version_test.py	313	0644	edit dl rm
vsftpd.conf	294	0644	edit dl rm
weakref_test.py	443	0644	edit dl rm
write_abort_test.py	1194	0644	edit dl rm
write_cb_bogus_test.py	1381	0644	edit dl rm
write_test.py	10048	0644	edit dl rm
write_to_stringio_test.py	1228	0644	edit dl rm
xferinfo_cb_test.py	2147	0644	edit dl rm
__init__.py	535	0644	edit dl rm

Edit: /usr/share/doc/python3-pycurl/tests/open_socket_cb_test.py (4761B)

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vi:ts=4:et

from . import localhost
import socket
import pycurl
import unittest

from . import appmanager
from . import util

setup_module, teardown_module = appmanager.setup(('app', 8380))

socket_open_called_ipv4 = False
socket_open_called_ipv6 = False
socket_open_called_unix = False
socket_open_address = None

def socket_open_ipv4(purpose, curl_address):
    family, socktype, protocol, address = curl_address
    global socket_open_called_ipv4
    global socket_open_address
    socket_open_called_ipv4 = True
    socket_open_address = address

s = socket.socket(family, socktype, protocol)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
    return s

def socket_open_ipv6(purpose, curl_address):
    family, socktype, protocol, address = curl_address
    global socket_open_called_ipv6
    global socket_open_address
    socket_open_called_ipv6 = True
    socket_open_address = address

s = socket.socket(family, socktype, protocol)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
    return s

def socket_open_unix(purpose, curl_address):
    family, socktype, protocol, address = curl_address
    global socket_open_called_unix
    global socket_open_address
    socket_open_called_unix = True
    socket_open_address = address

sockets = socket.socketpair()
    sockets[0].close()
    return sockets[1]

def socket_open_bad(purpose, curl_address):
    return pycurl.SOCKET_BAD

class OpenSocketCbTest(unittest.TestCase):
    def setUp(self):
        self.curl = util.DefaultCurl()

def tearDown(self):
        self.curl.close()

# This is failing too much on appveyor
    @util.only_unix
    def test_socket_open(self):
        self.curl.setopt(pycurl.OPENSOCKETFUNCTION, socket_open_ipv4)
        self.curl.setopt(self.curl.URL, 'http://%s:8380/success' % localhost)
        sio = util.BytesIO()
        self.curl.setopt(pycurl.WRITEFUNCTION, sio.write)
        self.curl.perform()

assert socket_open_called_ipv4
        self.assertEqual(("127.0.0.1", 8380), socket_open_address)
        self.assertEqual('success', sio.getvalue().decode())

@util.only_ipv6
    def test_socket_open_ipv6(self):
        self.curl.setopt(pycurl.OPENSOCKETFUNCTION, socket_open_ipv6)
        self.curl.setopt(self.curl.URL, 'http://[::1]:8380/success')
        sio = util.BytesIO()
        self.curl.setopt(pycurl.WRITEFUNCTION, sio.write)
        try:
            # perform fails because we do not listen on ::1
            self.curl.perform()
        except pycurl.error:
            pass

assert socket_open_called_ipv6

assert len(socket_open_address) == 4
        assert socket_open_address[0] == '::1'
        assert socket_open_address[1] == 8380
        assert type(socket_open_address[2]) == int
        assert type(socket_open_address[3]) == int

@util.min_libcurl(7, 40, 0)
    @util.only_unix
    def test_socket_open_unix(self):
        self.curl.setopt(pycurl.OPENSOCKETFUNCTION, socket_open_unix)
        self.curl.setopt(self.curl.URL, 'http://%s:8380/success' % localhost)
        self.curl.setopt(self.curl.UNIX_SOCKET_PATH, '/tmp/pycurl-test-path.sock')
        sio = util.BytesIO()
        self.curl.setopt(pycurl.WRITEFUNCTION, sio.write)
        try:
            # perform fails because we return a socket that is
            # not attached to anything
            self.curl.perform()
        except pycurl.error:
            pass

assert socket_open_called_unix
        if util.py3:
            assert isinstance(socket_open_address, bytes)
            self.assertEqual(b'/tmp/pycurl-test-path.sock', socket_open_address)
        else:
            assert isinstance(socket_open_address, str)
            self.assertEqual('/tmp/pycurl-test-path.sock', socket_open_address)

def test_socket_open_none(self):
        self.curl.setopt(pycurl.OPENSOCKETFUNCTION, None)

def test_unset_socket_open(self):
        self.curl.unsetopt(pycurl.OPENSOCKETFUNCTION)

def test_socket_bad(self):
        self.assertEqual(-1, pycurl.SOCKET_BAD)

def test_socket_open_bad(self):
        self.curl.setopt(pycurl.OPENSOCKETFUNCTION, socket_open_bad)
        self.curl.setopt(self.curl.URL, 'http://%s:8380/success' % localhost)
        try:
            self.curl.perform()
        except pycurl.error as e:
            # libcurl 7.38.0 for some reason fails with a timeout
            # (and spends 5 minutes on this test)
            if pycurl.version_info()[1].split('.') == ['7', '38', '0']:
                self.assertEqual(pycurl.E_OPERATION_TIMEDOUT, e.args[0])
            else:
                self.assertEqual(pycurl.E_COULDNT_CONNECT, e.args[0])
        else:
            self.fail('Should have raised')

#! /usr/bin/env python # -*- coding: utf-8 -*- # vi:ts=4:et from . import localhost import socket import pycurl import unittest from . import appmanager from . import util setup_module, teardown_module = appmanager.setup(('app', 8380)) socket_open_called_ipv4 = False socket_open_called_ipv6 = False socket_open_called_unix = False socket_open_address = None def socket_open_ipv4(purpose, curl_address): family, socktype, protocol, address = curl_address global socket_open_called_ipv4 global socket_open_address socket_open_called_ipv4 = True socket_open_address = address s = socket.socket(family, socktype, protocol) s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) return s def socket_open_ipv6(purpose, curl_address): family, socktype, protocol, address = curl_address global socket_open_called_ipv6 global socket_open_address socket_open_called_ipv6 = True socket_open_address = address s = socket.socket(family, socktype, protocol) s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) return s def socket_open_unix(purpose, curl_address): family, socktype, protocol, address = curl_address global socket_open_called_unix global socket_open_address socket_open_called_unix = True socket_open_address = address sockets = socket.socketpair() sockets[0].close() return sockets[1] def socket_open_bad(purpose, curl_address): return pycurl.SOCKET_BAD class OpenSocketCbTest(unittest.TestCase): def setUp(self): self.curl = util.DefaultCurl() def tearDown(self): self.curl.close() # This is failing too much on appveyor @util.only_unix def test_socket_open(self): self.curl.setopt(pycurl.OPENSOCKETFUNCTION, socket_open_ipv4) self.curl.setopt(self.curl.URL, 'http://%s:8380/success' % localhost) sio = util.BytesIO() self.curl.setopt(pycurl.WRITEFUNCTION, sio.write) self.curl.perform() assert socket_open_called_ipv4 self.assertEqual(("127.0.0.1", 8380), socket_open_address) self.assertEqual('success', sio.getvalue().decode()) @util.only_ipv6 def test_socket_open_ipv6(self): self.curl.setopt(pycurl.OPENSOCKETFUNCTION, socket_open_ipv6) self.curl.setopt(self.curl.URL, 'http://[::1]:8380/success') sio = util.BytesIO() self.curl.setopt(pycurl.WRITEFUNCTION, sio.write) try: # perform fails because we do not listen on ::1 self.curl.perform() except pycurl.error: pass assert socket_open_called_ipv6 assert len(socket_open_address) == 4 assert socket_open_address[0] == '::1' assert socket_open_address[1] == 8380 assert type(socket_open_address[2]) == int assert type(socket_open_address[3]) == int @util.min_libcurl(7, 40, 0) @util.only_unix def test_socket_open_unix(self): self.curl.setopt(pycurl.OPENSOCKETFUNCTION, socket_open_unix) self.curl.setopt(self.curl.URL, 'http://%s:8380/success' % localhost) self.curl.setopt(self.curl.UNIX_SOCKET_PATH, '/tmp/pycurl-test-path.sock') sio = util.BytesIO() self.curl.setopt(pycurl.WRITEFUNCTION, sio.write) try: # perform fails because we return a socket that is # not attached to anything self.curl.perform() except pycurl.error: pass assert socket_open_called_unix if util.py3: assert isinstance(socket_open_address, bytes) self.assertEqual(b'/tmp/pycurl-test-path.sock', socket_open_address) else: assert isinstance(socket_open_address, str) self.assertEqual('/tmp/pycurl-test-path.sock', socket_open_address) def test_socket_open_none(self): self.curl.setopt(pycurl.OPENSOCKETFUNCTION, None) def test_unset_socket_open(self): self.curl.unsetopt(pycurl.OPENSOCKETFUNCTION) def test_socket_bad(self): self.assertEqual(-1, pycurl.SOCKET_BAD) def test_socket_open_bad(self): self.curl.setopt(pycurl.OPENSOCKETFUNCTION, socket_open_bad) self.curl.setopt(self.curl.URL, 'http://%s:8380/success' % localhost) try: self.curl.perform() except pycurl.error as e: # libcurl 7.38.0 for some reason fails with a timeout # (and spends 5 minutes on this test) if pycurl.version_info()[1].split('.') == ['7', '38', '0']: self.assertEqual(pycurl.E_OPERATION_TIMEDOUT, e.args[0]) else: self.assertEqual(pycurl.E_COULDNT_CONNECT, e.args[0]) else: self.fail('Should have raised')