Commit 149708fd authored by Łukasz Nowak's avatar Łukasz Nowak

test: flake8ize

parent f5f7c68e
......@@ -31,7 +31,6 @@ import requests
import requests.exceptions
import shutil
import signal
import sys
import tarfile
import tempfile
import time
......@@ -62,6 +61,7 @@ def findFreeTCPPort(ip=''):
s.bind((ip, 0))
return str(s.getsockname()[1])
def retry(callback, try_count=10, try_delay=0.1):
"""
Poll <callback> every <try_delay> for <try_count> times or until it returns
......@@ -75,6 +75,7 @@ def retry(callback, try_count=10, try_delay=0.1):
time.sleep(try_delay)
return result
def canConnect(caucase_url):
"""
Returns True if a connection can be established to given address, False
......@@ -97,13 +98,15 @@ class KedifaMixin(object):
_clean_caucased_snapshot = None
class KedifaCaucaseMixin(KedifaMixin):
def _startCaucaseServer(self, argv=(), timeout=10):
"""
Start caucased server
"""
ip, port = os.environ['SLAPOS_TEST_IPV6'],\
findFreeTCPPort(os.environ['SLAPOS_TEST_IPV6'])
ip, port = os.environ[
'SLAPOS_TEST_IPV6'], findFreeTCPPort(os.environ['SLAPOS_TEST_IPV6'])
self.caucase_runtime = caucase_runtime = multiprocessing.Process(
target=caucase.http.main,
kwargs=dict(
......@@ -120,8 +123,8 @@ class KedifaCaucaseMixin(KedifaMixin):
if not retry(
lambda: (
self.assertTrue(caucase_runtime.is_alive()) or
canConnect(self.caucase_url)
self.assertTrue(caucase_runtime.is_alive()) or canConnect(
self.caucase_url)
),
try_count=timeout * 10,
):
......@@ -141,7 +144,6 @@ class KedifaCaucaseMixin(KedifaMixin):
if caucase_runtime.is_alive():
raise ValueError('%r does not wish to die' % (caucase_runtime, ))
def createKey(self):
key = rsa.generate_private_key(
public_exponent=65537, key_size=2048, backend=default_backend())
......@@ -401,14 +403,15 @@ class KedifaIntegrationTest(KedifaCaucaseMixin, unittest.TestCase):
'Entry %r not found in log:\n %s' % (entry, ''.join(log_line_list)))
def assertLastLogEntry(self, entry):
tries_left = 5 # try few times, as server can store the log line a bit
# later
# try few times, as server can store the log line a bit later
tries_left = 5
while True:
with open(self.logfile) as fh:
last_log_line = fh.readlines()[-1]
try:
self.assertTrue(
entry in last_log_line, '%r not found in %r' % (entry, last_log_line))
entry in last_log_line,
'%r not found in %r' % (entry, last_log_line))
except AssertionError:
if tries_left == 0:
raise
......
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