Skip to content

S
slapos
Commit 550f1417 authored by Jérome Perrin's avatar Jérome Perrin
Browse files
Options

python3 tests

parent 39377bee
Show whitespace changes
Inline Side-by-side
Showing with 258 additions and 122 deletions
software/caddy-frontend/test/setup.py
View file @ 550f1417
......@@ -24,6 +24,7 @@
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from __future__ import absolute_import
from setuptools import setup, find_packages
version = '0.0.1.dev0'
......
software/caddy-frontend/test/test.py
View file @ 550f1417
......@@ -25,29 +25,30 @@
#
##############################################################################
from __future__ import absolute_import
import glob
import os
import requests
import httplib
import six.moves.http_client
from requests_toolbelt.adapters import source
import json
import multiprocessing
import subprocess
from unittest import skip
import ssl
from BaseHTTPServer import HTTPServer
from BaseHTTPServer import BaseHTTPRequestHandler
from SocketServer import ThreadingMixIn
from six.moves.BaseHTTPServer import HTTPServer
from six.moves.BaseHTTPServer import BaseHTTPRequestHandler
from six.moves.socketserver import ThreadingMixIn
import time
import tempfile
import ipaddress
import StringIO
from io import StringIO
import gzip
import base64
import re
from slapos.recipe.librecipe import generateHashFromFiles
import xml.etree.ElementTree as ET
import urlparse
import six.moves.urllib.parse
import socket
import sys
import logging
......@@ -56,6 +57,8 @@ import string
from slapos.slap.standalone import SlapOSNodeInstanceError
import caucase.client
import caucase.utils
import six
from six.moves import range
try:
......@@ -130,7 +133,7 @@ def patch_broken_pipe_error():
"""Monkey Patch BaseServer.handle_error to not write
a stacktrace to stderr on broken pipe.
https://stackoverflow.com/a/7913160"""
from SocketServer import BaseServer
from six.moves.socketserver import BaseServer
handle_error = BaseServer.handle_error
......@@ -162,7 +165,7 @@ def createKey():
def createSelfSignedCertificate(name_list):
key, key_pem = createKey()
subject_alternative_name_list = x509.SubjectAlternativeName(
[x509.DNSName(unicode(q)) for q in name_list]
[x509.DNSName(six.text_type(q)) for q in name_list]
)
subject = issuer = x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, u'Test Self Signed Certificate'),
......@@ -192,10 +195,10 @@ def createCSR(common_name, ip=None):
subject_alternative_name_list = []
if ip is not None:
subject_alternative_name_list.append(
x509.IPAddress(ipaddress.ip_address(unicode(ip)))
x509.IPAddress(ipaddress.ip_address(six.text_type(ip)))
)
csr = x509.CertificateSigningRequestBuilder().subject_name(x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, unicode(common_name)),
x509.NameAttribute(NameOID.COMMON_NAME, six.text_type(common_name)),
]))
if len(subject_alternative_name_list):
......@@ -219,10 +222,10 @@ class CertificateAuthority(object):
public_key = self.key.public_key()
builder = x509.CertificateBuilder()
builder = builder.subject_name(x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, unicode(common_name)),
x509.NameAttribute(NameOID.COMMON_NAME, six.text_type(common_name)),
]))
builder = builder.issuer_name(x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, unicode(common_name)),
x509.NameAttribute(NameOID.COMMON_NAME, six.text_type(common_name)),
]))
builder = builder.not_valid_before(
datetime.datetime.utcnow() - datetime.timedelta(days=2))
......@@ -608,7 +611,7 @@ class TestHandler(BaseHTTPRequestHandler):
self.send_header('Via', 'http/1.1 backendvia')
if compress:
self.send_header('Content-Encoding', 'gzip')
out = StringIO.StringIO()
out = StringIO()
# compress with level 0, to find out if in the middle someting would
# like to alter the compression
with gzip.GzipFile(fileobj=out, mode="w", compresslevel=0) as f:
......@@ -717,7 +720,7 @@ class HttpFrontendTestCase(SlapOSInstanceTestCase):
master_parameter_dict = self.parseConnectionParameterDict()
caucase_url = master_parameter_dict['backend-client-caucase-url']
ca_certificate = requests.get(caucase_url + '/cas/crt/ca.crt.pem')
assert ca_certificate.status_code == httplib.OK
assert ca_certificate.status_code == six.moves.http_client.OK
ca_certificate_file = os.path.join(
self.working_directory, 'ca-backend-client.crt.pem')
with open(ca_certificate_file, 'w') as fh:
......@@ -925,7 +928,7 @@ class HttpFrontendTestCase(SlapOSInstanceTestCase):
frontend, url = entry
result = requests.get(url, verify=False)
self.assertEqual(
httplib.OK,
six.moves.http_client.OK,
result.status_code,
'While accessing %r of %r the status code was %r' % (
url, frontend, result.status_code))
......@@ -939,11 +942,11 @@ class HttpFrontendTestCase(SlapOSInstanceTestCase):
sorted([q['name'] for q in result.json()]),
['access.log', 'backend.log', 'error.log'])
self.assertEqual(
httplib.OK,
six.moves.http_client.OK,
requests.get(url + 'access.log', verify=False).status_code
)
self.assertEqual(
httplib.OK,
six.moves.http_client.OK,
requests.get(url + 'error.log', verify=False).status_code
)
# assert only for few tests, as backend log is not available for many of
......@@ -952,7 +955,7 @@ class HttpFrontendTestCase(SlapOSInstanceTestCase):
'test_url', 'test_auth_to_backend', 'test_compressed_result']:
if self.id().endswith(test_name):
self.assertEqual(
httplib.OK,
six.moves.http_client.OK,
requests.get(url + 'backend.log', verify=False).status_code
)
......@@ -1006,7 +1009,7 @@ class HttpFrontendTestCase(SlapOSInstanceTestCase):
backend_haproxy_statistic_url_dict[key] = parameter_dict.pop(key)
self.assertEqual(
[url_key],
backend_haproxy_statistic_url_dict.keys()
list(backend_haproxy_statistic_url_dict.keys())
)
backend_haproxy_statistic_url = backend_haproxy_statistic_url_dict[url_key]
......@@ -1014,7 +1017,7 @@ class HttpFrontendTestCase(SlapOSInstanceTestCase):
backend_haproxy_statistic_url,
verify=False,
)
self.assertEqual(httplib.OK, result.status_code)
self.assertEqual(six.moves.http_client.OK, result.status_code)
self.assertIn('testing partition 0', result.text)
self.assertIn('Statistics Report for HAProxy', result.text)
......@@ -2072,12 +2075,12 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
self.certificate_pem,
der2pem(result.peercert))
self.assertEqual(httplib.SERVICE_UNAVAILABLE, result.status_code)
self.assertEqual(six.moves.http_client.SERVICE_UNAVAILABLE, result.status_code)
result_http = fakeHTTPResult(
parameter_dict['domain'], 'test-path')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result_http.status_code
)
......@@ -2089,7 +2092,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
# check that 404 is as configured
result_missing = fakeHTTPSResult(
'forsuredoesnotexists.example.com', '')
self.assertEqual(httplib.NOT_FOUND, result_missing.status_code)
self.assertEqual(six.moves.http_client.NOT_FOUND, result_missing.status_code)
self.assertEqual(
"""<html>
<head>
......@@ -2245,7 +2248,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
'test-path/deep/.././deeper')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result_http.status_code
)
......@@ -2366,7 +2369,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
self.assertEqual(
result.status_code,
httplib.BAD_GATEWAY
six.moves.http_client.BAD_GATEWAY
)
finally:
self.stopAuthenticatedServerProcess()
......@@ -2408,7 +2411,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
'test-path/deep/.././deeper')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result_http.status_code
)
......@@ -2551,7 +2554,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
der2pem(result.peercert))
self.assertEqual(
httplib.MOVED_PERMANENTLY,
six.moves.http_client.MOVED_PERMANENTLY,
result.status_code
)
......@@ -2705,9 +2708,15 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
# as now the place to put the key is known put the key there
auth = requests.get(
<<<<<<< HEAD
self.current_generate_auth,
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, auth.status_code)
=======
generate_auth,
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, auth.status_code)
>>>>>>> c5ba49d90 (python3 tests)
data = self.customdomain_ca_certificate_pem + \
self.customdomain_ca_key_pem + \
......@@ -2716,8 +2725,13 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
upload = requests.put(
self.current_upload_url + auth.text,
data=data,
<<<<<<< HEAD
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, upload.status_code)
=======
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, upload.status_code)
>>>>>>> c5ba49d90 (python3 tests)
self.runKedifaUpdater()
result = fakeHTTPSResult(
......@@ -2741,9 +2755,15 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
self.assertSlaveBase('ssl_ca_crt_only')
# as now the place to put the key is known put the key there
auth = requests.get(
<<<<<<< HEAD
self.current_generate_auth,
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, auth.status_code)
=======
generate_auth,
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, auth.status_code)
>>>>>>> c5ba49d90 (python3 tests)
data = self.ca.certificate_pem
......@@ -2752,7 +2772,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
data=data,
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.UNPROCESSABLE_ENTITY, upload.status_code)
self.assertEqual(six.moves.http_client.UNPROCESSABLE_ENTITY, upload.status_code)
self.assertEqual('Key incorrect', upload.text)
def test_ssl_ca_crt_garbage(self):
......@@ -2760,9 +2780,15 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
# as now the place to put the key is known put the key there
auth = requests.get(
<<<<<<< HEAD
self.current_generate_auth,
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, auth.status_code)
=======
generate_auth,
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, auth.status_code)
>>>>>>> c5ba49d90 (python3 tests)
_, ca_key_pem, csr, _ = createCSR(
parameter_dict['domain'])
......@@ -2774,7 +2800,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
data=data,
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, upload.status_code)
self.assertEqual(six.moves.http_client.CREATED, upload.status_code)
self.runKedifaUpdater()
result = fakeHTTPSResult(
......@@ -2799,9 +2825,15 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
parameter_dict = self.assertSlaveBase('ssl_ca_crt_does_not_match')
# as now the place to put the key is known put the key there
auth = requests.get(
<<<<<<< HEAD
self.current_generate_auth,
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, auth.status_code)
=======
generate_auth,
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, auth.status_code)
>>>>>>> c5ba49d90 (python3 tests)
data = self.certificate_pem + self.key_pem + self.ca.certificate_pem
......@@ -2810,7 +2842,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
data=data,
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, upload.status_code)
self.assertEqual(six.moves.http_client.CREATED, upload.status_code)
self.runKedifaUpdater()
result = fakeHTTPSResult(
......@@ -2904,16 +2936,27 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
# as now the place to put the key is known put the key there
auth = requests.get(
<<<<<<< HEAD
self.current_generate_auth,
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, auth.status_code)
=======
generate_auth,
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, auth.status_code)
>>>>>>> c5ba49d90 (python3 tests)
data = self.customdomain_certificate_pem + \
self.customdomain_key_pem
upload = requests.put(
self.current_upload_url + auth.text,
data=data,
<<<<<<< HEAD
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, upload.status_code)
=======
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, upload.status_code)
>>>>>>> c5ba49d90 (python3 tests)
self.runKedifaUpdater()
result = fakeHTTPSResult(
......@@ -2954,7 +2997,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
'test-path/deep/.././deeper')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result.status_code
)
......@@ -3073,7 +3116,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
'test-path/deep/.././deeper')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result.status_code
)
......@@ -3114,7 +3157,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
headers={'Accept-Encoding': 'gzip, deflate'})
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result.status_code
)
......@@ -3234,7 +3277,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
j = result.json()
except Exception:
raise ValueError('JSON decode problem in:\n%s' % (result.text,))
parsed = urlparse.urlparse(self.backend_url)
parsed = six.moves.urllib.parse.urlparse(self.backend_url)
self.assertBackendHeaders(
j['Incoming Headers'], parsed.hostname, port='17', proto='irc',
ignore_header_list=['Host'])
......@@ -3340,7 +3383,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
j = result.json()
except Exception:
raise ValueError('JSON decode problem in:\n%s' % (result.text,))
parsed = urlparse.urlparse(self.backend_url)
parsed = six.moves.urllib.parse.urlparse(self.backend_url)
self.assertBackendHeaders(
j['Incoming Headers'], parsed.hostname, port='17', proto='irc',
ignore_header_list=['Host'])
......@@ -3406,7 +3449,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
der2pem(result.peercert))
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result.status_code
)
......@@ -3428,7 +3471,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
der2pem(result.peercert))
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result.status_code
)
......@@ -3449,7 +3492,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
der2pem(result.peercert))
self.assertEqual(
httplib.SERVICE_UNAVAILABLE,
six.moves.http_client.SERVICE_UNAVAILABLE,
result.status_code
)
......@@ -3457,7 +3500,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
parameter_dict['domain'], 'test-path')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result_http.status_code
)
......@@ -3496,7 +3539,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
parameter_dict['domain'], 'test-path')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result_http.status_code
)
......@@ -3517,7 +3560,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
der2pem(result.peercert))
self.assertEqual(
httplib.SERVICE_UNAVAILABLE,
six.moves.http_client.SERVICE_UNAVAILABLE,
result.status_code
)
......@@ -3531,12 +3574,12 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
self.certificate_pem,
der2pem(result.peercert))
self.assertEqual(httplib.SERVICE_UNAVAILABLE, result.status_code)
self.assertEqual(six.moves.http_client.SERVICE_UNAVAILABLE, result.status_code)
result_http = fakeHTTPResult(
parameter_dict['domain'], 'test-path')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result_http.status_code
)
......@@ -3568,12 +3611,12 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
self.certificate_pem,
der2pem(result.peercert))
self.assertEqual(httplib.SERVICE_UNAVAILABLE, result.status_code)
self.assertEqual(six.moves.http_client.SERVICE_UNAVAILABLE, result.status_code)
result_http = fakeHTTPResult(
parameter_dict['domain'], 'test-path')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result_http.status_code
)
......@@ -3606,13 +3649,13 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
self.certificate_pem,
der2pem(result.peercert))
self.assertEqual(httplib.SERVICE_UNAVAILABLE, result.status_code)
self.assertEqual(six.moves.http_client.SERVICE_UNAVAILABLE, result.status_code)
result_http = fakeHTTPResult(
parameter_dict['domain'], 'test-path')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result_http.status_code
)
......@@ -3694,7 +3737,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
'X-Reply-Header-Cache-Control': 'max-age=1, stale-while-'
'revalidate=3600, stale-if-error=3600'})
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result.status_code
)
......@@ -3733,10 +3776,25 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
'X-Reply-Header-Cache-Control': 'max-age=1, stale-while-'
'revalidate=3600, stale-if-error=3600'})
self.assertEqual(httplib.OK, result.status_code)
self.assertEqual(six.moves.http_client.OK, result.status_code)
self.assertEqualResultJson(result, 'Path', '/HTTPS/test')
<<<<<<< HEAD
self.assertResponseHeaders(result, cached=True)
=======
headers = result.headers.copy()
result = fakeHTTPSResult(
parameter_dict['domain'],
'HTTP/test', headers={
'X-Reply-Header-Cache-Control': 'max-age=1, stale-while-'
'revalidate=3600, stale-if-error=3600'})
self.assertEqual(six.moves.http_client.OK, result.status_code)
self.assertEqualResultJson(result, 'Path', '/HTTP/test')
headers = result.headers.copy()
>>>>>>> c5ba49d90 (python3 tests)
def test_enable_cache(self):
parameter_dict = self.assertSlaveBase('enable_cache')
......@@ -3880,7 +3938,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
# https://github.com/apache/trafficserver/issues/7880
'X-Drop-Header': 'Content-Length',
})
self.assertEqual(result.status_code, httplib.CREATED)
self.assertEqual(result.status_code, six.moves.http_client.CREATED)
def checkResult(status_code, body):
result = fakeHTTPSResult(
......@@ -3892,39 +3950,39 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
# backend returns something correctly
configureResult('200', body_200)
checkResult(httplib.OK, body_200)
checkResult(six.moves.http_client.OK, body_200)
configureResult('502', body_502)
time.sleep(1)
# even if backend returns 502, ATS gives cached result
checkResult(httplib.OK, body_200)
checkResult(six.moves.http_client.OK, body_200)
# interesting moment, time is between max_age and max_stale_age, triggers
# https://github.com/apache/trafficserver/issues/7880
time.sleep(max_age + 1)
checkResult(httplib.OK, body_200)
checkResult(six.moves.http_client.OK, body_200)
# max_stale_age passed, time to return 502 from the backend
time.sleep(max_stale_age + 2)
checkResult(httplib.BAD_GATEWAY, body_502)
checkResult(six.moves.http_client.BAD_GATEWAY, body_502)
configureResult('502', body_502_new)
time.sleep(1)
# even if there is new negative response on the backend, the old one is
# served from the cache
checkResult(httplib.BAD_GATEWAY, body_502)
checkResult(six.moves.http_client.BAD_GATEWAY, body_502)
time.sleep(max_age + 2)
# now as max-age of negative response passed, the new one is served
checkResult(httplib.BAD_GATEWAY, body_502_new)
checkResult(six.moves.http_client.BAD_GATEWAY, body_502_new)
configureResult('200', body_200_new)
time.sleep(1)
checkResult(httplib.BAD_GATEWAY, body_502_new)
checkResult(six.moves.http_client.BAD_GATEWAY, body_502_new)
time.sleep(max_age + 2)
# backend is back to normal, as soon as negative response max-age passed
# the new response is served
checkResult(httplib.OK, body_200_new)
checkResult(six.moves.http_client.OK, body_200_new)
@skip('Feature postponed')
def test_enable_cache_stale_if_error_respected(self):
......@@ -3976,7 +4034,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
},
source_ip=source_ip
)
self.assertEqual(result.status_code, httplib.BAD_GATEWAY)
self.assertEqual(result.status_code, six.moves.http_client.BAD_GATEWAY)
finally:
self.startServerProcess()
# END: check stale-if-error support
......@@ -3994,7 +4052,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
# ATS timed out
self.assertEqual(
httplib.GATEWAY_TIMEOUT,
six.moves.http_client.GATEWAY_TIMEOUT,
result.status_code
)
......@@ -4094,7 +4152,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
j = result.json()
except Exception:
raise ValueError('JSON decode problem in:\n%s' % (result.text,))
self.assertFalse('pragma' in j['Incoming Headers'].keys())
self.assertFalse('pragma' in list(j['Incoming Headers'].keys()))
def test_enable_cache_disable_via_header(self):
parameter_dict = self.assertSlaveBase('enable_cache-disable-via-header')
......@@ -4313,7 +4371,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
headers={'Accept-Encoding': 'gzip, deflate'})
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result.status_code
)
......@@ -4328,7 +4386,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
headers={'Accept-Encoding': 'deflate'})
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result.status_code
)
......@@ -4342,7 +4400,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
'test-path/deep/.././deeper')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result.status_code
)
......@@ -4356,7 +4414,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
'test-path/deep/.././deeper')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result.status_code
)
......@@ -4409,7 +4467,7 @@ class TestSlave(SlaveHttpFrontendTestCase, TestDataMixin):
'test-path/deep/.././deeper')
self.assertEqual(
httplib.FOUND,
six.moves.http_client.FOUND,
result_http.status_code
)
......@@ -4534,7 +4592,7 @@ class TestReplicateSlave(SlaveHttpFrontendTestCase, TestDataMixin):
result_http = fakeHTTPResult(
parameter_dict['domain'], 'test-path')
self.assertEqual(httplib.FOUND, result_http.status_code)
self.assertEqual(six.moves.http_client.FOUND, result_http.status_code)
# prove 2nd frontend by inspection of the instance
slave_configuration_name = '_replicate.conf'
......@@ -4885,7 +4943,15 @@ class TestSlaveSlapOSMasterCertificateCompatibilityOverrideMaster(
cls.untilNotReadyYetNotInMasterKeyGenerateAuthUrl)
parameter_dict = cls.requestDefaultInstance().getConnectionParameterDict()
<<<<<<< HEAD
cls._fetchKedifaCaucaseCaCertificateFile(parameter_dict)
=======
ca_certificate = requests.get(
parameter_dict['kedifa-caucase-url'] + '/cas/crt/ca.crt.pem')
assert ca_certificate.status_code == six.moves.http_client.OK
cls.ca_certificate_file = os.path.join(cls.working_directory, 'ca.crt.pem')
open(cls.ca_certificate_file, 'w').write(ca_certificate.text)
>>>>>>> c5ba49d90 (python3 tests)
# Do not upload certificates for the master partition
@classmethod
......@@ -4957,7 +5023,15 @@ class TestSlaveSlapOSMasterCertificateCompatibility(
cls.untilNotReadyYetNotInMasterKeyGenerateAuthUrl)
parameter_dict = cls.requestDefaultInstance().getConnectionParameterDict()
<<<<<<< HEAD
cls._fetchKedifaCaucaseCaCertificateFile(parameter_dict)
=======
ca_certificate = requests.get(
parameter_dict['kedifa-caucase-url'] + '/cas/crt/ca.crt.pem')
assert ca_certificate.status_code == six.moves.http_client.OK
cls.ca_certificate_file = os.path.join(cls.working_directory, 'ca.crt.pem')
open(cls.ca_certificate_file, 'w').write(ca_certificate.text)
>>>>>>> c5ba49d90 (python3 tests)
# Do not upload certificates for the master partition
@classmethod
......@@ -5190,17 +5264,28 @@ class TestSlaveSlapOSMasterCertificateCompatibility(
# as now the place to put the key is known put the key there
auth = requests.get(
<<<<<<< HEAD
self.current_generate_auth,
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, auth.status_code)
=======
generate_auth,
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, auth.status_code)
>>>>>>> c5ba49d90 (python3 tests)
data = certificate_pem + key_pem
upload = requests.put(
self.current_upload_url + auth.text,
data=data,
<<<<<<< HEAD
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, upload.status_code)
=======
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, upload.status_code)
>>>>>>> c5ba49d90 (python3 tests)
self.runKedifaUpdater()
result = fakeHTTPSResult(
......@@ -5253,17 +5338,28 @@ class TestSlaveSlapOSMasterCertificateCompatibility(
# as now the place to put the key is known put the key there
auth = requests.get(
<<<<<<< HEAD
self.current_generate_auth,
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, auth.status_code)
=======
generate_auth,
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, auth.status_code)
>>>>>>> c5ba49d90 (python3 tests)
data = certificate_pem + key_pem
upload = requests.put(
self.current_upload_url + auth.text,
data=data,
<<<<<<< HEAD
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, upload.status_code)
=======
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, upload.status_code)
>>>>>>> c5ba49d90 (python3 tests)
self.runKedifaUpdater()
......@@ -5308,17 +5404,28 @@ class TestSlaveSlapOSMasterCertificateCompatibility(
# as now the place to put the key is known put the key there
auth = requests.get(
<<<<<<< HEAD
self.current_generate_auth,
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, auth.status_code)
=======
generate_auth,
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, auth.status_code)
>>>>>>> c5ba49d90 (python3 tests)
data = certificate_pem + key_pem
upload = requests.put(
self.current_upload_url + auth.text,
data=data,
<<<<<<< HEAD
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, upload.status_code)
=======
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, upload.status_code)
>>>>>>> c5ba49d90 (python3 tests)
self.runKedifaUpdater()
......@@ -5375,17 +5482,28 @@ class TestSlaveSlapOSMasterCertificateCompatibility(
# as now the place to put the key is known put the key there
auth = requests.get(
<<<<<<< HEAD
self.current_generate_auth,
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, auth.status_code)
=======
generate_auth,
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, auth.status_code)
>>>>>>> c5ba49d90 (python3 tests)
data = certificate_pem + key_pem
upload = requests.put(
self.current_upload_url + auth.text,
data=data,
<<<<<<< HEAD
verify=self.kedifa_caucase_ca_certificate_file)
self.assertEqual(httplib.CREATED, upload.status_code)
=======
verify=self.ca_certificate_file)
self.assertEqual(six.moves.http_client.CREATED, upload.status_code)
>>>>>>> c5ba49d90 (python3 tests)
self.runKedifaUpdater()
......@@ -5555,7 +5673,15 @@ class TestSlaveSlapOSMasterCertificateCompatibilityUpdate(
cls.untilNotReadyYetNotInMasterKeyGenerateAuthUrl)
parameter_dict = cls.requestDefaultInstance().getConnectionParameterDict()
<<<<<<< HEAD
cls._fetchKedifaCaucaseCaCertificateFile(parameter_dict)
=======
ca_certificate = requests.get(
parameter_dict['kedifa-caucase-url'] + '/cas/crt/ca.crt.pem')
assert ca_certificate.status_code == six.moves.http_client.OK
cls.ca_certificate_file = os.path.join(cls.working_directory, 'ca.crt.pem')
open(cls.ca_certificate_file, 'w').write(ca_certificate.text)
>>>>>>> c5ba49d90 (python3 tests)
# Do not upload certificates for the master partition
instance_parameter_dict = {
......@@ -5710,11 +5836,11 @@ class TestSlaveCiphers(SlaveHttpFrontendTestCase, TestDataMixin):
self.certificate_pem,
der2pem(result.peercert))
self.assertEqual(httplib.OK, result.status_code)
self.assertEqual(six.moves.http_client.OK, result.status_code)
result_http = fakeHTTPResult(
parameter_dict['domain'], 'test-path')
self.assertEqual(httplib.FOUND, result_http.status_code)
self.assertEqual(six.moves.http_client.FOUND, result_http.status_code)
configuration_file = glob.glob(
os.path.join(
......@@ -5736,11 +5862,11 @@ class TestSlaveCiphers(SlaveHttpFrontendTestCase, TestDataMixin):
self.certificate_pem,
der2pem(result.peercert))
self.assertEqual(httplib.OK, result.status_code)
self.assertEqual(six.moves.http_client.OK, result.status_code)
result_http = fakeHTTPResult(
parameter_dict['domain'], 'test-path')
self.assertEqual(httplib.FOUND, result_http.status_code)
self.assertEqual(six.moves.http_client.FOUND, result_http.status_code)
configuration_file = glob.glob(
os.path.join(
......@@ -6204,7 +6330,7 @@ class TestSlaveRejectReportUnsafeDamaged(SlaveHttpFrontendTestCase):
der2pem(result.peercert))
self.assertEqual(
httplib.MOVED_PERMANENTLY,
six.moves.http_client.MOVED_PERMANENTLY,
result.status_code
)
......@@ -6224,11 +6350,11 @@ class TestSlaveRejectReportUnsafeDamaged(SlaveHttpFrontendTestCase):
self.certificate_pem,
der2pem(result.peercert))
self.assertEqual(httplib.SERVICE_UNAVAILABLE, result.status_code)
self.assertEqual(six.moves.http_client.SERVICE_UNAVAILABLE, result.status_code)
result_http = fakeHTTPResult(
parameter_dict['domain'], 'test-path')
self.assertEqual(httplib.FOUND, result_http.status_code)
self.assertEqual(six.moves.http_client.FOUND, result_http.status_code)
monitor_file = glob.glob(
os.path.join(
......@@ -6255,11 +6381,11 @@ class TestSlaveRejectReportUnsafeDamaged(SlaveHttpFrontendTestCase):
self.certificate_pem,
der2pem(result.peercert))
self.assertEqual(httplib.SERVICE_UNAVAILABLE, result.status_code)
self.assertEqual(six.moves.http_client.SERVICE_UNAVAILABLE, result.status_code)
result_http = fakeHTTPResult(
parameter_dict['domain'], 'test-path')
self.assertEqual(httplib.FOUND, result_http.status_code)
self.assertEqual(six.moves.http_client.FOUND, result_http.status_code)
monitor_file = glob.glob(
os.path.join(
......@@ -6553,8 +6679,8 @@ class TestPassedRequestParameter(HttpFrontendTestCase):
expected_partition_parameter_dict_dict = {
'caddy-frontend-1': {
'X-software_release_url': base_software_url,
u'apache-certificate': unicode(self.certificate_pem),
u'apache-key': unicode(self.key_pem),
u'apache-certificate': six.text_type(self.certificate_pem),
u'apache-key': six.text_type(self.key_pem),
u'authenticate-to-backend': u'True',
u'backend-client-caucase-url': backend_client_caucase_url,
u'backend-connect-retries': u'1',
......@@ -6580,8 +6706,8 @@ class TestPassedRequestParameter(HttpFrontendTestCase):
},
'caddy-frontend-2': {
'X-software_release_url': self.frontend_2_sr,
u'apache-certificate': unicode(self.certificate_pem),
u'apache-key': unicode(self.key_pem),
u'apache-certificate': six.text_type(self.certificate_pem),
u'apache-key': six.text_type(self.key_pem),
u'authenticate-to-backend': u'True',
u'backend-client-caucase-url': backend_client_caucase_url,
u'backend-connect-retries': u'1',
......@@ -6607,8 +6733,8 @@ class TestPassedRequestParameter(HttpFrontendTestCase):
},
'caddy-frontend-3': {
'X-software_release_url': self.frontend_3_sr,
u'apache-certificate': unicode(self.certificate_pem),
u'apache-key': unicode(self.key_pem),
u'apache-certificate': six.text_type(self.certificate_pem),
u'apache-key': six.text_type(self.key_pem),
u'authenticate-to-backend': u'True',
u'backend-client-caucase-url': backend_client_caucase_url,
u'backend-connect-retries': u'1',
......@@ -6653,8 +6779,8 @@ class TestPassedRequestParameter(HttpFrontendTestCase):
'-sla-2-computer_guid': 'local',
'-sla-3-computer_guid': 'local',
'X-software_release_url': base_software_url,
'apache-certificate': unicode(self.certificate_pem),
'apache-key': unicode(self.key_pem),
'apache-certificate': six.text_type(self.certificate_pem),
'apache-key': six.text_type(self.key_pem),
'authenticate-to-backend': 'True',
'automatic-internal-backend-client-caucase-csr': 'False',
'automatic-internal-kedifa-caucase-csr': 'False',
......@@ -6809,7 +6935,7 @@ class TestSlaveHealthCheck(SlaveHttpFrontendTestCase, TestDataMixin):
@classmethod
def setUpAssertionDict(cls):
backend = urlparse.urlparse(cls.backend_url).netloc
backend = six.moves.urllib.parse.urlparse(cls.backend_url).netloc
cls.assertion_dict = {
'health-check-disabled': """\
backend _health-check-disabled-http
......@@ -6894,7 +7020,7 @@ backend _health-check-default-http
self.backend_url + slave_parameter_dict[
'health-check-http-path'].strip('/'),
headers={'X-Reply-Status-Code': '502'})
self.assertEqual(result.status_code, httplib.CREATED)
self.assertEqual(result.status_code, six.moves.http_client.CREATED)
def restoreBackend():
result = requests.put(
......@@ -7001,7 +7127,7 @@ backend _health-check-default-http
self.backend_url + slave_parameter_dict[
'health-check-http-path'].strip('/'),
headers={'X-Reply-Status-Code': '502'})
self.assertEqual(result.status_code, httplib.CREATED)
self.assertEqual(result.status_code, six.moves.http_client.CREATED)
time.sleep(3) # > health-check-timeout + health-check-interval
......@@ -7034,7 +7160,7 @@ backend _health-check-default-http
self.backend_url + slave_parameter_dict[
'health-check-http-path'].strip('/'),
headers={'X-Reply-Status-Code': '502'})
self.assertEqual(result.status_code, httplib.CREATED)
self.assertEqual(result.status_code, six.moves.http_client.CREATED)
time.sleep(3) # > health-check-timeout + health-check-interval
......@@ -7063,7 +7189,7 @@ backend _health-check-default-http
self.backend_url + slave_parameter_dict[
'health-check-http-path'].strip('/'),
headers={'X-Reply-Status-Code': '502'})
self.assertEqual(result.status_code, httplib.CREATED)
self.assertEqual(result.status_code, six.moves.http_client.CREATED)
time.sleep(3) # > health-check-timeout + health-check-interval
......@@ -7075,7 +7201,7 @@ backend _health-check-default-http
der2pem(result.peercert))
# as ssl proxy verification failed, service is unavailable
self.assertEqual(result.status_code, httplib.SERVICE_UNAVAILABLE)
self.assertEqual(result.status_code, six.moves.http_client.SERVICE_UNAVAILABLE)
def test_health_check_failover_url_ssl_proxy_missing(self):
parameter_dict = self.assertSlaveBase(
......@@ -7093,7 +7219,7 @@ backend _health-check-default-http
self.backend_url + slave_parameter_dict[
'health-check-http-path'].strip('/'),
headers={'X-Reply-Status-Code': '502'})
self.assertEqual(result.status_code, httplib.CREATED)
self.assertEqual(result.status_code, six.moves.http_client.CREATED)
time.sleep(3) # > health-check-timeout + health-check-interval
......@@ -7105,7 +7231,7 @@ backend _health-check-default-http
der2pem(result.peercert))
# as ssl proxy verification failed, service is unavailable
self.assertEqual(result.status_code, httplib.SERVICE_UNAVAILABLE)
self.assertEqual(result.status_code, six.moves.http_client.SERVICE_UNAVAILABLE)
if __name__ == '__main__':
......@@ -7120,5 +7246,5 @@ if __name__ == '__main__':
url_template = 'http://%s:%s/'
server = klass((ip, port), TestHandler)
print url_template % server.server_address[:2]
print(url_template % server.server_address[:2])
server.serve_forever()
software/slapos-master/test/test/__init__.py
View file @ 550f1417
......@@ -25,6 +25,7 @@
#
##############################################################################
from __future__ import absolute_import
import json
import os
......
software/slapos-master/test/test/test_balancer.py
View file @ 550f1417
from __future__ import absolute_import
import glob
import hashlib
import json
......@@ -8,9 +9,9 @@ import shutil
import subprocess
import tempfile
import time
import urllib
import urlparse
from BaseHTTPServer import BaseHTTPRequestHandler
import six.moves.urllib.request, six.moves.urllib.parse, six.moves.urllib.error
import six.moves.urllib.parse
from six.moves.BaseHTTPServer import BaseHTTPRequestHandler
from typing import Dict
import mock
......@@ -28,6 +29,7 @@ from slapos.testing.utils import (CrontabMixin, ManagedHTTPServer,
findFreeTCPPort)
from . import ERP5InstanceTestCase, setUpModule
from six.moves import range
setUpModule # pyflakes
......@@ -214,7 +216,7 @@ class TestLog(BalancerTestCase, CrontabMixin):
def test_access_log_format(self):
# type: () -> None
requests.get(
urlparse.urljoin(self.default_balancer_url, '/url_path'),
six.moves.urllib.parse.urljoin(self.default_balancer_url, '/url_path'),
verify=False,
)
time.sleep(.5) # wait a bit more until access is logged
......@@ -373,7 +375,7 @@ class TestBalancer(BalancerTestCase):
# if backend provides a "SERVERID" cookie, balancer will overwrite it with the
# backend selected by balancing algorithm
self.assertIn(
requests.get(urlparse.urljoin(self.default_balancer_url, '/set_cookie'), verify=False).cookies['SERVERID'],
requests.get(six.moves.urllib.parse.urljoin(self.default_balancer_url, '/set_cookie'), verify=False).cookies['SERVERID'],
('default-0', 'default-1'),
)
......@@ -427,18 +429,18 @@ class TestTestRunnerEntryPoints(BalancerTestCase):
)['default-test-runner-url-list']
url_0, url_1, url_2 = test_runner_url_list
self.assertEqual(
urlparse.urlparse(url_0).netloc,
urlparse.urlparse(url_1).netloc)
six.moves.urllib.parse.urlparse(url_0).netloc,
six.moves.urllib.parse.urlparse(url_1).netloc)
self.assertEqual(
urlparse.urlparse(url_0).netloc,
urlparse.urlparse(url_2).netloc)
six.moves.urllib.parse.urlparse(url_0).netloc,
six.moves.urllib.parse.urlparse(url_2).netloc)
path_0 = '/VirtualHostBase/https/{netloc}/VirtualHostRoot/_vh_unit_test_0/something'.format(
netloc=urlparse.urlparse(url_0).netloc)
netloc=six.moves.urllib.parse.urlparse(url_0).netloc)
path_1 = '/VirtualHostBase/https/{netloc}/VirtualHostRoot/_vh_unit_test_1/something'.format(
netloc=urlparse.urlparse(url_0).netloc)
netloc=six.moves.urllib.parse.urlparse(url_0).netloc)
path_2 = '/VirtualHostBase/https/{netloc}/VirtualHostRoot/_vh_unit_test_2/something'.format(
netloc=urlparse.urlparse(url_0).netloc)
netloc=six.moves.urllib.parse.urlparse(url_0).netloc)
self.assertEqual(
{
......@@ -516,7 +518,7 @@ class TestHTTP(BalancerTestCase):
session.get(self.default_balancer_url).raise_for_status()
new_conn.assert_not_called()
parsed_url = urlparse.urlparse(self.default_balancer_url)
parsed_url = six.moves.urllib.parse.urlparse(self.default_balancer_url)
# check that we have an open file for the ip connection
self.assertTrue([
c for c in psutil.Process(os.getpid()).connections()
......@@ -588,7 +590,7 @@ class TestContentEncoding(BalancerTestCase):
'application/x-font-opentype',
'application/wasm',):
resp = requests.get(
urlparse.urljoin(self.default_balancer_url, content_type),
six.moves.urllib.parse.urljoin(self.default_balancer_url, content_type),
verify=False,
headers={"Accept-Encoding": "gzip, deflate",})
self.assertEqual(resp.headers['Content-Type'], content_type)
......@@ -600,7 +602,7 @@ class TestContentEncoding(BalancerTestCase):
def test_no_gzip_encoding(self):
# type: () -> None
resp = requests.get(urlparse.urljoin(self.default_balancer_url, '/image/png'), verify=False)
resp = requests.get(six.moves.urllib.parse.urljoin(self.default_balancer_url, '/image/png'), verify=False)
self.assertNotIn('Content-Encoding', resp.headers)
self.assertEqual(resp.text, 'OK')
......
software/slapos-master/test/test/test_erp5.py
View file @ 550f1417
......@@ -25,14 +25,15 @@
#
##############################################################################
from __future__ import absolute_import
import os
import json
import glob
import urlparse
import six.moves.urllib.parse
import socket
import time
import re
import BaseHTTPServer
import six.moves.BaseHTTPServer
import multiprocessing
import subprocess
......@@ -41,6 +42,9 @@ import requests
from . import ERP5InstanceTestCase
from . import setUpModule
import six
from six.moves import map
from six.moves import range
setUpModule # pyflakes
......@@ -52,7 +56,7 @@ class TestPublishedURLIsReachableMixin(object):
# We access ERP5 trough a "virtual host", which should make
# ERP5 produce URLs using https://virtual-host-name:1234/virtual_host_root
# as base.
virtual_host_url = urlparse.urljoin(
virtual_host_url = six.moves.urllib.parse.urljoin(
base_url,
'/VirtualHostBase/https/virtual-host-name:1234/{}/VirtualHostRoot/_vh_virtual_host_root/'
.format(site_id))
......@@ -80,7 +84,7 @@ class TestPublishedURLIsReachableMixin(object):
# login page can be rendered and contain the text "ERP5"
r = session.get(
urlparse.urljoin(base_url, '{}/login_form'.format(site_id)),
six.moves.urllib.parse.urljoin(base_url, '{}/login_form'.format(site_id)),
verify=verify,
allow_redirects=False,
)
......@@ -172,7 +176,7 @@ class TestApacheBalancerPorts(ERP5InstanceTestCase):
}
def checkValidHTTPSURL(self, url):
parsed = urlparse.urlparse(url)
parsed = six.moves.urllib.parse.urlparse(url)
self.assertEqual(parsed.scheme, 'https')
self.assertTrue(parsed.hostname)
self.assertTrue(parsed.port)
......@@ -291,7 +295,7 @@ class TestZopeNodeParameterOverride(ERP5InstanceTestCase, TestPublishedURLIsReac
storage["storage"] = "root"
storage["server"] = zeo_addr
with open('%s/etc/zope-%s.conf' % (partition, zope)) as f:
conf = map(str.strip, f.readlines())
conf = list(map(str.strip, f.readlines()))
i = conf.index("<zodb_db root>") + 1
conf = iter(conf[i:conf.index("</zodb_db>", i)])
for line in conf:
......@@ -300,23 +304,23 @@ class TestZopeNodeParameterOverride(ERP5InstanceTestCase, TestPublishedURLIsReac
if line == '</zeoclient>':
break
checkParameter(line, storage)
for k, v in storage.iteritems():
for k, v in six.iteritems(storage):
self.assertIsNone(v, k)
del storage
else:
checkParameter(line, zodb)
for k, v in zodb.iteritems():
for k, v in six.iteritems(zodb):
self.assertIsNone(v, k)
partition = self.getComputerPartitionPath('zope-a')
for zope in xrange(3):
for zope in range(3):
checkConf({
"cache-size-bytes": "20MB",
}, {
"cache-size": "50MB",
})
partition = self.getComputerPartitionPath('zope-bb')
for zope in xrange(5):
for zope in range(5):
checkConf({
"cache-size-bytes": "500MB" if zope else 1<<20,
}, {
......@@ -336,7 +340,7 @@ def popenCommunicate(command_list, input_=None, **kwargs):
return result
class TestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
class TestHandler(six.moves.BaseHTTPServer.BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
......@@ -464,7 +468,7 @@ class TestDeploymentScriptInstantiation(ERP5InstanceTestCase):
ip, port = re.search(
r'.*http:\/\/(.*):(\d*)\/.*', portal_slap_line).groups()
port = int(port)
server = BaseHTTPServer.HTTPServer((ip, port), TestHandler)
server = six.moves.BaseHTTPServer.HTTPServer((ip, port), TestHandler)
server_process = multiprocessing.Process(
target=server.serve_forever, name='HTTPServer')
server_process.start()
......
software/slapos-master/test/test/test_mariadb.py
View file @ 550f1417
......@@ -26,10 +26,11 @@
#
##############################################################################
from __future__ import absolute_import
import os
import json
import glob
import urlparse
import six.moves.urllib.parse
import socket
import sys
import time
......@@ -80,7 +81,7 @@ class MariaDBTestCase(ERP5InstanceTestCase):
def getDatabaseConnection(self):
connection_parameter_dict = json.loads(
self.computer_partition.getConnectionParameterDict()['_'])
db_url = urlparse.urlparse(connection_parameter_dict['database-list'][0])
db_url = six.moves.urllib.parse.urlparse(connection_parameter_dict['database-list'][0])
self.assertEqual('mysql', db_url.scheme)
self.assertTrue(db_url.path.startswith('/'))
......
software/slapos-master/test/test/test_wcfs.py
View file @ 550f1417
......@@ -16,6 +16,7 @@
# See COPYING file for full licensing terms.
# See https://www.nexedi.com/licensing for rationale and options.
from __future__ import absolute_import
import json
import os.path
import unittest
......
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
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7