Commit 85640261 authored by Łukasz Nowak's avatar Łukasz Nowak

tests: Use simple capture system

capturer does not work in some of tests environment.
parent f6a34458
...@@ -17,7 +17,7 @@ ...@@ -17,7 +17,7 @@
# See COPYING file for full licensing terms. # See COPYING file for full licensing terms.
# See https://www.nexedi.com/licensing for rationale and options. # See https://www.nexedi.com/licensing for rationale and options.
import capturer import contextlib
import datetime import datetime
import httplib import httplib
import ipaddress import ipaddress
...@@ -27,6 +27,8 @@ import requests ...@@ -27,6 +27,8 @@ import requests
import requests.exceptions import requests.exceptions
import shutil import shutil
import signal import signal
import StringIO
import sys
import tempfile import tempfile
import time import time
import unittest import unittest
...@@ -45,6 +47,17 @@ from cli import getter ...@@ -45,6 +47,17 @@ from cli import getter
from cli import http from cli import http
@contextlib.contextmanager
def captured_output():
new_out, new_err = StringIO.StringIO(), StringIO.StringIO()
old_out, old_err = sys.stdout, sys.stderr
try:
sys.stdout, sys.stderr = new_out, new_err
yield sys.stdout, sys.stderr
finally:
sys.stdout, sys.stderr = old_out, old_err
class KedifaIntegrationTest(unittest.TestCase): class KedifaIntegrationTest(unittest.TestCase):
def createKey(self): def createKey(self):
key = rsa.generate_private_key( key = rsa.generate_private_key(
...@@ -162,16 +175,17 @@ class KedifaIntegrationTest(unittest.TestCase): ...@@ -162,16 +175,17 @@ class KedifaIntegrationTest(unittest.TestCase):
self.cas = cas.split() self.cas = cas.split()
kedifa_key_pem, csr_file = self.generateCSR(unicode(common_name)) kedifa_key_pem, csr_file = self.generateCSR(unicode(common_name))
with capturer.CaptureOutput() as captured: with captured_output() as (out, err):
caucase.cli.main(argv=self.cas + [ caucase.cli.main(argv=self.cas + [
'--send-csr', csr_file '--send-csr', csr_file
]) ])
self.assertEqual('', err.getvalue().strip())
output = out.getvalue().strip()
try: try:
csr_id = captured.get_text().split()[0] csr_id = output.split()[0]
except IndexError: except IndexError:
self.fail('capturer failed on %s' % (captured.get_text().split(),)) self.fail('csr_id parse failed: %s' % (output,))
caucase.cli.main(argv=self.cas + [ caucase.cli.main(argv=self.cas + [
'--get-crt', csr_id, kedifa_key_pem '--get-crt', csr_id, kedifa_key_pem
]) ])
...@@ -331,12 +345,13 @@ class KedifaIntegrationTest(unittest.TestCase): ...@@ -331,12 +345,13 @@ class KedifaIntegrationTest(unittest.TestCase):
self.kedifa_url + self.reference, cert=incorrect_key_pem) self.kedifa_url + self.reference, cert=incorrect_key_pem)
def revokeCaucaseServiceCertifice(self): def revokeCaucaseServiceCertifice(self):
with capturer.CaptureOutput() as captured: with captured_output() as (out, err):
caucase.cli.main(argv=self.cas + [ caucase.cli.main(argv=self.cas + [
'--revoke-crt', self.client_key_pem, self.client_key_pem '--revoke-crt', self.client_key_pem, self.client_key_pem
]) ])
self.assertEqual('', captured.get_text()) self.assertEqual('', out.getvalue().strip())
self.assertEqual('', err.getvalue().strip())
def test_GET_revoked_identity(self): def test_GET_revoked_identity(self):
self.put() self.put()
......
...@@ -21,7 +21,6 @@ from setuptools import setup, find_packages ...@@ -21,7 +21,6 @@ from setuptools import setup, find_packages
import versioneer import versioneer
tests_require = [ tests_require = [
'capturer',
'urllib3 >= 1.18', # https://github.com/urllib3/urllib3/issues/258 'urllib3 >= 1.18', # https://github.com/urllib3/urllib3/issues/258
'ipaddress', 'ipaddress',
] ]
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment