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

caddy-frontend: Test cases

Basic slave configuration is covered.

Software release can be selected in order to test apache-frontend.

Assert that correct certificates are served.

Use forcediphttpsadapter for SNI.

Make it compatbile with apache-frontend SR.

Implement local server to test against.

Improve test speed radically by requesting all slaves at once.

Use method list hooks for own setup and teardown.
parent 7abe516b
Tests for ProFTPd software release
##############################################################################
#
# Copyright (c) 2018 Nexedi SA and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from setuptools import setup, find_packages
import glob
import os
version = '0.0.1.dev0'
name = 'slapos.test.caddy-frontend'
long_description = open("README.md").read()
setup(name=name,
version=version,
description="Test for SlapOS' ProFTPd",
long_description=long_description,
long_description_content_type='text/markdown',
maintainer="Nexedi",
maintainer_email="info@nexedi.com",
url="https://lab.nexedi.com/nexedi/slapos",
packages=find_packages(),
install_requires=[
'slapos.core',
'erp5.util',
'requests',
'forcediphttpsadapter',
],
zip_safe=True,
test_suite='test',
)
-----BEGIN CERTIFICATE-----
MIIDnTCCAoWgAwIBAgIJAP0hc6dbc1GcMA0GCSqGSIb3DQEBCwUAMGQxCzAJBgNV
BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBX
aWRnaXRzIFB0eSBMdGQxHTAbBgNVBAMMFHNwZWNpZmljLmV4YW1wbGUuY29tMCAX
DTE4MDQxNzExNDIyM1oYDzIxMTgwMzI0MTE0MjIzWjBkMQswCQYDVQQGEwJBVTET
MBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50ZXJuZXQgV2lkZ2l0cyBQ
dHkgTHRkMR0wGwYDVQQDDBRzcGVjaWZpYy5leGFtcGxlLmNvbTCCASIwDQYJKoZI
hvcNAQEBBQADggEPADCCAQoCggEBALBHf15WA3X7iZR8o588CX4WoQrGpAOT6thH
aiJpcoWC4d7SkNBt5iDNffvcw33oR6SJ6ynVLW6k0a1hmWyJzMX/3T8Nu4MI9Rig
CG+g++QFoXVwip8w/1UImpX/nJnwei7l65JGabAR2XnlOCj2TpPCEvaOOqGu88AW
JIeSHUnVyFOrgSrkX81ZWczQZ9Q59LbaR5J96haSt88b+x4w/UE1ZT8uL1O//KXD
EhKEe70ENJYuWVGeFEQNzGhAhKskxfvtPRjy+hX44/nC+fpCaCIvGRxs7TMYOpr4
eRvdUNvsNcSamZQzvY/NrWUa4icLIO//k+BAo4z63E223KG7BmsCAwEAAaNQME4w
HQYDVR0OBBYEFHQcAIO3xKIwIUsaVdl6jBAMCKj/MB8GA1UdIwQYMBaAFHQcAIO3
xKIwIUsaVdl6jBAMCKj/MAwGA1UdEwQFMAMBAf8wDQYJKoZIhvcNAQELBQADggEB
AF1N1cRplgXvxoi5faVuNEEoVY+VzrNVBC6g2y58MlVihzBwSHZq98rFPzi/GW9F
hLLdZHEl9HM0cxaMqsvhag75uPWliisYU1TLLR8wTrDKpSZZ744X3a3hB0yP+PR/
6SRunBEpnxrPtw1uQw1+dD4n413uVD1XJtpJpk+5/en1psMCSqO5MKE5LRhaQ8V3
psoA5HRThCl7AXvfyoNKu8RMxFtnhH/JlAzujcbA6YxNtKkjyRU6o/1KniCJYvPp
hEHLO1iBOm5xlL1eVwV6TAbbAsI93utIN4v8KzHDFMczCwCe9VuTLhy7OI/+G/ck
10Jvsq51udlC0WiO4SMt1Wk=
-----END CERTIFICATE-----
-----BEGIN PRIVATE KEY-----
MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCwR39eVgN1+4mU
fKOfPAl+FqEKxqQDk+rYR2oiaXKFguHe0pDQbeYgzX373MN96Eekiesp1S1upNGt
YZlsiczF/90/DbuDCPUYoAhvoPvkBaF1cIqfMP9VCJqV/5yZ8Hou5euSRmmwEdl5
5Tgo9k6TwhL2jjqhrvPAFiSHkh1J1chTq4Eq5F/NWVnM0GfUOfS22keSfeoWkrfP
G/seMP1BNWU/Li9Tv/ylwxIShHu9BDSWLllRnhREDcxoQISrJMX77T0Y8voV+OP5
wvn6QmgiLxkcbO0zGDqa+Hkb3VDb7DXEmpmUM72Pza1lGuInCyDv/5PgQKOM+txN
ttyhuwZrAgMBAAECggEAUQ2u/Ptjv/4WtpXQzkjWvIk/JytLAvIii6vam1QM+mLB
5KjBOXUu2WpjQUmzAdOqDXf1D1kb+MIEP0qfNw7tfyU03+c6jvexas860ckdySzl
npBOzP/Eh57bJZlQsl+mbqX9eVazBehPLQM+nWcTnlEgUwn2350IKwyDY2rF2MqV
aBjO0xNYyaGKcuHXWOqNvfFWaXTrwgSHBhx/ySIBT4loqHekirIoShQ97/Zkf4XC
iKWAKZrzTKcsCT3OqRK1laXXGmDDh3o2JM5W21fFlX1OvwlQwsjMBO+kSaDCvd6D
a4E65Bj/hgSm8CA/Hc3CT8VSBKX3g7mXlxOa8B32AQKBgQDq5Ad1wI3N8XD5KPVt
KHIiFPBu2r7gw0PkO2lySffUcamGh1hz8yuZaBaAGCAJ+xOCgmJIvAIQ0QlFj9if
mk7mH5DGwo5pNRMl06MB9PWynsZeq/BBrkYclcmFd/dREk7eOVyMjCKtWHhUvPTn
3jqH2qA5ou4I2xB+HyOR0vSegQKBgQDAHwjk3UMmhR21+ysjw+iGQTW1FNPOSp+K
cXcM+CkZcB68kVTMIJYhHF4fqZTY4BRki4OKkZttmBIQAiLycm+RQihlaa2x7Gtc
hz1zYF1nMS7jwg8mMP32sFnx6dLUl3bUdb5YShOmL2FVbAZhmtsY6LlNrwIRDip/
NNFcATaG6wKBgQDRWqR7gz/gio2SxbrZq+7uCZOyei0ez2s3hOya/Opy40uhqAhi
cB4wAOHtEZOc0be13tgS099Tdby5z1fYCQTBfuje/pFHnBKQ6GylVduXdISB5OWi
5N2ApNCUsTQxggvhbWbbrhtKPUwi2re94ghxBZ9BP8LBCd9NShu94TQ+AQKBgGTw
qVqdygc2D2sqwPq0lrnbDx3KCJKtcG9ruBIPXICqAOwwK25ofJh+8BEQp0zfRreg
+YbXg+Mr7+mLRdM+PtICAXlu8by2q53czHOPIOPz2l/9+LmnKffvpgJNSOE7h0qP
Zme62+IHqMsZy6yCkTTgKWPgXGSbncZh+NGvLXfrAoGAR+8uupHnBz3guFH30Pdy
lSCL2asJUKRaQ4jP+ZEj2NLFXmtq8lGD3uzWOAqlfEO65tG97hTHR9JlHHAk2xRE
z2Akxz902nV5Y65ms5J28DbnAtzCzVWBnUht8SZMBZcqKpsGJVOBG43K4hVaiwkn
DZugTWfeMx5sg5hrG4GC8TY=
-----END PRIVATE KEY-----
##############################################################################
#
# Copyright (c) 2018 Nexedi SA and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
import os
import subprocess
import requests
import json
import multiprocessing
from BaseHTTPServer import HTTPServer
from BaseHTTPServer import BaseHTTPRequestHandler
from forcediphttpsadapter.adapters import ForcedIPHTTPSAdapter
import utils
# for development: debugging logs and install Ctrl+C handler
if os.environ.get('DEBUG'):
import logging
logging.basicConfig(level=logging.DEBUG)
import unittest
unittest.installHandler()
class HttpFrontendTestCase(utils.SlapOSInstanceTestCase):
@classmethod
def getSoftwareURLList(cls):
default_sr = os.path.abspath(
os.path.join(os.path.dirname(__file__), '..', 'software.cfg'))
return (os.environ.get('TEST_SR', default_sr), )
def assertKeyWithPop(self, key, d):
self.assertTrue(key in d, 'Key %r is missing' % (key,))
d.pop(key)
class TestMasterRequest(HttpFrontendTestCase):
def test_instance_parameter_default_request(self):
parameter_dict = self.computer_partition.getConnectionParameterDict()
self.assertKeyWithPop('monitor-setup-url', parameter_dict)
self.assertEqual(
{
'monitor-base-url': None,
'domain': 'None',
'accepted-slave-amount': '0',
'rejected-slave-amount': '0',
'slave-amount': '0',
'rejected-slave-list': '[]'},
parameter_dict
)
class TestMasterRequestDomain(HttpFrontendTestCase):
@classmethod
def getInstanceParmeterDict(cls):
return {'domain': 'example.com'}
def test_instance_parameter_default_request(self):
parameter_dict = self.computer_partition.getConnectionParameterDict()
self.assertKeyWithPop('monitor-setup-url', parameter_dict)
self.assertEqual(
{
'monitor-base-url': None,
'domain': 'example.com',
'accepted-slave-amount': '0',
'rejected-slave-amount': '0',
'slave-amount': '0',
'rejected-slave-list': '[]'},
parameter_dict
)
def der2pem(der):
certificate, error = subprocess.Popen(
'openssl x509 -inform der'.split(), stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.PIPE
).communicate(der)
if error:
raise ValueError(error)
return certificate
class TestHandler(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header("Content-type", "text/json")
self.end_headers()
response = {
'Path': self.path,
'Incoming Headers': self.headers.dict
}
self.wfile.write(json.dumps(response, indent=2))
class SlaveHttpFrontendTestCase(HttpFrontendTestCase):
@classmethod
def startServerProcess(cls):
ip = os.environ.get('LOCAL_IPV4', '127.0.1.1')
server = HTTPServer(
(ip, utils.findFreeTCPPort(ip)),
TestHandler)
cls.backend_url = 'http://%s:%s' % server.server_address
cls.server_process = multiprocessing.Process(target=server.serve_forever)
cls.server_process.start()
@classmethod
def stopServerProcess(cls):
cls.server_process.terminate()
@classmethod
def setUpSlaves(cls):
cls.slave_connection_parameter_dict_dict = {}
request = cls.slapos_controler.slap.registerOpenOrder().request
for slave_reference, partition_parameter_kw in cls\
.getSlaveParameterDictDict().items():
slave_instance = request(
software_release=cls.software_url_list[0],
partition_reference=slave_reference,
partition_parameter_kw=partition_parameter_kw,
shared=True
)
cls.runComputerPartition()
for slave_reference, partition_parameter_kw in cls\
.getSlaveParameterDictDict().items():
slave_instance = request(
software_release=cls.software_url_list[0],
partition_reference=slave_reference,
partition_parameter_kw=partition_parameter_kw,
shared=True
)
cls.slave_connection_parameter_dict_dict[slave_reference] = \
slave_instance.getConnectionParameterDict()
@classmethod
def getBeforeSetUpClassMethodList(cls):
return [cls.startServerProcess]
@classmethod
def getAfterSetUpClassMethodList(cls):
return [cls.setUpSlaves]
@classmethod
def getTearDownClassMethodList(cls):
return [cls.stopServerProcess]
def fakeHTTPSResult(self, domain, real_ip, path, port=4443, headers=None):
session = requests.Session()
session.mount(
'https://%s:%s' % (domain, port),
ForcedIPHTTPSAdapter(
dest_ip=real_ip))
return session.get(
'https://%s:%s/%s' % (domain, port, path),
verify=False,
allow_redirects=False,
headers=headers
)
def fakeHTTPResult(self, domain, real_ip, path, port=8080, headers=None):
if headers is None:
headers = {}
headers['Host'] = domain
return requests.get(
'http://%s:%s/%s' % (real_ip, port, path),
headers=headers,
allow_redirects=False,
)
def patchRequests(self):
HTTPResponse = requests.packages.urllib3.response.HTTPResponse
HTTPResponse.orig__init__ = HTTPResponse.__init__
def new_HTTPResponse__init__(self, *args, **kwargs):
self.orig__init__(*args, **kwargs)
try:
self.peercert = self._connection.sock.getpeercert(binary_form=True)
except AttributeError:
pass
HTTPResponse.__init__ = new_HTTPResponse__init__
HTTPAdapter = requests.adapters.HTTPAdapter
HTTPAdapter.orig_build_response = HTTPAdapter.build_response
def new_HTTPAdapter_build_response(self, request, resp):
response = self.orig_build_response(request, resp)
try:
response.peercert = resp.peercert
except AttributeError:
pass
return response
HTTPAdapter.build_response = new_HTTPAdapter_build_response
def unpatchRequests(self):
HTTPResponse = requests.packages.urllib3.response.HTTPResponse
if getattr(HTTPResponse, 'orig__init__', None) is not None:
HTTPResponse.__init__ = HTTPResponse.orig__init__
del(HTTPResponse.orig__init__)
HTTPAdapter = requests.adapters.HTTPAdapter
if getattr(HTTPAdapter, 'orig_build_response', None) is not None:
HTTPAdapter.build_response = HTTPAdapter.orig_build_response
del(HTTPAdapter.orig_build_response)
def setUp(self):
# patch requests in order to being able to extract SSL certs
self.patchRequests()
def tearDown(self):
self.unpatchRequests()
class TestMasterSlaveDomains(SlaveHttpFrontendTestCase):
@classmethod
def getInstanceParmeterDict(cls):
local_ipv4 = os.environ.get('LOCAL_IPV4', '127.0.1.1')
return {
'domain': 'example.com',
'nginx-domain': 'nginx.example.com',
'public-ipv4': local_ipv4,
'apache-certificate': open('wildcard.example.com.crt').read(),
'apache-key': open('wildcard.example.com.key').read(),
}
@classmethod
def getSlaveParameterDictDict(cls):
return {
'default-slave': {
'url': cls.backend_url,
},
'server-alias-slave': {
'url': cls.backend_url,
'server-alias': 'alias1.example.com alias2.example.com',
},
'default-slave-https-only': {
'url': cls.backend_url,
'https-only': True,
},
'owndomain-slave': {
'url': cls.backend_url,
'custom_domain': 'owndomain.example.com',
},
'specificdomain-slave': {
'url': cls.backend_url,
'custom_domain': 'specific.example.com',
'ssl_crt': open('specific.example.com.crt').read(),
'ssl_key': open('specific.example.com.key').read(),
},
'zope-type-slave': {
'url': cls.backend_url,
'type': 'zope',
},
'zope-type-path-slave': {
'url': cls.backend_url,
'type': 'zope',
'path': 'path',
},
'zope-type-default-path-slave': {
'url': cls.backend_url,
'type': 'zope',
'default-path': 'default-path',
},
'notebook-type-slave': {
'url': cls.backend_url,
'type': 'notebook',
},
'eventsource-type-slave': {
'url': cls.backend_url,
'type': 'eventsource',
},
'redirect-type-slave': {
'url': cls.backend_url,
'type': 'redirect',
},
'enable-cache-slave': {
'url': cls.backend_url,
'enable_cache': True,
},
'enable-cache-disable-no-cache-request-slave': {
'url': cls.backend_url,
'enable_cache': True,
'disable-no-cache-request': True,
},
'enable-cache-disable-via-header-slave': {
'url': cls.backend_url,
'enable_cache': True,
'disable-via-header': True,
},
}
def test_slave_default_domain(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'default-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'defaultslave.example.com',
'replication_number': '1',
'url': 'http://defaultslave.example.com',
'site_url': 'http://defaultslave.example.com',
'secure_access': 'https://defaultslave.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/test-path'
)
result_http = self.fakeHTTPResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
result_http.json()['Path'],
'/test-path'
)
def test_zope_type_slave_path(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'zope-type-path-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'zopetypepathslave.example.com',
'replication_number': '1',
'url': 'http://zopetypepathslave.example.com',
'site_url': 'http://zopetypepathslave.example.com',
'secure_access': 'https://zopetypepathslave.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/VirtualHostBase/https//'
'zopetypepathslave.example.com:443/path/VirtualHostRoot/test-path'
)
def test_zope_type_default_path_slave(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'zope-type-default-path-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'zopetypedefaultpathslave.example.com',
'replication_number': '1',
'url': 'http://zopetypedefaultpathslave.example.com',
'site_url': 'http://zopetypedefaultpathslave.example.com',
'secure_access': 'https://zopetypedefaultpathslave.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], '')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.headers['Location'],
'https://zopetypedefaultpathslave.example.com:4443/default-path'
)
def test_slave_server_alias(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'server-alias-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'serveraliasslave.example.com',
'replication_number': '1',
'url': 'http://serveraliasslave.example.com',
'site_url': 'http://serveraliasslave.example.com',
'secure_access': 'https://serveraliasslave.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/test-path'
)
result = self.fakeHTTPSResult(
'alias1.example.com', parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/test-path'
)
result = self.fakeHTTPSResult(
'alias2.example.com', parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/test-path'
)
def test_slave_https_only_default_domain(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'default-slave-https-only']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'defaultslavehttpsonly.example.com',
'replication_number': '1',
'url': 'http://defaultslavehttpsonly.example.com',
'site_url': 'http://defaultslavehttpsonly.example.com',
'secure_access': 'https://defaultslavehttpsonly.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/test-path'
)
result_http = self.fakeHTTPResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
result_http.headers['Location'],
'https://defaultslavehttpsonly.example.com/test-path'
)
def test_slave_own_domain(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'owndomain-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'owndomain.example.com',
'replication_number': '1',
'url': 'http://owndomain.example.com',
'site_url': 'http://owndomain.example.com',
'secure_access': 'https://owndomain.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/test-path'
)
def test_slave_own_domain_own_certificate(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'specificdomain-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'specific.example.com',
'replication_number': '1',
'url': 'http://specific.example.com',
'site_url': 'http://specific.example.com',
'secure_access': 'https://specific.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('specific.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/test-path'
)
def test_slave_zope_type(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'zope-type-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'zopetypeslave.example.com',
'replication_number': '1',
'url': 'http://zopetypeslave.example.com',
'site_url': 'http://zopetypeslave.example.com',
'secure_access': 'https://zopetypeslave.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/VirtualHostBase/https//zopetypeslave.example.com:443/'
'/VirtualHostRoot/test-path'
)
def test_slave_notebook_type(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'notebook-type-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'notebooktypeslave.nginx.example.com',
'replication_number': '1',
'url': 'http://notebooktypeslave.nginx.example.com',
'site_url': 'http://notebooktypeslave.nginx.example.com',
'secure_access': 'https://notebooktypeslave.nginx.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path',
9443)
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/test-path'
)
def test_slave_eventsource_type(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'eventsource-type-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'eventsourcetypeslave.nginx.example.com',
'replication_number': '1',
'url': 'http://eventsourcetypeslave.nginx.example.com',
'site_url': 'http://eventsourcetypeslave.nginx.example.com',
'secure_access': 'https://eventsourcetypeslave.nginx.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'pub',
9443)
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.content,
''
)
headers = result.headers.copy()
self.assertKeyWithPop('Expires', headers)
self.assertKeyWithPop('Date', headers)
self.assertEqual(
headers,
{
'X-Nginx-PushStream-Explain': 'No channel id provided.',
'Content-Length': '0',
'Cache-Control': 'no-cache, no-store, must-revalidate',
'Connection': 'keep-alive',
'Server': 'nginx'
}
)
def test_slave_redirect_type(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'redirect-type-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'redirecttypeslave.example.com',
'replication_number': '1',
'url': 'http://redirecttypeslave.example.com',
'site_url': 'http://redirecttypeslave.example.com',
'secure_access': 'https://redirecttypeslave.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.headers['Location'],
'%s/test-path' % (self.backend_url,)
)
def test_slave_ssl_proxy_verify(self):
raise NotImplementedError(self.id())
def test_slave_ssl_proxy_verify_with_ssl_proxy_ca_crt(self):
raise NotImplementedError(self.id())
def test_slave_monitor_ipv6_rest(self):
raise NotImplementedError(self.id())
def test_slave_monitor_ipv4_rest(self):
raise NotImplementedError(self.id())
def test_slave_re6st_optimal_test(self):
raise NotImplementedError(self.id())
def test_slave_enable_cache(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'enable-cache-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'enablecacheslave.example.com',
'replication_number': '1',
'url': 'http://enablecacheslave.example.com',
'site_url': 'http://enablecacheslave.example.com',
'secure_access': 'https://enablecacheslave.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/test-path'
)
headers = result.headers.copy()
self.assertKeyWithPop('Via', headers)
self.assertKeyWithPop('Server', headers)
self.assertKeyWithPop('Date', headers)
# drop keys appearing randomly in headers
headers.pop('Transfer-Encoding', None)
headers.pop('Content-Length', None)
self.assertEqual(
headers,
{'Age': '0', 'Keep-Alive': 'timeout=5, max=100',
'Connection': 'Keep-Alive', 'Content-type': 'text/json'}
)
def test_slave_enable_cache_disable_no_cache_request(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'enable-cache-disable-no-cache-request-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'enablecachedisablenocacherequestslave.example.com',
'replication_number': '1',
'url': 'http://enablecachedisablenocacherequestslave.example.com',
'site_url': 'http://enablecachedisablenocacherequestslave.example.com',
'secure_access':
'https://enablecachedisablenocacherequestslave.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path',
headers=dict(Pragma='no-cache'))
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/test-path'
)
headers = result.headers.copy()
self.assertKeyWithPop('Via', headers)
self.assertKeyWithPop('Server', headers)
self.assertKeyWithPop('Date', headers)
# drop keys appearing randomly in headers
headers.pop('Transfer-Encoding', None)
headers.pop('Content-Length', None)
self.assertEqual(
headers,
{'Age': '0', 'Keep-Alive': 'timeout=5, max=100',
'Connection': 'Keep-Alive', 'Content-type': 'text/json'}
)
self.assertFalse('pragma' in result.json()['Incoming Headers'].keys())
def test_slave_enable_cache_disable_via_header(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'enable-cache-disable-via-header-slave']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'enablecachedisableviaheaderslave.example.com',
'replication_number': '1',
'url': 'http://enablecachedisableviaheaderslave.example.com',
'site_url': 'http://enablecachedisableviaheaderslave.example.com',
'secure_access':
'https://enablecachedisableviaheaderslave.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/test-path'
)
headers = result.headers.copy()
self.assertKeyWithPop('Server', headers)
self.assertKeyWithPop('Date', headers)
# drop keys appearing randomly in headers
headers.pop('Transfer-Encoding', None)
headers.pop('Content-Length', None)
self.assertEqual(
headers,
{'Age': '0', 'Keep-Alive': 'timeout=5, max=100',
'Connection': 'Keep-Alive', 'Content-type': 'text/json'}
)
def test_slave_enable_http2(self):
raise NotImplementedError(self.id())
def test_slave_enable_cache_prefer_gzip_encoding(self):
raise NotImplementedError(self.id())
def test_slave_disabled_cookie_list(self):
raise NotImplementedError(self.id())
def test_slave_apache_custom_http(self):
raise NotImplementedError(self.id())
def test_slave_apache_custom_https(self):
raise NotImplementedError(self.id())
def test_slave_virtualhostroot_http_port(self):
raise NotImplementedError(self.id())
def test_slave_virtualhostroot_https_port(self):
raise NotImplementedError(self.id())
def test_slave_https_port(self):
raise NotImplementedError(self.id())
class TestMasterReplicateSlave(SlaveHttpFrontendTestCase):
@classmethod
def getInstanceParmeterDict(cls):
local_ipv4 = os.environ.get('LOCAL_IPV4', '127.0.1.1')
return {
'domain': 'example.com',
'nginx-domain': 'nginx.example.com',
'public-ipv4': local_ipv4,
'apache-certificate': open('wildcard.example.com.crt').read(),
'apache-key': open('wildcard.example.com.key').read(),
'-frontend-quantity': 2,
'-sla-2-computer_guid': 'slapos.test',
'-frontend-2-state': 'stopped',
}
@classmethod
def getSlaveParameterDictDict(cls):
return {
'default-slave-replicate': {
'url': cls.backend_url,
},
}
def test_master_replicate_default_slave(self):
parameter_dict = self.slave_connection_parameter_dict_dict[
'default-slave-replicate']
self.assertKeyWithPop('log-access-url', parameter_dict)
self.assertEqual(
parameter_dict,
{
'domain': 'defaultslavereplicate.example.com',
'replication_number': '2',
'url': 'http://defaultslavereplicate.example.com',
'site_url': 'http://defaultslavereplicate.example.com',
'secure_access': 'https://defaultslavereplicate.example.com',
'public-ipv4': '127.0.1.1',
}
)
result = self.fakeHTTPSResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
der2pem(result.peercert),
open('wildcard.example.com.crt').read())
self.assertEqual(
result.json()['Path'],
'/test-path'
)
result_http = self.fakeHTTPResult(
parameter_dict['domain'], parameter_dict['public-ipv4'], 'test-path')
self.assertEqual(
result_http.json()['Path'],
'/test-path'
)
# prove 2nd frontend by inspection of the instance
slave_configuration_name = '_default-slave-replicate.conf'
slave_configuration_file_list = [
'/'.join([f[0], slave_configuration_name]) for f in [
q for q in os.walk(self.instance_path)
if slave_configuration_name in q[2]
]
]
self.assertEqual(
2, len(slave_configuration_file_list), slave_configuration_file_list)
-----BEGIN CERTIFICATE-----
MIIDjzCCAnegAwIBAgIJAP4QzV9d5d8KMA0GCSqGSIb3DQEBCwUAMF0xCzAJBgNV
BAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBX
aWRnaXRzIFB0eSBMdGQxFjAUBgNVBAMMDSouZXhhbXBsZS5jb20wIBcNMTgwNDE3
MTE0MjAxWhgPMjExODAzMjQxMTQyMDFaMF0xCzAJBgNVBAYTAkFVMRMwEQYDVQQI
DApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRlcm5ldCBXaWRnaXRzIFB0eSBMdGQx
FjAUBgNVBAMMDSouZXhhbXBsZS5jb20wggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAw
ggEKAoIBAQC0MNnojTq6Yp60ecvP5lZGa6owZp64sumaT66JInTFJcSV/ls0AbPN
KPTcNRgbFLIuv/3Xhgi6YAxIz2PXdCDPj1fp4rVIijhpedaWZaIOeTqSGEg+5/iL
4UTxm166NgCU1zp0QsmU8MXJFv4YxTZsF9LlGFcP6uUQ9sY98yv+hErCxRXo+dXd
GGG4LTVLGPeRYNX2JVD5BxoNkL5/3IMylrFvI8aRFTKVn1P8UPJ5K9du3E0wHJY6
MkstZX3xKcGKSn2w1zIaq/NcrbLXvp6eZauLtMpZ9vW+CtYV3diOm2TiMvQIyCRM
4mWbaf76bpGiz9/+11w6U9n6zMzKTq+5AgMBAAGjUDBOMB0GA1UdDgQWBBT/Kswu
BKM3GXaIWvQ8Vs1xhs/1vzAfBgNVHSMEGDAWgBT/KswuBKM3GXaIWvQ8Vs1xhs/1
vzAMBgNVHRMEBTADAQH/MA0GCSqGSIb3DQEBCwUAA4IBAQAeeV7VIqbNL6Xe0I5y
fl44ogBY0zoYJH+p7HCebazZKepy7f+EeWv8B3dHODrilSYCnsOqhta8mMSMrgsT
TFByr/8NN/4UFe9FeKnxD9Sun0EetESkaeMqAGCK921lT0eU4IhA0JLRj16HomuK
I/JBFwUXFn6vVypw5R41zwnikv+EB0fSC4biSXWXUarGqlRCF7o00CPKqAaTE0yX
H4Yz3lR/Vuce7uFvkIY+F4vXtq1sy/tb4QBWlDS6t7cmeBGcvaQ8mlndA9T+us0l
/wb9mzdTcwkrkM2kk++GnS0NCAQfOalF3x8wG6j8DrU2dQ1NXh205c/zo0BljQr8
P84v
-----END CERTIFICATE-----
-----BEGIN PRIVATE KEY-----
MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC0MNnojTq6Yp60
ecvP5lZGa6owZp64sumaT66JInTFJcSV/ls0AbPNKPTcNRgbFLIuv/3Xhgi6YAxI
z2PXdCDPj1fp4rVIijhpedaWZaIOeTqSGEg+5/iL4UTxm166NgCU1zp0QsmU8MXJ
Fv4YxTZsF9LlGFcP6uUQ9sY98yv+hErCxRXo+dXdGGG4LTVLGPeRYNX2JVD5BxoN
kL5/3IMylrFvI8aRFTKVn1P8UPJ5K9du3E0wHJY6MkstZX3xKcGKSn2w1zIaq/Nc
rbLXvp6eZauLtMpZ9vW+CtYV3diOm2TiMvQIyCRM4mWbaf76bpGiz9/+11w6U9n6
zMzKTq+5AgMBAAECggEAR9n2+pFearYqnMK4b9Vkb7485gH1pqbJGdxON6bCs16F
Dl6X1ZwcK2H6idiuHRZamuO5//gVgOQN4fa41FAdSUbagowBR8S+C+kmlWA/h8/1
eA4wuMzdQkH4sPMIie5Auxk72OJM6ZQ8+huuBQiW0/GICgxzowhCgUo18LwHvfwt
XWLHplRFczgM4WEu+HVx1W9y/irOhEUEzwNfXdDTVE4bnwzb24qR0CRIdp3gc8Os
JoUFMgbV3qfJLmzqURLIlVc3Y9FFbWrr5zhzTx2vvoT1j83e56qKHbKanr2R6NQZ
UOsCgDvzrc9WqKE9/hwvxvPlx/L3UWuhPMpljZ8gBQKBgQDX4Qo27l8KPk9fXA18
SrRcAAGN+moc+Y0T9WbcbXCJLLU6TO3eJqW9H3Kwb80O/zycUBFK/dIOwZZYgsuQ
oXFG4ykSXFY/JGTEBIkvQ7LI/7t5mn0xlw3tqB7xuMO5QEyZMQFi0xP5h1B5coiO
Wv9T1xit4FHX2EN/CJLwlzzl1wKBgQDVrdvGi+j/1pGaEST1yhvrDduHrlMsF4BI
X8VaNXNXg9+GxsAA6+pRPRoHvmmz/RUr/km+JxWLHEZpFDOhIzJV4JYXv/q7KDH+
Mjf7uyRshBnR+QU38RqkENIJhEZJSa8ZYmkgU9yCDozkfVzYKdxUy6Z0+o2Omzls
cpZP1PFE7wKBgQCGXZx0+kMPZh8TFIGURg8iYCKXkzBu3miP7qNaOYfc6YXXRsCb
D+UC5MsGxF+WoQjBphhNW9RduOJyLt6zI7kUzRjoQ66u2GEbnFMipvlln765fo3D
yugxbv3rp/uylzHV+6mIMCbzneRZ4w7ZxAu9zFihCMkIFqRUMir7MrcFuwKBgFbK
88ZF9jJU+XdXF2gu3AAx9MW77VSvhw/ets7ZfyxBCH46JKs7KEYvR291zIGrfvoL
o/B09681oPP1nLMLFNsFCnJDLJjwzr2tsEez0CuzzLkZKSF78ZJKssXi0JncMB9j
dcgHyD2bo2b79MZo2nIm9kn1q6INMtn2AVAT8pxJAoGAZfysuUMc0bdKVi41SD0A
Ej7qX+qbXX+aA4OkTqJP3QxssD6CuZZwUKW7qj8AoOcLKsmwdjetetny46OMEXzN
5Fk3ahCD1cycFGBawNY10x5Xy+4iApzPiO1Ujd3t7FAIrLPWFwzmLIL8pxUNuHEn
OC9RrI4tzsvWHhMaLKVCh0g=
-----END PRIVATE KEY-----
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