the-littlest-jupyterhub/tests/test_traefik.py

"""Test traefik configuration"""
import os

import pytest
import toml

from tljh import config, traefik


def test_download_traefik(tmpdir):
    traefik_bin = tmpdir.mkdir("bin").join("traefik")
    traefik.ensure_traefik_binary(str(tmpdir))
    assert traefik_bin.exists()
    # ignore higher-order permission bits, only verify ugo permissions
    assert (traefik_bin.stat().mode & 0o777) == 0o755


def _read_toml(path):
    """Read a toml file

    print config for debugging on failure
    """
    print(path)
    with open(path) as f:
        toml_cfg = f.read()
        print(toml_cfg)
        return toml.loads(toml_cfg)


def _read_static_config(state_dir):
    return _read_toml(os.path.join(state_dir, "traefik.toml"))


def _read_dynamic_config(state_dir):
    return _read_toml(os.path.join(state_dir, "rules", "dynamic.toml"))


def test_default_config(tmpdir, tljh_dir):
    state_dir = tmpdir.mkdir("state")
    traefik.ensure_traefik_config(str(state_dir))
    assert state_dir.join("traefik.toml").exists()
    os.path.join(state_dir, "traefik.toml")
    rules_dir = os.path.join(state_dir, "rules")

    cfg = _read_static_config(state_dir)
    assert cfg["api"] == {}
    assert cfg["entryPoints"] == {
        "http": {
            "address": ":80",
            "transport": {"respondingTimeouts": {"idleTimeout": "10m"}},
        },
        "auth_api": {
            "address": "localhost:8099",
        },
    }
    assert cfg["providers"] == {
        "providersThrottleDuration": "0s",
        "file": {"directory": rules_dir, "watch": True},
    }

    dynamic_config = _read_dynamic_config(state_dir)
    assert dynamic_config == {}


def test_letsencrypt_config(tljh_dir):
    state_dir = config.STATE_DIR
    config.set_config_value(config.CONFIG_FILE, "https.enabled", True)
    config.set_config_value(
        config.CONFIG_FILE, "https.letsencrypt.email", "fake@jupyter.org"
    )
    config.set_config_value(
        config.CONFIG_FILE, "https.letsencrypt.domains", ["testing.jovyan.org"]
    )
    traefik.ensure_traefik_config(str(state_dir))

    cfg = _read_static_config(state_dir)
    assert cfg["entryPoints"] == {
        "http": {
            "address": ":80",
            "http": {
                "redirections": {
                    "entryPoint": {
                        "scheme": "https",
                        "to": "https",
                    },
                },
            },
            "transport": {"respondingTimeouts": {"idleTimeout": "10m"}},
        },
        "https": {
            "address": ":443",
            "http": {"tls": {"options": "default"}},
            "transport": {"respondingTimeouts": {"idleTimeout": "10m"}},
        },
        "auth_api": {
            "address": "localhost:8099",
        },
    }
    assert "tls" not in cfg

    dynamic_config = _read_dynamic_config(state_dir)

    assert dynamic_config["tls"] == {
        "options": {"default": {"minVersion": "VersionTLS12"}},
        "stores": {
            "default": {
                "defaultGeneratedCert": {
                    "resolver": "letsencrypt",
                    "domain": {
                        "main": "testing.jovyan.org",
                        "sans": [],
                    },
                }
            }
        },
    }
    assert "certificatesResolvers" in cfg
    assert "letsencrypt" in cfg["certificatesResolvers"]

    assert cfg["certificatesResolvers"]["letsencrypt"]["acme"] == {
        "email": "fake@jupyter.org",
        "storage": "acme.json",
        "tlsChallenge": {},
    }


def test_manual_ssl_config(tljh_dir):
    state_dir = config.STATE_DIR
    config.set_config_value(config.CONFIG_FILE, "https.enabled", True)
    config.set_config_value(config.CONFIG_FILE, "https.tls.key", "/path/to/ssl.key")
    config.set_config_value(config.CONFIG_FILE, "https.tls.cert", "/path/to/ssl.cert")
    traefik.ensure_traefik_config(str(state_dir))

    cfg = _read_static_config(state_dir)

    assert cfg["entryPoints"] == {
        "http": {
            "address": ":80",
            "http": {
                "redirections": {
                    "entryPoint": {
                        "scheme": "https",
                        "to": "https",
                    },
                },
            },
            "transport": {
                "respondingTimeouts": {
                    "idleTimeout": "10m",
                }
            },
        },
        "https": {
            "address": ":443",
            "http": {"tls": {"options": "default"}},
            "transport": {"respondingTimeouts": {"idleTimeout": "10m"}},
        },
        "auth_api": {
            "address": "localhost:8099",
        },
    }
    assert "tls" not in cfg

    dynamic_config = _read_dynamic_config(state_dir)

    assert "tls" in dynamic_config

    assert dynamic_config["tls"] == {
        "options": {"default": {"minVersion": "VersionTLS12"}},
        "stores": {
            "default": {
                "defaultCertificate": {
                    "certFile": "/path/to/ssl.cert",
                    "keyFile": "/path/to/ssl.key",
                }
            }
        },
    }


def test_extra_config(tmpdir, tljh_dir):
    extra_config_dir = os.path.join(tljh_dir, config.CONFIG_DIR, "traefik_config.d")
    state_dir = tmpdir.mkdir("state")
    traefik_toml = os.path.join(state_dir, "traefik.toml")

    # Generate default config
    traefik.ensure_traefik_config(str(state_dir))

    # Read the default config
    toml_cfg = toml.load(traefik_toml)

    # Make sure the defaults are what we expect
    assert toml_cfg["log"]["level"] == "INFO"
    with pytest.raises(KeyError):
        toml_cfg["api"]["dashboard"]
    assert toml_cfg["entryPoints"]["auth_api"]["address"] == "localhost:8099"

    extra_config = {
        # modify existing value
        "log": {
            "level": "ERROR",
        },
        # add new setting
        "api": {"dashboard": True},
    }

    with open(os.path.join(extra_config_dir, "extra.toml"), "w+") as extra_config_file:
        toml.dump(extra_config, extra_config_file)

    # Merge the extra config with the defaults
    traefik.ensure_traefik_config(str(state_dir))

    # Read back the merged config
    toml_cfg = toml.load(traefik_toml)

    # Check that the defaults were updated by the extra config
    assert toml_cfg["log"]["level"] == "ERROR"
    assert toml_cfg["api"]["dashboard"] == True