"""Offline TLS integration tests; requires the openssl command."""
import json
from pathlib import Path
import socket
import ssl
import subprocess
import sys
import tempfile
import threading
from unittest.mock import patch
import check_tls


def run(arguments):
    return subprocess.run(
        [sys.executable, str(Path(check_tls.__file__)), *arguments],
        capture_output=True, text=True, timeout=10, check=False,
    )


def serve(listener, context, failures):
    try:
        for _ in range(4):
            raw, _ = listener.accept()
            with raw:
                raw.settimeout(3)
                try:
                    with context.wrap_socket(raw, server_side=True):
                        pass
                except ssl.SSLError:
                    pass  # Two clients intentionally reject this certificate.
    except Exception as exc:
        failures.append(exc)


with tempfile.TemporaryDirectory(prefix="tls-test-") as directory:
    cert = Path(directory) / "cert.pem"
    key = Path(directory) / "key.pem"
    config = Path(directory) / "openssl.cnf"
    config.write_text("[req]\ndistinguished_name = dn\n[dn]\n", encoding="utf-8")
    subprocess.run([
        "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
        "-config", str(config),
        "-days", "2", "-subj", "/CN=localhost",
        "-addext", "subjectAltName=DNS:localhost",
        "-addext", "basicConstraints=critical,CA:TRUE",
        "-addext", "keyUsage=critical,digitalSignature,keyEncipherment,keyCertSign",
        "-addext", "subjectKeyIdentifier=hash",
        "-addext", "authorityKeyIdentifier=keyid:always",
        "-keyout", str(key), "-out", str(cert),
    ], check=True, capture_output=True, timeout=30)
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.load_cert_chain(cert, key)
    failures = []
    with socket.socket() as listener:
        listener.bind(("127.0.0.1", 0))
        listener.listen(5)
        listener.settimeout(8)
        port = str(listener.getsockname()[1])
        worker = threading.Thread(target=serve, args=(listener, context, failures),
                                  daemon=True)
        worker.start()
        try:
            for host, extra, expected, code in [
                ("localhost", ["--cafile", str(cert), "--warn-days", "0"], "ok", 0),
                ("localhost", ["--cafile", str(cert), "--warn-days", "3"], "expiring", 1),
                ("localhost", [], "verify_error", 1),
                ("127.0.0.1", ["--cafile", str(cert)], "verify_error", 1),
            ]:
                completed = run([host, "--port", port, "--timeout", "1", *extra])
                report = json.loads(completed.stdout)
                assert completed.returncode == code and not completed.stderr, completed
                assert report["status"] == expected, report
                if expected in ("ok", "expiring"):
                    assert 1 < report["days_remaining"] <= 2, report
                    assert report["expires_utc"].endswith("+00:00"), report
                else:
                    assert report["error"], report
        finally:
            worker.join(timeout=9)
        assert not worker.is_alive() and not failures, failures

    with socket.socket() as reserved:
        reserved.bind(("127.0.0.1", 0))  # Bound but deliberately not listening.
        completed = run(["127.0.0.1", "--port", str(reserved.getsockname()[1])])
        assert completed.returncode == 1, completed
        assert json.loads(completed.stdout)["status"] == "error", completed

for extra in [["--timeout", "nan"], ["--timeout", "0"],
              ["--timeout", "61"], ["--port", "0"], ["--warn-days", "-1"]]:
    completed = run(["localhost", *extra])
    assert completed.returncode == 2 and not completed.stdout and completed.stderr

with patch("check_tls.socket.create_connection", side_effect=socket.timeout("test timeout")):
    assert check_tls.check("localhost", 443, 1, 14)["status"] == "error"
print("OK: valid, expiring, untrusted, hostname mismatch, refused, arguments, timeout")