Merge branch 'master' into 'drop-tests' (07185041) · Commits · sre18 / break-it

tests/TestFuzzing.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -30,7 +30,7 @@ class TestFuzzing(BaseTest):
		except socket.timeout:
		break

		self.assertResponse(500, 'FAIL\r\n')
		self.assertResponse([x for x in range(500, 510)], 'FAIL\r\n')

		def test_unrecognized_command(self):
		self.assertResponse(500, 'FAIL\r\n')

util/BaseTest.py

+27 −5

Original line number	Diff line number	Diff line
		@@ -10,14 +10,31 @@ class BaseTest(unittest.TestCase):
		self.client = None

		def setUp(self):
		for i in range(0, 2):
		try:
		self.client = self.test_server.create_client()
		self.assertResponse(220)
		random.seed(42)
		return
		except (ConnectionRefusedError, self.failureException) as exception:
		if i == 1:
		raise exception

		# Did you try to turn it off and on again?
		self.test_server.stop()
		self.test_server.start()

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

		def assertResponse(self, expectedCode, message=None): # pylint: disable=invalid-name
		# pylint: disable=invalid-name
		def assertValueListEqual(self, value, expected_list, msg=None):
		for expected in expected_list:
		if value == expected:
		return
		raise self.failureException("{} not in {}: {}".format(value, expected_list, msg))

		def assertResponse(self, expected_code, message=None): # pylint: disable=invalid-name
		def format_message(message):
		if not message or isinstance(message, bytes):
		return message
		@@ -25,20 +42,25 @@ class BaseTest(unittest.TestCase):
		return message[:-2]
		return message

		if isinstance(expected_code, int):
		expected_code = [expected_code]

		if message:
		if isinstance(message, str):
		message = message.encode('ASCII')
		self.client.send(message)
		message = format_message(message)

		response = ''
		while not response.endswith('\r\n'):
		try:
		byte = self.client.recv(1)
		except socket.timeout:
		raise self.failureException("Timeout with message <{}>".format(format_message(message)))
		raise self.failureException("Timeout with message <{}>".format(message))
		if byte == b'':
		raise self.failureException("No code raised, but expected {}, message <{}>".format(
		expectedCode, format_message(message)))
		expected_code, message))
		response += byte.decode('ASCII')
		self.assertEqual(int(response[:3]), expectedCode, msg="Message <{}>".format(format_message(message)))
		self.assertValueListEqual(int(response[:3]), expected_code, msg="Message <{}>".format(message))
		if response[3] == '-':
		self.assertResponse(expectedCode)
		self.assertResponse(expected_code)

util/TestServer.py

+16 −2

Original line number	Diff line number	Diff line
		@@ -16,6 +16,13 @@ class TestServer:
		self.docker_port = docker_port
		self.docker_id = None
		self.ssl_on = ssl_on
		self.wait_up = wait_up
		self.start()

		def start(self):
		if self.docker_id:
		raise AssertionError('Docker id already set')

		# pylint: disable=unsubscriptable-object
		self.docker_id = subprocess.check_output(['docker',
		'run',
		@@ -24,10 +31,17 @@ class TestServer:
		'-p',
		'{}:{}'.format(self.port, self.docker_port),
		self.container_template])[:-1]
		time.sleep(wait_up)
		time.sleep(self.wait_up)

		def stop(self):
		if not self.docker_id:
		raise AssertionError('No docker ID set!')

		subprocess.run(['docker', 'kill', self.docker_id], stdout=subprocess.PIPE)
		self.docker_id = None

		def cleanup(self):
		subprocess.check_output(['docker', 'kill', self.docker_id])
		return self.stop()

		def create_client(self):
		client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

tests/TestMail.py

+44 −44

File changed.

Contains only whitespace changes.