Skip to content

Commit

Permalink
Merge branch 'python-unittest'
Browse files Browse the repository at this point in the history
Improved selftest for the test suite
  • Loading branch information
airtower-luna committed Jul 17, 2021
2 parents 70cf137 + caffc86 commit 5168eb0
Show file tree
Hide file tree
Showing 4 changed files with 178 additions and 49 deletions.
9 changes: 5 additions & 4 deletions test/Makefile.am
Expand Up @@ -49,7 +49,7 @@ dist_check_SCRIPTS = netns_py.bash test-template.bash.in

TEST_EXTENSIONS = .bash .py
PY_LOG_COMPILER = $(PYTHON)
TESTS = doctest-mgstest.py $(test_scripts)
TESTS = doctest-mgstest.py unittest-mgstest.py $(test_scripts)

check_PROGRAMS = pgpcrc
pgpcrc_SOURCES = pgpcrc.c
Expand All @@ -66,9 +66,10 @@ endif
# Python tools for tests
noinst_PYTHON = https-test-client.py mgstest/http.py mgstest/__init__.py \
mgstest/hooks.py mgstest/ocsp.py mgstest/services.py \
mgstest/softhsm.py mgstest/tests.py mgstest/valgrind.py runtest.py \
softhsm-init.py doctest-mgstest.py required-modules.py data/ocsp.py \
check_test_ips.py
mgstest/softhsm.py mgstest/tests.py mgstest/valgrind.py \
mgstest/test_tests.py runtest.py \
softhsm-init.py doctest-mgstest.py unittest-mgstest.py \
required-modules.py data/ocsp.py check_test_ips.py

# Identities in the miniature CA, server, and client environment for
# the test suite
Expand Down
162 changes: 162 additions & 0 deletions test/mgstest/test_tests.py
@@ -0,0 +1,162 @@
import unittest
import unittest.mock
import yaml
from http import HTTPStatus

from . import TestExpectationFailed
from .tests import TestRequest


def mock_response(status=HTTPStatus.OK, headers=dict(), body=b''):
response = unittest.mock.Mock()
response.status = status
response.reason = status.phrase
response.getheaders.return_value = [(k, v) for k, v in headers.items()]
response.read.return_value = body
return response


class TestTestRequest(unittest.TestCase):
def test_run(self):
"""Check that TestRequest matches regular response correctly."""
response = mock_response(
HTTPStatus.OK, {'X-Required-Header': 'Hi!'}, b'Hello World!\n')
conn = unittest.mock.Mock()
conn.getresponse.return_value = response

r = TestRequest(path='/test.txt',
expect={'status': 200,
'headers': {'X-Forbidden-Header': None,
'X-Required-Header': 'Hi!'},
'body': {'contains': 'Hello'}})
r.run(conn)
conn.request.assert_called_with(
'GET', '/test.txt', body=None, headers={})

def test_run_unexpected_reset(self):
"""An unexpected exception breaks out of the TestRequest run."""
conn = unittest.mock.Mock()
conn.request.side_effect = ConnectionResetError
r = TestRequest(path='/test.txt',
expect={'status': 200})
with self.assertRaises(ConnectionResetError):
r.run(conn)
conn.request.assert_called_with(
'GET', '/test.txt', body=None, headers={})

def test_run_expected_reset(self):
"""If the TestRequest expects an exception, it gets caught."""
conn = unittest.mock.Mock()
conn.request.side_effect = ConnectionResetError
r = TestRequest(path='/test.txt',
expect={'reset': True})
r.run(conn)
conn.request.assert_called_with(
'GET', '/test.txt', body=None, headers={})

def test_check_headers(self):
r = TestRequest(path='/test.txt',
expect={'headers': {'X-Forbidden-Header': None,
'X-Required-Header': 'Hi!'}})
r.check_headers({'X-Required-Header': 'Hi!'})

with self.assertRaisesRegex(TestExpectationFailed,
'Unexpected value in header'):
r.check_headers({'X-Required-Header': 'Hello!'})

with self.assertRaisesRegex(TestExpectationFailed,
'Unexpected value in header'):
r.check_headers({'X-Forbidden-Header': 'Hi!'})

def test_check_body_exact(self):
r = TestRequest(
path='/test.txt', method='GET', headers={},
expect={'status': 200, 'body': {'exactly': 'test\n'}})
r.check_body('test\n')

with self.assertRaisesRegex(
TestExpectationFailed,
r"Unexpected body: 'xyz\\n' != 'test\\n'"):
r.check_body('xyz\n')

def test_check_body_contains(self):
r1 = TestRequest(
path='/test.txt', method='GET', headers={},
expect={'status': 200, 'body': {'contains': ['tes', 'est']}})
r1.check_body('test\n')
with self.assertRaisesRegex(
TestExpectationFailed,
r"Unexpected body: 'est\\n' does not contain 'tes'"):
r1.check_body('est\n')

r2 = TestRequest(
path='/test.txt', method='GET', headers={},
expect={'status': 200, 'body': {'contains': 'test'}})
r2.check_body('test\n')

def test_expects_conn_reset(self):
r1 = TestRequest(path='/test.txt', method='GET', headers={},
expect={'status': 200, 'body': {'contains': 'test'}})
self.assertFalse(r1.expects_conn_reset())

r2 = TestRequest(path='/test.txt', method='GET', headers={},
expect={'reset': True})
self.assertTrue(r2.expects_conn_reset())


class TestTestConnection(unittest.TestCase):
def test_run(self):
"""TestConnection with a successful and a failing TestRequest."""

test = """
!connection
gnutls_params:
- x509cafile=authority/x509.pem
actions:
- !request
path: /test.txt
headers:
X-Test: mgstest
expect:
status: 200
headers:
X-Required: 'Hi!'
body:
exactly: |
Hello World!
- !request
method: POST
path: /test.txt
expect:
status: 200
body:
exactly: |
Hello World!
"""
conn = yaml.load(test, Loader=yaml.Loader)

responses = [
mock_response(
HTTPStatus.OK, {'X-Required': 'Hi!'},
b'Hello World!\n'),
mock_response(
HTTPStatus.METHOD_NOT_ALLOWED, {}, b'Cannot POST here!\n')]

# note that this patches HTTPSubprocessConnection as imported
# into mgstest.tests, not in the origin package
with unittest.mock.patch(
'mgstest.tests.HTTPSubprocessConnection', spec=True) as P:
# the mock provided by patch acts as class, get the instance
instance = P.return_value
instance.getresponse.side_effect = responses
with self.assertRaisesRegex(
TestExpectationFailed,
r"Unexpected status: 405 != 200"):
conn.run()

instance.request.assert_has_calls([
unittest.mock.call(
'GET', '/test.txt', body=None, headers={'X-Test': 'mgstest'}),
unittest.mock.call('POST', '/test.txt', body=None, headers={})
])
self.assertEqual(instance.request.call_count, 2)
45 changes: 0 additions & 45 deletions test/mgstest/tests.py
Expand Up @@ -250,22 +250,6 @@ def run(self, conn, response_log=None):
self.check_response(resp, body)

def check_headers(self, headers):
"""
>>> r1 = TestRequest(path='/test.txt',
... expect={'headers': {'X-Forbidden-Header': None,
... 'X-Required-Header': 'Hi!'}})
>>> r1.check_headers({'X-Required-Header': 'Hi!'})
>>> r1.check_headers({'X-Required-Header': 'Hello!'})
Traceback (most recent call last):
...
mgstest.TestExpectationFailed: Unexpected value in header \
X-Required-Header: 'Hello!', expected 'Hi!'
>>> r1.check_headers({'X-Forbidden-Header': 'Hi!'})
Traceback (most recent call last):
...
mgstest.TestExpectationFailed: Unexpected value in header \
X-Forbidden-Header: 'Hi!', expected None
"""
for name, expected in self.expect['headers'].items():
value = headers.get(name)
expected = subst_env(expected)
Expand All @@ -275,26 +259,6 @@ def check_headers(self, headers):
f'expected {expected!r}')

def check_body(self, body):
"""
>>> r1 = TestRequest(path='/test.txt', method='GET', headers={}, \
expect={'status': 200, 'body': {'exactly': 'test\\n'}})
>>> r1.check_body('test\\n')
>>> r1.check_body('xyz\\n')
Traceback (most recent call last):
...
mgstest.TestExpectationFailed: Unexpected body: 'xyz\\n' != 'test\\n'
>>> r2 = TestRequest(path='/test.txt', method='GET', headers={}, \
expect={'status': 200, 'body': {'contains': ['tes', 'est']}})
>>> r2.check_body('test\\n')
>>> r2.check_body('est\\n')
Traceback (most recent call last):
...
mgstest.TestExpectationFailed: Unexpected body: 'est\\n' \
does not contain 'tes'
>>> r3 = TestRequest(path='/test.txt', method='GET', headers={}, \
expect={'status': 200, 'body': {'contains': 'test'}})
>>> r3.check_body('test\\n')
"""
if 'exactly' in self.expect['body'] \
and body != self.expect['body']['exactly']:
raise TestExpectationFailed(
Expand Down Expand Up @@ -327,15 +291,6 @@ def expects_conn_reset(self):
"""Returns True if running this request is expected to fail due to the
connection being reset. That usually means the underlying TLS
connection failed.
>>> r1 = TestRequest(path='/test.txt', method='GET', headers={}, \
expect={'status': 200, 'body': {'contains': 'test'}})
>>> r1.expects_conn_reset()
False
>>> r2 = TestRequest(path='/test.txt', method='GET', headers={}, \
expect={'reset': True})
>>> r2.expects_conn_reset()
True
"""
if 'reset' in self.expect:
return self.expect['reset']
Expand Down
11 changes: 11 additions & 0 deletions test/unittest-mgstest.py
@@ -0,0 +1,11 @@
#!/usr/bin/python3
import os
import sys
import unittest

if __name__ == "__main__":
suite = unittest.defaultTestLoader.discover(
'mgstest', top_level_dir=os.environ.get('srcdir', '.'))
result = unittest.TextTestRunner(verbosity=2, buffer=True).run(suite)
if not result.wasSuccessful():
sys.exit(1)

0 comments on commit 5168eb0

Please sign in to comment.