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
|
#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vi:ts=4:et
from . import localhost
import unittest
import pycurl
import sys
import tempfile
import os
from . import appmanager, util
setup_module, teardown_module = appmanager.setup(('app', 8380))
STDOUT_FD_NUM = 1
def try_fsync(fd):
try:
os.fsync(fd)
except OSError:
# On travis:
# OSError: [Errno 22] Invalid argument
# ignore
pass
class DefaultWriteCbTest(unittest.TestCase):
def setUp(self):
self.curl = util.DefaultCurl()
def tearDown(self):
self.curl.close()
def test_perform_get(self):
# This test performs a GET request without doing anything else.
# Unfortunately, the default curl behavior is to print response
# body to standard output, which spams test output.
# As a result this test is commented out. Uncomment for debugging.
# test_perform_get_with_default_write_function is the test
# which exercises default curl write handler.
self.curl.setopt(pycurl.URL, 'http://%s:8380/success' % localhost)
self.curl.perform()
# If this flush is not done, stdout output bleeds into the next test
# that is executed (without nose output capture)
sys.stdout.flush()
try_fsync(STDOUT_FD_NUM)
# I have a really hard time getting this to work with nose output capture
def skip_perform_get_with_default_write_function(self):
self.curl.setopt(pycurl.URL, 'http://%s:8380/success' % localhost)
f = tempfile.NamedTemporaryFile()
try:
#with open('w', 'w+') as f:
# nose output capture plugin replaces sys.stdout with a StringIO
# instance. We want to redirect the underlying file descriptor
# anyway because underlying C code uses it.
# Therefore:
# 1. Use file descriptor 1 rather than sys.stdout.fileno() to
# reference the standard output file descriptor.
# 2. We do not touch sys.stdout. This means anything written to
# sys.stdout will be captured by nose, and not make it to our code.
# But the output we care about happens at libcurl level, below
# nose, therefore this is fine.
saved_stdout_fd = os.dup(STDOUT_FD_NUM)
os.dup2(f.fileno(), STDOUT_FD_NUM)
#os.dup2(1, 100)
#os.dup2(2, 1)
# We also need to flush the output that libcurl wrote to stdout.
# Since sys.stdout might be nose's StringIO instance, open the
# stdout file descriptor manually.
try:
self.curl.perform()
sys.stdout.flush()
finally:
try_fsync(STDOUT_FD_NUM)
os.dup2(saved_stdout_fd, STDOUT_FD_NUM)
os.close(saved_stdout_fd)
#os.dup2(100, 1)
f.seek(0)
body = f.read()
finally:
f.close()
self.assertEqual('success', body)
|
