157 lines
6.7 KiB
Python
157 lines
6.7 KiB
Python
import os
|
|
import time
|
|
|
|
import docker
|
|
import pytest
|
|
|
|
from functions import Functions
|
|
from processor import ProcessorInterface
|
|
|
|
|
|
def _get_hydrated_object(parsed_objects, lookup_key):
|
|
hydrated_object = {}
|
|
for key in parsed_objects:
|
|
for keys in parsed_objects[key]:
|
|
if lookup_key in keys:
|
|
hydrated_object[keys] = parsed_objects[key][keys]
|
|
return hydrated_object
|
|
|
|
|
|
def _get_ip_host(parsed_objects, lookup_key):
|
|
for key in parsed_objects:
|
|
for keys in parsed_objects[key]:
|
|
if lookup_key in keys:
|
|
return key
|
|
|
|
|
|
def test_processor_docker():
|
|
try:
|
|
client = docker.from_env()
|
|
except docker.errors.DockerException:
|
|
pytest.skip("There is no docker environment")
|
|
|
|
if len(client.containers.list()) > 0:
|
|
pytest.skip("I cannot run this test with other containers running.")
|
|
|
|
container = client.containers.run("byjg/static-httpserver",
|
|
name="test_processor_docker",
|
|
detach=True,
|
|
auto_remove=True,
|
|
remove=True,
|
|
labels={
|
|
"easyhaproxy.http.port": "80",
|
|
"easyhaproxy.http.localport": "8080",
|
|
"easyhaproxy.http.host": "host1.local",
|
|
|
|
"easyhaproxy.http2.port": "90",
|
|
"easyhaproxy.http2.localport": "9000",
|
|
"easyhaproxy.http2.host": "host2.local",
|
|
"easyhaproxy.http2.certbot": "true",
|
|
})
|
|
container2 = client.containers.run("byjg/static-httpserver",
|
|
name="test2_processor_docker",
|
|
detach=True,
|
|
auto_remove=True,
|
|
remove=True,
|
|
labels={
|
|
"easyhaproxy.ssl.port": "443",
|
|
"easyhaproxy.ssl.localport": "8080",
|
|
"easyhaproxy.ssl.host": "hostssl.local",
|
|
"easyhaproxy.ssl.sslcert": "U29tZSBQRU0gQ2VydGlmaWNhdGU="
|
|
})
|
|
try:
|
|
time.sleep(1)
|
|
|
|
os.environ['EASYHAPROXY_CERTBOT_EMAIL'] = 'docker@example.org'
|
|
|
|
static = ProcessorInterface.factory(ProcessorInterface.DOCKER)
|
|
assert static.get_certbot_hosts() is None
|
|
|
|
assert {
|
|
'easyhaproxy.http.host': 'host1.local',
|
|
'easyhaproxy.http.localport': '8080',
|
|
'easyhaproxy.http.port': '80',
|
|
'easyhaproxy.http2.host': 'host2.local',
|
|
'easyhaproxy.http2.localport': '9000',
|
|
'easyhaproxy.http2.port': '90',
|
|
'easyhaproxy.http2.certbot': 'true',
|
|
} == _get_hydrated_object(static.get_parsed_object(), "easyhaproxy.http")
|
|
assert {
|
|
'easyhaproxy.ssl.host': 'hostssl.local',
|
|
'easyhaproxy.ssl.localport': '8080',
|
|
'easyhaproxy.ssl.port': '443',
|
|
'easyhaproxy.ssl.sslcert': 'U29tZSBQRU0gQ2VydGlmaWNhdGU='
|
|
} == _get_hydrated_object(static.get_parsed_object(), "easyhaproxy.ssl.")
|
|
|
|
assert static.get_hosts() is None
|
|
assert static.get_certs() == {}
|
|
|
|
haproxy_cfg = static.get_haproxy_conf()
|
|
assert haproxy_cfg == Functions.load(os.path.join(os.path.dirname(os.path.realpath(__file__)), "./expected/docker.txt")).replace("test_processor_docker", _get_ip_host(
|
|
static.get_parsed_object(), "easyhaproxy.http")).replace("test2_processor_docker", _get_ip_host(static.get_parsed_object(), "easyhaproxy.ssl"))
|
|
|
|
assert static.get_certbot_hosts() == ['host2.local']
|
|
assert static.get_hosts() == [
|
|
'hostssl.local:443',
|
|
'host1.local:80',
|
|
'host2.local:90'
|
|
]
|
|
assert static.get_certs() == {
|
|
'hostssl.local.pem': 'Some PEM Certificate'
|
|
}
|
|
finally:
|
|
del os.environ['EASYHAPROXY_CERTBOT_EMAIL']
|
|
container.stop()
|
|
container2.stop()
|
|
|
|
|
|
def test_processor_docker_ignores_unlabeled():
|
|
try:
|
|
client = docker.from_env()
|
|
except docker.errors.DockerException:
|
|
pytest.skip("There is no docker environment")
|
|
|
|
if len(client.containers.list()) > 0:
|
|
pytest.skip("I cannot run this test with other containers running.")
|
|
|
|
network = client.networks.create("test_processor_docker_network", driver="bridge")
|
|
# The most recent container is inspected first, so the labeled one defines the network to use.
|
|
unlabeled = client.containers.run("byjg/static-httpserver",
|
|
name="test_processor_docker_unlabeled",
|
|
detach=True,
|
|
auto_remove=True,
|
|
remove=True)
|
|
time.sleep(1)
|
|
container = client.containers.run("byjg/static-httpserver",
|
|
name="test_processor_docker_labeled",
|
|
detach=True,
|
|
auto_remove=True,
|
|
remove=True,
|
|
network=network.name,
|
|
labels={
|
|
"easyhaproxy.labeled.port": "80",
|
|
"easyhaproxy.labeled.localport": "8080",
|
|
"easyhaproxy.labeled.host": "labeled.local",
|
|
})
|
|
try:
|
|
time.sleep(1)
|
|
|
|
static = ProcessorInterface.factory(ProcessorInterface.DOCKER)
|
|
|
|
# The labeled container is served through the network it already belongs to.
|
|
labeled_ip = client.containers.get(container.name).attrs["NetworkSettings"]["Networks"][network.name][
|
|
"IPAddress"]
|
|
assert _get_ip_host(static.get_parsed_object(), "easyhaproxy.labeled") == labeled_ip
|
|
|
|
# The container without any label is neither inspected nor connected to the network.
|
|
unlabeled_networks = client.containers.get(unlabeled.name).attrs["NetworkSettings"]["Networks"]
|
|
assert list(unlabeled_networks.keys()) == ["bridge"]
|
|
assert unlabeled_networks["bridge"]["IPAddress"] not in static.get_parsed_object()
|
|
finally:
|
|
container.stop()
|
|
unlabeled.stop()
|
|
time.sleep(1)
|
|
network.remove()
|
|
|
|
|
|
# test_processor_docker()
|