"""Only a vhost Certbot already installed with the same lineage and redirect may skip `certbot install`."""
import os
from pathlib import Path
import shutil
import socket
import ssl
import subprocess
import sys
import tempfile
import threading
import unittest

import certificate_vhost
import configure_vhost
from nginx_test_fixture import LEGACY_VHOST

DOMAIN = "api-fixture-0123456789abcdef.example"
STATES = {"current", "absent", "http_only", "other_certificate", "no_redirect", "lineage_unreadable", "duplicate",
          "config_unknown", "not_served", "unreachable", "ambiguous", "unparsable", "unreadable"}
# Statements nginx loads as naming the domain, where the words up to the first `;` hold another name or none.
MISREAD = {
    "a comment holding a semicolon": f"server_name other.example # was public.example; kept\n        {DOMAIN};",
    "a double-quoted directive name": f'"server_name" {DOMAIN};',
    "a single-quoted directive name": f"'server_name' {DOMAIN};",
    "a quoted semicolon": f'server_name "a;b" {DOMAIN};',
    "an escaped semicolon": f"server_name a\\;b {DOMAIN};",
}


def self_signed(directory, name):
    """A task-owned certificate and unencrypted key, as PEM files (`openssl` is a CI prerequisite)."""
    certificate, key = directory / f"{name}.crt", directory / f"{name}.key"
    subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "1",
                    "-subj", "/CN=" + DOMAIN, "-keyout", str(key), "-out", str(certificate)],
                   capture_output=True, timeout=60, check=True)
    return certificate, key


def pieces(chain, key, tls_listen="    listen 443 ssl; # managed by Certbot\n", http_listen="    listen 80;\n"):
    """Map, TLS and redirect blocks as `certbot install --nginx --redirect` leaves an edge vhost (read on the
    host 2026-09-26: comments, blank lines and the ingress include included)."""
    edge = configure_vhost.content("", DOMAIN, 32946).replace(
        "    location / {\n", "    location / {\n        include /opt/fixture/.backend-ingress.conf;\n")
    mapping, server = edge.split("\n\n", 1)
    tls = server.replace("    listen 80;\n", "").replace("    }\n}\n", (
        "    }\n\n" + tls_listen
        + f"    ssl_certificate {chain}; # managed by Certbot\n"
        + f"    ssl_certificate_key {key}; # managed by Certbot\n"
        + "    include /etc/letsencrypt/options-ssl-nginx.conf; # managed by Certbot\n"
        + "    ssl_dhparam /etc/letsencrypt/ssl-dhparams.pem; # managed by Certbot\n\n\n}\n"))
    http = ("\n\nserver {\n"
            f"    if ($host = {DOMAIN}) {{\n        return 301 https://$host$request_uri;\n"
            "    } # managed by Certbot\n\n\n" + http_listen
            + f"    server_name {DOMAIN};\n    return 404; # managed by Certbot\n\n\n}}")
    return mapping + "\n\n", tls, http


def installed(*arguments, **keywords):
    return "".join(pieces(*arguments, **keywords))


def dumped(*files):
    """`nginx -T` standard output: each loaded file's bytes after its `# configuration file <path>:` line."""
    return b"".join(b"# configuration file " + bytes(path) + b":\n" + text + b"\n" for path, text in files)


class VhostState(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        directory = tempfile.TemporaryDirectory()
        cls.addClassCleanup(directory.cleanup)
        material = Path(directory.name)
        leaf, cls.leaf_key = self_signed(material, "leaf")
        intermediate, cls.other_key = self_signed(material, "intermediate")
        # A fullchain is the leaf, then its chain; nginx presents its first certificate.
        cls.fullchain = leaf.read_text() + intermediate.read_text()
        cls.intermediate = certificate_vhost.leaf_certificate(str(intermediate))
        cls.encrypted_key = material / "encrypted.key"
        subprocess.run(["openssl", "rsa", "-in", str(cls.leaf_key), "-aes256", "-passout", "pass:fixture",
                        "-out", str(cls.encrypted_key)], capture_output=True, timeout=60, check=True)

    def setUp(self):
        directory = tempfile.TemporaryDirectory()
        self.addCleanup(directory.cleanup)
        self.root = Path(directory.name)
        self.nginx = self.root / "conf.d"
        self.nginx.mkdir()
        self.vhost = self.nginx / "backend-fixture-edge.conf"
        live = self.root / "live" / DOMAIN
        live.mkdir(parents=True)
        self.chain, self.key = str(live / "fullchain.pem"), str(live / "privkey.pem")
        Path(self.chain).write_text(self.fullchain)
        shutil.copyfile(self.leaf_key, self.key)
        self.served, self.probed = certificate_vhost.leaf_certificate(self.chain), []
        # The loaded configuration beyond the managed vhost: other files `nginx -T` dumps, and its warnings.
        self.included, self.warnings, self.dump_error, self.dumps = [], b"", None, 0

    def probe(self, domain):
        self.probed.append(domain)
        if isinstance(self.served, Exception):
            raise self.served
        return self.served

    def dump(self):
        self.dumps += 1
        if self.dump_error:
            raise self.dump_error
        main = (Path("/etc/nginx/nginx.conf"), b"http {\n    include /etc/nginx/conf.d/*.conf;\n}\n")
        managed = [(self.vhost, self.vhost.read_bytes())] if self.vhost.is_file() else []
        return dumped(main, *managed, *self.included), self.warnings

    def state(self, text=None):
        if text is not None:
            self.vhost.write_text(text)
        return certificate_vhost.main([str(self.vhost), DOMAIN, self.chain, self.key], self.probe, self.dump)

    def test_the_installed_shape_is_current(self):
        self.assertEqual(self.state(installed(self.chain, self.key)), "current")
        self.assertEqual(self.probed, [DOMAIN])
        _, tls, http = pieces(self.chain, self.key)
        self.assertEqual(self.state(tls.replace("\n        include /opt", "\n#") + http), "current")
        self.assertEqual(self.state(installed(self.chain, self.key,
            tls_listen="    listen 443 ssl;\n    listen [::]:443 ssl ipv6only=on;\n",
            http_listen="    listen 80;\n    listen [::]:80;\n")), "current")

    def test_an_http_only_vhost_is_installed(self):
        self.assertEqual(self.state(configure_vhost.content("", DOMAIN, 32000)), "http_only")
        self.assertEqual(self.state(LEGACY_VHOST.replace("legacy.example", DOMAIN)), "http_only")
        self.assertEqual(self.probed, [], "only a current file is worth a handshake")

    def test_the_running_nginx_must_present_the_lineage(self):
        # The file alone is not enough: a failed reload leaves an older configuration serving the name.
        text = installed(self.chain, self.key)
        for served, expected in ((b"other leaf", "not_served"), (self.intermediate, "not_served"),
                                 (None, "not_served"), (ConnectionRefusedError(), "unreachable"),
                                 (socket.timeout(), "unreachable"), (ssl.SSLError(), "unreachable")):
            with self.subTest(expected=expected, served=served):
                self.served = served
                self.assertEqual(self.state(text), expected)
        self.assertEqual(self.dumps, 0, "a name nginx does not serve is installed before any configuration load")

    def test_another_lineage_is_installed(self):
        text = installed(self.chain, self.key)
        other = self.chain.replace(DOMAIN, "other.example")
        self.assertEqual(self.state(text.replace(f"ssl_certificate {self.chain}", f"ssl_certificate {other}")),
                         "other_certificate")
        self.assertEqual(self.state(text.replace(self.key, self.key + ".old")), "other_certificate")

    def test_a_missing_vhost_or_redirect_is_installed(self):
        self.assertEqual(self.state(), "absent")
        mapping, tls, _ = pieces(self.chain, self.key)
        self.assertEqual(self.state(mapping + tls), "no_redirect")
        text = installed(self.chain, self.key)
        self.assertEqual(self.state(text.replace(f"($host = {DOMAIN})", "($host = other.example)")), "no_redirect")
        self.assertEqual(self.state(text.replace("return 301 https://$host$request_uri;",
                                                 "return 302 https://$host$request_uri;")), "no_redirect")

    def test_ambiguous_shapes_are_installed(self):
        mapping, tls, http = pieces(self.chain, self.key)
        text = mapping + tls + http
        for label, changed in {
            "second tls block": mapping + tls + tls + http,
            "second name": text.replace(f"server_name {DOMAIN};", f"server_name {DOMAIN} www.{DOMAIN};", 1),
            "other domain": text.replace(f"server_name {DOMAIN};", "server_name other.example;", 1),
            "http2": text.replace("listen 443 ssl;", "listen 443 ssl http2;"),
            "other port": text.replace("listen 443 ssl;", "listen 8443 ssl;"),
            "mixed listens": text.replace("listen 443 ssl;", "listen 443 ssl;\n    listen 80;"),
            "duplicate listen": text.replace("listen 443 ssl;", "listen 443 ssl;\n    listen 443 ssl;"),
            "no listen": text.replace("    listen 80;\n", ""),
            "ssl directive": text.replace("listen 443 ssl;", "listen 443 ssl;\n    ssl on;"),
            "second certificate": text.replace("    include /etc/letsencrypt", f"    ssl_certificate {self.chain};\n"
                                               "    include /etc/letsencrypt"),
            "missing key": text.replace(f"    ssl_certificate_key {self.key};", ""),
            "upstream block": "upstream fixture { server 127.0.0.1:1; }\n" + text,
            "top-level directive": "proxy_buffering off;\n" + text,
        }.items():
            with self.subTest(label):
                self.assertEqual(self.state(changed), "ambiguous")
        self.vhost.unlink()
        self.vhost.symlink_to(self.root / "elsewhere.conf")
        self.assertEqual(self.state(), "ambiguous")
        self.vhost.unlink()
        self.vhost.mkdir()
        self.assertEqual(self.state(), "ambiguous")
        for arguments in (["../x", DOMAIN, self.chain, self.key], [str(self.vhost), "Bad_Domain", self.chain,
                          self.key], [str(self.vhost), DOMAIN, "fullchain.pem", self.key], [str(self.vhost)]):
            self.assertEqual(certificate_vhost.main(arguments), "ambiguous")

    def test_unparsable_text_is_installed(self):
        text = installed(self.chain, self.key)
        for label, changed in {
            "unclosed block": text[:-1],
            "extra brace": text + "}\n",
            "unterminated quote": text + "map '\n",
            "unterminated directive": text + "\nlisten 80",
            "braced variable": text.replace("$host;", "${host};"),
            "escaped quote": text.replace("'' close;", "'\\'' close;"),
            "glued quote": text.replace("'' close", "''close"),
            "brace in word": text.replace("return 404;", "return 404};"),
            "empty directive": text.replace("404;", "404;;"),
        }.items():
            with self.subTest(label):
                self.assertEqual(self.state(changed), "unparsable")
        self.vhost.write_bytes(text.encode().replace(b"# managed", b"# \xff managed", 1))
        self.assertEqual(self.state(), "unparsable")

    def test_unreadable_lineage_files_are_installed(self):
        text = installed(self.chain, self.key)
        Path(self.chain).write_text("no certificate here\n")
        self.assertEqual(self.state(text), "lineage_unreadable")
        Path(self.chain).write_bytes(b"\xff")
        self.assertEqual(self.state(), "unparsable")
        Path(self.chain).write_text(self.fullchain)
        os.unlink(self.key)
        self.assertEqual(self.state(), "lineage_unreadable")
        self.assertEqual(self.probed, [])

    def test_a_key_nginx_could_not_load_is_installed(self):
        # nginx refuses a key that does not match the leaf at its next reload, for every tenant at once.
        text = installed(self.chain, self.key)
        self.assertEqual(self.state(text), "current")
        self.probed.clear()
        for label, key in (("another key", self.other_key), ("encrypted key", self.encrypted_key)):
            with self.subTest(label):
                shutil.copyfile(key, self.key)
                self.assertEqual(self.state(), "lineage_unreadable")
                self.assertEqual(self.probed, [])
        # The key is right but the fullchain now leads with another certificate.
        shutil.copyfile(self.leaf_key, self.key)
        leaf, intermediate = self.fullchain.split("-----END CERTIFICATE-----\n")[:2]
        Path(self.chain).write_text(intermediate + "-----END CERTIFICATE-----\n" + leaf + "-----END CERTIFICATE-----\n")
        self.assertEqual(self.state(), "lineage_unreadable")
        self.assertEqual(self.probed, [])

    def test_another_file_serving_the_name_is_installed(self):
        text = installed(self.chain, self.key)
        sibling = self.nginx / "app-robot-domains-fixture.conf"
        # A public domain proxying to the API domain names it only as an upstream host.
        self.included = [(sibling, f"server {{\n    listen 443 ssl;\n    server_name public.example;\n"
                          f"    location /api {{ set $backend {DOMAIN}; proxy_pass https://{DOMAIN}; }}\n}}\n"
                          .encode())]
        self.assertEqual(self.state(text), "current")
        self.assertEqual(self.dumps, 1)
        for names in (DOMAIN, f"www.example {DOMAIN.upper()}", f'"{DOMAIN}"'):
            with self.subTest(names):
                self.included = [(sibling, f"server {{\n    listen 80;\n    server_name\n        {names};\n}}\n"
                                  .encode())]
                self.assertEqual(self.state(), "duplicate")

    def test_a_plain_statement_elsewhere_is_read_exactly(self):
        text = installed(self.chain, self.key)
        sibling = self.nginx / "other-fixture.conf"
        for statement, expected in (
            (f"server_name www.example {DOMAIN};", "duplicate"),
            (f"server_name .{DOMAIN};", "duplicate"),  # nginx and Certbot read the same exact name
            (f"#server_name {DOMAIN};", "duplicate"),  # scanned, not parsed: a commented-out name counts
            ("server_name other.example www.other.example;", "current"),
            (f"server_name \"\" 'www.{DOMAIN}' {DOMAIN}.other;", "current"),
            ("# server_name comes from the edge file; see README\n", "current"),
            ("server_name_in_redirect off;\n    proxy_set_header Host $server_name;", "current"),
            # a regex's own escapes are read cleanly, and `*.example` (which matches DOMAIN as a wildcard) is not a
            # duplicate: only exact names count, the exact-name managed block outranks it in nginx and Certbot
            ("server_name ~^(www\\.)?other\\.example$ *.example;", "current"),
            (f"server_name other.example\x0b{DOMAIN};", "current"),  # one word: nginx splits at four blanks only
        ):
            with self.subTest(statement):
                self.included = [(sibling, f"server {{\n    listen 8443 ssl;\n    {statement}\n}}\n".encode())]
                self.assertEqual(self.state(text), expected)

    def test_a_statement_a_plain_scan_could_misread_is_installed(self):
        # On another port nginx draws no conflict warning, so only the scan can see these; LoadedConfiguration
        # proves the real nginx reads each one as naming the domain.
        text = installed(self.chain, self.key)
        sibling = self.nginx / "other-fixture.conf"
        for label, statement in MISREAD.items():
            with self.subTest(label):
                self.included = [(sibling, f"server {{\n    listen 8443 ssl;\n    {statement}\n}}\n".encode())]
                self.assertEqual(self.state(text), "duplicate")
        # Whatever it names: the words up to the first `;` cannot tell what such a statement names.
        for statement in (b"server_name a # b;\n        c;", b"server_name a\\ b;"):
            with self.subTest(statement):
                self.included = [(sibling, b"server {\n    listen 8443 ssl;\n    " + statement + b"\n}\n")]
                self.assertEqual(self.state(), "duplicate")
        # The managed file's own statement too: the strict parse reads it, but the scan could not count it.
        self.included = []
        self.assertEqual(self.state(text.replace(f"server_name {DOMAIN};", f"server_name # the API; kept\n"
                                                 f"        {DOMAIN};", 1)), "duplicate")

    def test_a_server_block_reached_through_an_include_is_installed(self):
        # Certbot weighs every server block nginx loads, not only the conf.d files: a snippet included from a
        # conf.d file, or from nginx.conf, is a file of its own in the dump.
        text = installed(self.chain, self.key)
        snippet = Path("/etc/nginx/app-robot-domains/fixture/server.inc")
        includer = (self.nginx / "app-robot-domains-fixture.conf",
                    f"include {snippet};\nserver {{ listen 80; server_name other.example; }}\n".encode())
        self.included = [includer, (snippet, b"proxy_ssl_server_name on;\nproxy_ssl_name " + DOMAIN.encode()
                                    + b";\n")]
        self.assertEqual(self.state(text), "current", "a proxied name and proxy_ssl_server_name are not server names")
        self.included = [includer, (snippet, f"server {{\n    listen 8443 ssl;\n    server_name {DOMAIN};\n}}\n"
                                    .encode())]
        self.assertEqual(self.state(), "duplicate")

    def test_the_managed_file_must_be_loaded_once_and_alone(self):
        text = installed(self.chain, self.key).encode()
        self.vhost.write_bytes(text)
        main = (Path("/etc/nginx/nginx.conf"), b"events {}\n")
        other, copy = self.nginx / "other.conf", self.root / "sites-enabled" / "backend-fixture-edge.conf"
        conflict = b'nginx: [warn] conflicting server name "%s" on 0.0.0.0:80, ignored\n'
        for label, loaded in {
            "a copy under another name": (dumped(main, (self.vhost, text), (copy, text)), b""),
            # A section line written inside another file is counted as the managed file loaded twice.
            "a forged section line": (dumped(main, (self.vhost, text), (other, dumped((self.vhost, b"")))), b""),
            # nginx dumps a file once however often it is included, but warns of the second copy.
            "the file included twice": (dumped(main, (self.vhost, text)), conflict % DOMAIN.upper().encode()),
            # The name is served, but from a copy nginx loads instead of the managed file.
            "only a copy loaded": (dumped(main, (copy, text)), b""),
            "the file not loaded": (dumped(main), b""),
        }.items():
            with self.subTest(label):
                self.dump = lambda: loaded
                self.assertEqual(self.state(), "duplicate")
        self.dump = lambda: (dumped(main, (self.vhost, text)), conflict % b"other.example")
        self.assertEqual(self.state(), "current", "only a conflict for this name is doubt")

    def test_a_configuration_that_cannot_be_dumped_is_installed(self):
        text = installed(self.chain, self.key)
        for error in (subprocess.CalledProcessError(1, ["nginx", "-T"]), subprocess.TimeoutExpired(["nginx"], 30),
                      FileNotFoundError(), PermissionError()):
            with self.subTest(error=type(error).__name__):
                self.dump_error = error
                self.assertEqual(self.state(text), "config_unknown")

    def test_the_command_prints_one_state_and_writes_nothing(self):
        self.vhost.write_text(installed(self.chain, self.key))
        before = {path: (path.stat().st_mtime_ns, path.read_bytes()) for path in self.root.rglob("*")
                  if path.is_file()}
        helper = Path(certificate_vhost.__file__)
        for arguments in ([self.chain, self.key], [self.chain + ".old", self.key], []):
            # The command probes the host's own nginx; any answer here is still one fixed token.
            result = subprocess.run([sys.executable, "-B", str(helper), *(
                [str(self.vhost), DOMAIN, *arguments] if arguments else [])],
                capture_output=True, text=True, timeout=30, check=True)
            self.assertIn(result.stdout.strip(), STATES)
            self.assertEqual(result.stdout.count("\n"), 1)
            self.assertEqual(result.stderr, "")
        after = {path: (path.stat().st_mtime_ns, path.read_bytes()) for path in self.root.rglob("*")
                 if path.is_file()}
        self.assertEqual(after, before)


class LoadedConfiguration(unittest.TestCase):
    """The dump format and include resolution of the real nginx that CI installs, against an owned prefix."""

    def setUp(self):
        self.assertIsNotNone(shutil.which("nginx"), "nginx is required for the loaded-configuration contract")
        directory = tempfile.TemporaryDirectory()
        self.addCleanup(directory.cleanup)
        root = Path(directory.name)
        (root / "conf.d").mkdir()
        (root / "snippets").mkdir()
        self.chain, self.key = self_signed(root, "leaf")
        (root / "options-ssl-nginx.conf").write_text("ssl_protocols TLSv1.2 TLSv1.3;\n")
        (root / "ingress.conf").write_text("")
        self.vhost = root / "conf.d" / "backend-fixture-edge.conf"
        self.vhost.write_text(installed(self.chain, self.key).replace("/etc/letsencrypt/options-ssl-nginx.conf",
            f"{root}/options-ssl-nginx.conf").replace(
            "    ssl_dhparam /etc/letsencrypt/ssl-dhparams.pem; # managed by Certbot\n", "")
            .replace("/opt/fixture/.backend-ingress.conf", f"{root}/ingress.conf"))
        (root / "conf.d" / "includer.conf").write_text(f"include {root}/snippets/*.inc;\n")
        self.snippet = root / "snippets" / "extra.inc"
        # A distribution nginx (CI's apt build) opens its compiled-in error log and creates its compiled-in temp
        # directories under /var before and while it reads the configuration; as a non-root runner that is a
        # permission error and exit 1. `-e` and the *_temp_path directives keep both inside the scratch prefix.
        temps = "".join(f"    {kind}_temp_path {root}/tmp/{kind};\n"
                        for kind in ("client_body", "proxy", "fastcgi", "uwsgi", "scgi"))
        (root / "tmp").mkdir()
        (root / "nginx.conf").write_text(
            f"pid {root}/nginx.pid;\nerror_log {root}/error.log;\nevents {{}}\n"
            f"http {{\n    access_log off;\n{temps}    include {root}/conf.d/*.conf;\n}}\n")
        self.command = ("nginx", "-T", "-e", f"{root}/error.log", "-p", f"{root}/", "-c", "nginx.conf")
        self.leaf = certificate_vhost.leaf_certificate(str(self.chain))

    def check(self):
        return certificate_vhost.main([str(self.vhost), DOMAIN, str(self.chain), str(self.key)],
                                      lambda _domain: self.leaf,
                                      lambda: certificate_vhost.loaded_configuration(self.command))

    def test_real_nginx_dumps_a_duplicate_reached_through_an_include(self):
        # The name as an upstream's TLS name and Host header only; nginx would resolve a proxy_pass host itself.
        self.snippet.write_text(f"server {{\n    listen 8081;\n    server_name upstream.example;\n"
                                f"    proxy_ssl_server_name on;\n    proxy_ssl_name {DOMAIN};\n"
                                f"    location / {{ proxy_set_header Host {DOMAIN}; "
                                "proxy_pass https://127.0.0.1:9; }\n}\n")
        self.assertEqual(self.check(), "current")
        self.snippet.write_text(f"server {{\n    listen 8081;\n    server_name {DOMAIN};\n}}\n")
        self.assertEqual(self.check(), "duplicate")
        self.snippet.write_text(f"include {self.vhost};\n")
        self.assertEqual(self.check(), "duplicate", "the managed file included twice is dumped once, with a warning")
        self.snippet.write_text("server {\n")
        self.assertEqual(self.check(), "config_unknown", "a configuration nginx rejects decides nothing")

    def test_real_nginx_names_the_domain_where_a_plain_scan_would_not(self):
        conflict = f'conflicting server name "{DOMAIN}"'.encode()
        for label, statement in MISREAD.items():
            with self.subTest(label):
                # Beside a plain twin on its port, nginx warns: it reads the statement as naming the domain.
                self.snippet.write_text(f"server {{\n    listen 8081;\n    {statement}\n}}\n"
                                        f"server {{\n    listen 8081;\n    server_name {DOMAIN};\n}}\n")
                self.assertIn(conflict, certificate_vhost.loaded_configuration(self.command)[1])
                # Alone on its port it draws no warning, so only the scan can see it.
                self.snippet.write_text(f"server {{\n    listen 8081;\n    {statement}\n}}\n")
                self.assertNotIn(conflict, certificate_vhost.loaded_configuration(self.command)[1])
                self.assertEqual(self.check(), "duplicate")


class ServedCertificate(unittest.TestCase):
    def test_the_handshake_reads_the_certificate_presented_for_the_name(self):
        self.assertIsNotNone(shutil.which("openssl"), "openssl creates the task-owned origin certificate")
        directory = tempfile.TemporaryDirectory()
        self.addCleanup(directory.cleanup)
        root = Path(directory.name)
        certificate, key = self_signed(root, "origin")
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.load_cert_chain(certificate, key)
        names = []
        context.sni_callback = lambda _connection, name, _context: names.append(name)
        listener = socket.socket()
        self.addCleanup(listener.close)
        listener.bind(("127.0.0.1", 0))
        listener.listen(1)

        def accept():
            connection, _ = listener.accept()
            try:
                with context.wrap_socket(connection, server_side=True):
                    pass
            except OSError:
                pass

        thread = threading.Thread(target=accept, daemon=True)
        thread.start()
        origin = listener.getsockname()
        served = certificate_vhost.served_certificate(DOMAIN, origin)
        thread.join(5)
        self.assertEqual(served, certificate_vhost.leaf_certificate(str(certificate)))
        self.assertEqual(names, [DOMAIN])
        listener.close()
        with self.assertRaises(OSError):
            certificate_vhost.served_certificate(DOMAIN, origin)


if __name__ == "__main__":
    unittest.main()