"""Tests for daemon/handlers/acme.py — handler endpoint logic.""" import asyncio import urllib.error from pathlib import Path from unittest.mock import MagicMock, patch import pytest import daemon.handlers.acme as acme_mod from daemon.handlers.acme import ( IssueRequest, _check_account_registered, _check_acme_account, _check_acme_home_writable, _check_dns_public, _check_dns_resolves, _check_firewall_port_80, _check_nginx_config, _check_nginx_running, _check_openssl_available, _check_port_80_listening, _get_account_info, _get_external_ip, _is_private_ip, _validate, deactivate_account, generate_self_signed, get_account, get_renew_status, issue_cert, register_account, renew_cert, ) from daemon.server import ConflictError, NotFoundError def _await_renew(body): """Start a renewal via renew_cert and await its background task.""" async def _run(): result = renew_cert(None, body) task = acme_mod._ISSUANCE_TASKS.get(result.get("request_id")) if task is not None: await task return result return asyncio.run(_run()) class TestGenerateSelfSigned: def test_generate_creates_files(self, tmp_path): with ( patch("daemon.handlers.acme.PROJECT_DIR", tmp_path), ): result = generate_self_signed(None, {"domain": "test.local"}) assert result["domain"] == "test.local" assert result["generated"] is True certs_dir = tmp_path / "data" / "certs" assert result["cert"] == str(certs_dir / "test.local.crt") assert result["key"] == str(certs_dir / "test.local.key") assert (certs_dir / "test.local.crt").is_file() assert (certs_dir / "test.local.key").is_file() def test_generate_idempotent_skips_existing(self, tmp_path): certs_dir = tmp_path / "data" / "certs" certs_dir.mkdir(parents=True) (certs_dir / "test.local.crt").write_text("dummy-cert") (certs_dir / "test.local.key").write_text("dummy-key") with patch("daemon.handlers.acme.PROJECT_DIR", tmp_path): result = generate_self_signed(None, {"domain": "test.local"}) assert result["generated"] is False def test_generate_partial_existing(self, tmp_path): certs_dir = tmp_path / "data" / "certs" certs_dir.mkdir(parents=True) (certs_dir / "test.local.crt").write_text("dummy-cert") # key missing -> should regenerate with patch("daemon.handlers.acme.PROJECT_DIR", tmp_path): result = generate_self_signed(None, {"domain": "test.local"}) assert result["generated"] is True def test_generate_custom_days(self, tmp_path): with ( patch("daemon.handlers.acme.PROJECT_DIR", tmp_path), patch("subprocess.run") as mock_run, ): def _create_files(*args, **kwargs): certs_dir = tmp_path / "data" / "certs" certs_dir.mkdir(parents=True, exist_ok=True) (certs_dir / "test.local.crt").touch() (certs_dir / "test.local.key").touch() return Path("") mock_run.side_effect = _create_files generate_self_signed(None, {"domain": "test.local", "days": 730}) args = mock_run.call_args[0][0] assert "-days" in args idx = args.index("-days") assert args[idx + 1] == "730" certs_dir = tmp_path / "data" / "certs" if (certs_dir / "test.local.crt").is_file(): assert certs_dir.is_dir() def test_generate_creates_directory(self, tmp_path): with patch("daemon.handlers.acme.PROJECT_DIR", tmp_path): generate_self_signed(None, {"domain": "test.local"}) assert (tmp_path / "data" / "certs").is_dir() def test_generate_requires_domain(self): with pytest.raises(ValueError, match="domain"): generate_self_signed(None, {"foo": "bar"}) def test_generate_requires_body(self): with pytest.raises(ValueError, match="body"): generate_self_signed(None, None) class TestCheckNginxRunning: def test_via_systemctl_active(self): from unittest.mock import patch mock_result = MagicMock(returncode=0, stdout="active\n") with patch("subprocess.run", return_value=mock_result): passed, msg = _check_nginx_running() assert passed is True assert "running" in msg def test_via_systemctl_inactive(self, tmp_path): from unittest.mock import patch mock_result = MagicMock(returncode=3, stdout="inactive\n") with ( patch("subprocess.run", return_value=mock_result), patch.object(Path, "is_file", return_value=False), ): passed, _ = _check_nginx_running() assert passed is False def test_via_pid_file(self): from unittest.mock import patch def run_side_effect(cmd, **kwargs): raise FileNotFoundError() with ( patch("subprocess.run", side_effect=run_side_effect), patch.object(Path, "read_text", return_value="1234\n"), ): def fake_is_file(self): if self == Path("/var/run/nginx.pid"): return True if str(self) == "/proc/1234/status": return True return Path(self).is_file() with patch.object(Path, "is_file", fake_is_file): passed, _ = _check_nginx_running() assert passed is True class TestCheckNginxConfig: def test_valid_config(self): with patch("lib.nginx.test_config", return_value=(True, "test passed")): passed, _ = _check_nginx_config() assert passed is True def test_invalid_config(self): with patch("lib.nginx.test_config", return_value=(False, "test failed: blah")): passed, msg = _check_nginx_config() assert passed is False assert "blah" in msg class TestCheckFirewallPort80: def test_port_open_via_services(self): from unittest.mock import patch def proc_side_effect(cmd, **kwargs): if "--get-active-zones" in cmd: return MagicMock(returncode=0, stdout="public\n eth0\n") if "--list-services" in cmd: return MagicMock(returncode=0, stdout="http https dns ssh\n") return MagicMock(returncode=1, stdout="") with ( patch("lib.common.run_proc", side_effect=proc_side_effect), ): passed, _ = _check_firewall_port_80() assert passed is True def test_blocked_by_firewall(self): from unittest.mock import patch def proc_side_effect(cmd, **kwargs): if "--get-active-zones" in cmd: return MagicMock(returncode=0, stdout="public\n eth0\n") if "--list-services" in cmd: return MagicMock(returncode=0, stdout="https dns ssh\n") if "--list-ports" in cmd: return MagicMock(returncode=0, stdout="443/tcp\n") return MagicMock(returncode=1, stdout="") with ( patch("lib.common.run_proc", side_effect=proc_side_effect), ): passed, msg = _check_firewall_port_80() assert passed is False assert "80" in msg def test_firewalld_not_detected(self): from unittest.mock import patch with patch( "lib.common.run_proc", return_value=MagicMock(returncode=1, stdout="") ): passed, msg = _check_firewall_port_80() assert passed is True assert "skipping" in msg class TestCheckAcmeHomeWritable: def test_writable(self, tmp_path): acme_dir = tmp_path / "acme" acme_dir.mkdir() with patch("daemon.handlers.acme._ACME_HOME", acme_dir): passed, _ = _check_acme_home_writable() assert passed is True class TestCheckAcmeHomeWritable_Missing: def test_missing_dir(self, tmp_path): acme_dir = tmp_path / "acme" with patch("daemon.handlers.acme._ACME_HOME", acme_dir): passed, msg = _check_acme_home_writable() assert passed is False assert "does not exist" in msg class TestCheckAcmeHomeWritable_Permissions: def test_not_writable(self, tmp_path): acme_dir = tmp_path / "acme" acme_dir.mkdir() acme_dir.chmod(0o444) try: with patch("daemon.handlers.acme._ACME_HOME", acme_dir): passed, _ = _check_acme_home_writable() assert passed is False finally: acme_dir.chmod(0o755) class TestCheckOpensslAvailable: def test_available(self): from unittest.mock import patch mock_result = MagicMock(returncode=0, stdout="OpenSSL 3.0.0\n") with ( patch("shutil.which", return_value="/usr/bin/openssl"), patch("subprocess.run", return_value=mock_result), ): passed, msg = _check_openssl_available() assert passed is True assert "OpenSSL" in msg def test_not_found(self): with patch("shutil.which", return_value=None): passed, _ = _check_openssl_available() assert passed is False class TestCheckPort80Listening: def test_listening(self): from unittest.mock import patch mock_sock = MagicMock() mock_sock.connect_ex.return_value = 0 mock_sock.__enter__ = MagicMock(return_value=mock_sock) mock_sock.__exit__ = MagicMock(return_value=False) with patch("socket.socket", return_value=mock_sock): passed, _ = _check_port_80_listening() assert passed is True def test_not_listening(self): from unittest.mock import patch mock_sock = MagicMock() mock_sock.connect_ex.return_value = 111 mock_sock.__enter__ = MagicMock(return_value=mock_sock) mock_sock.__exit__ = MagicMock(return_value=False) with patch("socket.socket", return_value=mock_sock): passed, _ = _check_port_80_listening() assert passed is False class TestCheckAcmeAccount: def test_via_acme_info(self, tmp_path): from unittest.mock import patch acme_dir = tmp_path / "acme" acme_dir.mkdir() (acme_dir / "acme.sh").write_text("#!/bin/sh\nexit 0") (acme_dir / "acme.sh").chmod(0o755) with ( patch("daemon.handlers.acme._ACME_HOME", acme_dir), patch( "daemon.handlers.acme._find_acme_bin", return_value=str(acme_dir / "acme.sh"), ), patch("subprocess.run", return_value=MagicMock(returncode=0, stdout="ok")), ): passed, _ = _check_acme_account() assert passed is True def test_via_account_conf(self, tmp_path): from unittest.mock import patch acme_dir = tmp_path / "acme" acme_dir.mkdir() (acme_dir / ".account.conf").write_text( "ACME_LEEMAIL='test@example.com'\nACME_MCA='letsencrypt'\n" ) def run_side_effect(cmd, **kwargs): raise FileNotFoundError() with ( patch("daemon.handlers.acme._ACME_HOME", acme_dir), patch("subprocess.run", side_effect=run_side_effect), ): passed, _ = _check_acme_account() assert passed is True def test_not_configured(self, tmp_path): from unittest.mock import patch acme_dir = tmp_path / "acme" acme_dir.mkdir() def run_side_effect(cmd, **kwargs): raise FileNotFoundError() with ( patch("daemon.handlers.acme._ACME_HOME", acme_dir), patch("subprocess.run", side_effect=run_side_effect), ): passed, _ = _check_acme_account() assert passed is False class TestCheckDnsPublic: def test_resolves_correctly(self): from unittest.mock import patch mock_result = MagicMock( returncode=0, stdout="example.com has address 52.14.150.110" ) with ( patch( "daemon.handlers.acme._get_local_ips", return_value={"52.14.150.110"} ), patch("subprocess.run", return_value=mock_result), ): passed, _ = _check_dns_public("example.com") assert passed is True def test_does_not_resolve(self): from unittest.mock import patch mock_result = MagicMock(returncode=1, stdout="NXDOMAIN") with ( patch("daemon.handlers.acme._get_local_ips", return_value={"8.8.8.8"}), patch("subprocess.run", return_value=mock_result), ): passed, _ = _check_dns_public("example.com") assert passed is False def test_nat_detected_skips_check(self): from unittest.mock import patch with patch( "daemon.handlers.acme._get_local_ips", return_value={"192.168.1.1"}, ): passed, msg = _check_dns_public("example.com") assert passed is True assert "NAT" in msg class TestCheckDomainFormat: def test_valid(self): from daemon.handlers.acme import _check_domain_format passed, _ = _check_domain_format("example.com") assert passed is True def test_invalid(self): from daemon.handlers.acme import _check_domain_format passed, _ = _check_domain_format("-bad.com") assert passed is False class TestValidate: def test_returns_all_checks(self): with ( patch( "daemon.handlers.acme._check_acme_installed", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_openssl_available", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_acme_home_writable", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_account_registered", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_email_configured", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_acme_account", return_value=(True, "ok") ), patch("daemon.handlers.acme._check_webroot", return_value=(True, "ok")), patch( "daemon.handlers.acme._check_nginx_running", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_nginx_config", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_challenge_config", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_port_80_listening", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_firewall_port_80", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_domain_format", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_dns_resolves", return_value=(True, "ok") ), patch("daemon.handlers.acme._check_dns_public", return_value=(True, "ok")), patch( "daemon.handlers.acme._check_existing_cert", return_value=(True, "ok") ), ): result = _validate("example.com") assert result["domain"] == "example.com" assert result["ready"] is True assert len(result["checks"]) == 16 for c in result["checks"]: assert c["passed"] is True def test_failing_check_blocks_ready(self): with ( patch( "daemon.handlers.acme._check_acme_installed", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_openssl_available", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_acme_home_writable", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_account_registered", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_email_configured", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_acme_account", return_value=(True, "ok") ), patch("daemon.handlers.acme._check_webroot", return_value=(True, "ok")), patch( "daemon.handlers.acme._check_nginx_running", return_value=(False, "not running"), ), patch( "daemon.handlers.acme._check_nginx_config", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_challenge_config", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_port_80_listening", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_firewall_port_80", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_dns_resolves", return_value=(True, "ok") ), patch("daemon.handlers.acme._check_dns_public", return_value=(True, "ok")), patch( "daemon.handlers.acme._check_existing_cert", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_domain_format", return_value=(True, "ok") ), ): result = _validate("example.com") assert result["ready"] is False nginx_check = next(c for c in result["checks"] if c["name"] == "nginx_running") assert nginx_check["passed"] is False assert nginx_check["blocking"] is True def test_non_blocking_failure_allows_ready(self): with ( patch( "daemon.handlers.acme._check_acme_installed", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_openssl_available", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_acme_home_writable", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_account_registered", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_email_configured", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_acme_account", return_value=(True, "ok") ), patch("daemon.handlers.acme._check_webroot", return_value=(True, "ok")), patch( "daemon.handlers.acme._check_nginx_running", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_nginx_config", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_challenge_config", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_port_80_listening", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_firewall_port_80", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_dns_resolves", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_dns_public", return_value=(False, "skipped"), ), patch( "daemon.handlers.acme._check_existing_cert", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_domain_format", return_value=(True, "ok") ), ): result = _validate("example.com") assert result["ready"] is True dns_pub = next(c for c in result["checks"] if c["name"] == "dns_public") assert dns_pub["passed"] is False assert dns_pub["blocking"] is False class TestExternalIp: def test_success(self): mock_resp = MagicMock() mock_resp.read.return_value = b"93.184.216.34" mock_resp.__enter__ = MagicMock(return_value=mock_resp) mock_resp.__exit__ = MagicMock(return_value=False) with patch("urllib.request.urlopen", return_value=mock_resp): result = _get_external_ip() assert result == "93.184.216.34" def test_primary_fails_fallback_succeeds(self): def urlopen_side_effect(url, timeout=None): if "ipify" in url.full_url: raise urllib.error.URLError("primary down") mock_resp = MagicMock() mock_resp.read.return_value = b"93.184.216.50" mock_resp.__enter__ = MagicMock(return_value=mock_resp) mock_resp.__exit__ = MagicMock(return_value=False) return mock_resp with patch("urllib.request.urlopen", side_effect=urlopen_side_effect): result = _get_external_ip() assert result == "93.184.216.50" def test_both_fail_returns_none(self): with patch("urllib.request.urlopen", side_effect=urllib.error.URLError("fail")): result = _get_external_ip() assert result is None def test_env_override_url(self): mock_resp = MagicMock() mock_resp.read.return_value = b"8.8.8.8" mock_resp.__enter__ = MagicMock(return_value=mock_resp) mock_resp.__exit__ = MagicMock(return_value=False) with ( patch.dict( "os.environ", {"VACUUM_WALL_EXTERNAL_IP_URL": "https://my.ip.api"} ), patch("urllib.request.urlopen", return_value=mock_resp) as mock_urlopen, ): result = _get_external_ip() assert result == "8.8.8.8" call_url = mock_urlopen.call_args[0][0] assert "my.ip.api" in call_url.full_url class TestIsPrivateIp: def test_public_ip(self): assert _is_private_ip("8.8.8.8") is False def test_private_10(self): assert _is_private_ip("10.0.0.1") is True def test_private_192(self): assert _is_private_ip("192.168.1.1") is True def test_private_172(self): assert _is_private_ip("172.16.0.1") is True def test_invalid_ip(self): assert _is_private_ip("not-an-ip") is False def test_doc_range_203(self): assert _is_private_ip("203.0.113.5") is True class TestCheckDnsResolves: def test_match_local_ip(self): mock_results = [("AF_INET", "STREAM", 6, "", ("192.168.1.10", 80))] with ( patch("socket.getaddrinfo", return_value=mock_results), patch("daemon.handlers.acme._get_local_ips", return_value={"192.168.1.10"}), ): passed, msg = _check_dns_resolves("example.com") assert passed is True assert "DNS resolves correctly" in msg def test_match_external_ip(self): mock_results = [("AF_INET", "STREAM", 6, "", ("93.184.216.34", 80))] with ( patch("socket.getaddrinfo", return_value=mock_results), patch("daemon.handlers.acme._get_local_ips", return_value={"192.168.1.10"}), patch( "daemon.handlers.acme._get_external_ip", return_value="93.184.216.34" ), ): passed, msg = _check_dns_resolves("example.com") assert passed is True assert "NAT" in msg def test_external_ip_mismatch(self): mock_results = [("AF_INET", "STREAM", 6, "", ("93.184.216.99", 80))] with ( patch("socket.getaddrinfo", return_value=mock_results), patch("daemon.handlers.acme._get_local_ips", return_value={"192.168.1.10"}), patch( "daemon.handlers.acme._get_external_ip", return_value="93.184.216.34" ), ): passed, msg = _check_dns_resolves("example.com") assert passed is False assert "93.184.216.99" in msg assert "93.184.216.34" in msg def test_external_ip_unavailable(self): mock_results = [("AF_INET", "STREAM", 6, "", ("93.184.216.34", 80))] with ( patch("socket.getaddrinfo", return_value=mock_results), patch("daemon.handlers.acme._get_local_ips", return_value={"192.168.1.10"}), patch("daemon.handlers.acme._get_external_ip", return_value=None), ): passed, msg = _check_dns_resolves("example.com") assert passed is False assert "Cannot verify via external IP" in msg def test_private_ip(self): mock_results = [("AF_INET", "STREAM", 6, "", ("192.168.1.50", 80))] with ( patch("socket.getaddrinfo", return_value=mock_results), patch("daemon.handlers.acme._get_local_ips", return_value={"10.0.0.1"}), patch("daemon.handlers.acme._get_external_ip", return_value=None), ): passed, msg = _check_dns_resolves("example.com") assert passed is False assert "private IP" in msg def test_unresolved(self): import socket with patch("socket.getaddrinfo", side_effect=socket.gaierror("NXDOMAIN")): passed, msg = _check_dns_resolves("nonexistent.example") assert passed is False assert "NXDOMAIN" in msg class TestCheckAccountRegistered: def test_registered_with_keys(self, tmp_path): acme_dir = tmp_path / "acme" acme_dir.mkdir() (acme_dir / ".account.conf").write_text( "ACME_LEEMAIL='user@example.com'\nACME_MCA='letsencrypt'\n" ) with patch("daemon.handlers.acme._ACME_HOME", acme_dir): passed, msg = _check_account_registered() assert passed is True assert "registered" in msg def test_missing_account_conf(self, tmp_path): acme_dir = tmp_path / "acme" acme_dir.mkdir() with patch("daemon.handlers.acme._ACME_HOME", acme_dir): passed, msg = _check_account_registered() assert passed is False assert "Register" in msg def test_incomplete_account_conf(self, tmp_path): acme_dir = tmp_path / "acme" acme_dir.mkdir() (acme_dir / ".account.conf").write_text("ACME_LEEMAIL='user@example.com'\n") with patch("daemon.handlers.acme._ACME_HOME", acme_dir): passed, msg = _check_account_registered() assert passed is False assert "Register" in msg class TestValidateAccountCheck: def test_account_not_registered_blocks(self): with ( patch( "daemon.handlers.acme._check_acme_installed", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_openssl_available", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_acme_home_writable", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_account_registered", return_value=( False, "Register an ACME account before issuing certificates", ), ), patch( "daemon.handlers.acme._check_email_configured", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_acme_account", return_value=(True, "ok") ), patch("daemon.handlers.acme._check_webroot", return_value=(True, "ok")), patch( "daemon.handlers.acme._check_nginx_running", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_nginx_config", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_challenge_config", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_port_80_listening", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_firewall_port_80", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_domain_format", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_dns_resolves", return_value=(True, "ok") ), patch("daemon.handlers.acme._check_dns_public", return_value=(True, "ok")), patch( "daemon.handlers.acme._check_existing_cert", return_value=(True, "ok") ), ): result = _validate("example.com") assert result["ready"] is False acct = next(c for c in result["checks"] if c["name"] == "account_registered") assert acct["passed"] is False assert acct["blocking"] is True def test_account_registered_passes(self): with ( patch( "daemon.handlers.acme._check_acme_installed", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_openssl_available", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_acme_home_writable", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_account_registered", return_value=(True, "ACME account is registered"), ), patch( "daemon.handlers.acme._check_email_configured", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_acme_account", return_value=(True, "ok") ), patch("daemon.handlers.acme._check_webroot", return_value=(True, "ok")), patch( "daemon.handlers.acme._check_nginx_running", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_nginx_config", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_challenge_config", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_port_80_listening", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_firewall_port_80", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_domain_format", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_dns_resolves", return_value=(True, "ok") ), patch("daemon.handlers.acme._check_dns_public", return_value=(True, "ok")), patch( "daemon.handlers.acme._check_existing_cert", return_value=(True, "ok") ), ): result = _validate("example.com") assert result["ready"] is True acct = next(c for c in result["checks"] if c["name"] == "account_registered") assert acct["passed"] is True assert acct["blocking"] is True def test_email_missing_non_blocking(self): with ( patch( "daemon.handlers.acme._check_acme_installed", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_openssl_available", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_acme_home_writable", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_account_registered", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_email_configured", return_value=(False, "Contact email not set"), ), patch( "daemon.handlers.acme._check_acme_account", return_value=(True, "ok") ), patch("daemon.handlers.acme._check_webroot", return_value=(True, "ok")), patch( "daemon.handlers.acme._check_nginx_running", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_nginx_config", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_challenge_config", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_port_80_listening", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_firewall_port_80", return_value=(True, "ok"), ), patch( "daemon.handlers.acme._check_domain_format", return_value=(True, "ok") ), patch( "daemon.handlers.acme._check_dns_resolves", return_value=(True, "ok") ), patch("daemon.handlers.acme._check_dns_public", return_value=(True, "ok")), patch( "daemon.handlers.acme._check_existing_cert", return_value=(True, "ok") ), ): result = _validate("example.com") assert result["ready"] is True email_chk = next(c for c in result["checks"] if c["name"] == "email_configured") assert email_chk["passed"] is False assert email_chk["blocking"] is False class TestGetAccount: def test_registered_account(self, tmp_path): acme_dir = tmp_path / "acme" acme_dir.mkdir() (acme_dir / ".account.conf").write_text( "ACME_LEEMAIL='admin@example.com'\nACME_MCA='letsencrypt'\nACME_CERTKEYSIZE='2048'\n" ) with patch("daemon.handlers.acme._ACME_HOME", acme_dir): result = _get_account_info() assert result["registered"] is True assert result["email"] == "admin@example.com" assert result["ca"] == "Let's Encrypt" assert result["key_length"] == 2048 def test_unregistered_account(self, tmp_path): acme_dir = tmp_path / "acme" acme_dir.mkdir() with patch("daemon.handlers.acme._ACME_HOME", acme_dir): result = _get_account_info() assert result["registered"] is False assert result["email"] == "" assert result["ca"] == "" assert result["key_length"] is None def test_zerossl_account(self, tmp_path): acme_dir = tmp_path / "acme" acme_dir.mkdir() (acme_dir / ".account.conf").write_text( "ACME_LEEMAIL='user@zerossl.com'\nACME_MCA='zerossl'\n" ) with patch("daemon.handlers.acme._ACME_HOME", acme_dir): result = _get_account_info() assert result["ca"] == "ZeroSSL" def test_get_account_endpoint(self, tmp_path): acme_dir = tmp_path / "acme" acme_dir.mkdir() (acme_dir / ".account.conf").write_text( "ACME_LEEMAIL='test@example.com'\nACME_MCA='letsencrypt'\n" ) with patch("daemon.handlers.acme._ACME_HOME", acme_dir): result = get_account(None, None) assert result["registered"] is True assert result["email"] == "test@example.com" class TestRegisterAccount: def test_success(self, tmp_path): proj = tmp_path / "project" proj.mkdir() acme_dir = proj / "data" / "acme" acme_dir.mkdir(parents=True) config_dir = proj / "config" / "acme" config_dir.mkdir(parents=True) (config_dir / "config.json").write_text("{}\n") def fake_run_acme(args): (acme_dir / ".account.conf").write_text( "ACME_LEEMAIL='test@example.com'\nACME_MCA='letsencrypt'\n" ) with ( patch("daemon.handlers.acme._ACME_HOME", acme_dir), patch("daemon.handlers.acme.PROJECT_DIR", proj), patch("daemon.handlers.acme._run_acme", side_effect=fake_run_acme), patch("daemon.handlers.acme.refresh_state"), ): result = register_account( None, {"email": "test@example.com", "server": "letsencrypt"} ) assert result["registered"] is True assert result["email"] == "test@example.com" assert result["ca"] == "letsencrypt" def test_missing_email(self): with pytest.raises(ValueError, match="email"): register_account(None, {"foo": "bar"}) def test_server_default_letsencrypt(self, tmp_path): proj = tmp_path / "project" proj.mkdir() acme_dir = proj / "data" / "acme" acme_dir.mkdir(parents=True) config_dir = proj / "config" / "acme" config_dir.mkdir(parents=True) (config_dir / "config.json").write_text("{}\n") captured_args = [] def fake_run_acme(args): captured_args.append(args) (acme_dir / ".account.conf").write_text( "ACME_LEEMAIL='test@example.com'\nACME_MCA='letsencrypt'\n" ) with ( patch("daemon.handlers.acme._ACME_HOME", acme_dir), patch("daemon.handlers.acme.PROJECT_DIR", proj), patch("daemon.handlers.acme._run_acme", side_effect=fake_run_acme), patch("daemon.handlers.acme.refresh_state"), ): register_account(None, {"email": "test@example.com"}) assert "--server" in captured_args[0] assert "letsencrypt" in captured_args[0] class TestDeactivateAccount: def test_success(self, tmp_path): proj = tmp_path / "project" proj.mkdir() config_dir = proj / "config" / "acme" config_dir.mkdir(parents=True) (config_dir / "config.json").write_text( '{"email": "test@example.com", "ca": "letsencrypt"}\n' ) with ( patch("daemon.handlers.acme.PROJECT_DIR", proj), patch("daemon.handlers.acme._run_acme"), patch("daemon.handlers.acme.refresh_state"), ): result = deactivate_account(None, None) assert result["email"] == "" cfg_text = (config_dir / "config.json").read_text() assert "email" not in cfg_text assert "ca" not in cfg_text def test_cleans_account_conf_files(self, tmp_path): proj = tmp_path / "project" proj.mkdir() config_dir = proj / "config" / "acme" config_dir.mkdir(parents=True) (config_dir / "config.json").write_text( '{"email": "test@example.com", "ca": "letsencrypt"}\n' ) acme_dir = proj / "data" / "acme" acme_dir.mkdir(parents=True) (acme_dir / ".account.conf").write_text( "ACME_LEEMAIL='test@example.com'\nACME_MCA='letsencrypt'\n" ) (acme_dir / "account.conf").write_text( "ACME_LEEMAIL='test@example.com'\nACME_MCA='letsencrypt'\n" ) with ( patch("daemon.handlers.acme._ACME_HOME", acme_dir), patch("daemon.handlers.acme.PROJECT_DIR", proj), patch("daemon.handlers.acme._run_acme"), patch("daemon.handlers.acme.refresh_state"), ): deactivate_account(None, None) assert not (acme_dir / ".account.conf").is_file() assert not (acme_dir / "account.conf").is_file() class TestIssueCertExistingCerts: """Phase 3: issue_cert blocks when cert expires today (days == 0) or tomorrow (days == 1).""" def test_days_zero_blocks(self): """days_until_expiry returns 0 (expires today) — should block.""" with ( patch( "daemon.handlers.acme._validate", return_value={"ready": True, "checks": []}, ), patch( "lib.acme.list_certs", return_value=[{"domain": "example.com", "days_until_expiry": 0}], ), pytest.raises(ConflictError, match="0 days remaining"), ): asyncio.run(issue_cert(None, {"domain": "example.com"})) def test_days_one_blocks(self): """days_until_expiry returns 1 (expires tomorrow) — should still block.""" with ( patch( "daemon.handlers.acme._validate", return_value={"ready": True, "checks": []}, ), patch( "lib.acme.list_certs", return_value=[{"domain": "example.com", "days_until_expiry": 1}], ), pytest.raises(ConflictError, match="1 day remaining"), ): asyncio.run(issue_cert(None, {"domain": "example.com"})) def test_days_negative_one_allows(self): """days_until_expiry returns -1 (already expired) — should not block.""" async def _fake_run_issue(req): pass with ( patch( "daemon.handlers.acme._validate", return_value={"ready": True, "checks": []}, ), patch( "lib.acme.list_certs", return_value=[{"domain": "example.com", "days_until_expiry": -1}], ), patch.dict("daemon.handlers.acme._ISSUANCES", clear=True), patch( "daemon.handlers.acme._run_issue", new=MagicMock(side_effect=_fake_run_issue), ) as mock_run_issue, ): result = asyncio.run(issue_cert(None, {"domain": "example.com"})) assert result["domain"] == "example.com" assert "request_id" in result mock_run_issue.assert_called_once() class TestRenewCert: def test_requires_body(self): with pytest.raises(ValueError, match="Request body required"): renew_cert(None, None) def test_requires_domain(self): with pytest.raises(ValueError, match="'domain' is required"): renew_cert(None, {"force": True}) def test_starts_background_task(self): with ( patch.dict("daemon.handlers.acme._ISSUANCES", clear=True), patch.dict("daemon.handlers.acme._ISSUANCE_TASKS", clear=True), patch( "daemon.handlers.acme._run_acme", return_value="Renewed 'example.com'" ), patch("daemon.handlers.acme.refresh_state") as mock_refresh, ): result = _await_renew({"domain": "example.com"}) req = acme_mod._ISSUANCES[result["request_id"]] assert result["domain"] == "example.com" assert "request_id" in result assert req.status == "completed" assert [s.status for s in req.steps] == ["done", "done", "done"] mock_refresh.assert_called_once_with(["acme"]) def test_skip_when_not_due(self): output = ( "[Thu Aug 20 01:27:41 AM UTC 2026] Skipping. " "Next renewal time is: 1785068261 (2026-07-26T12:17:41Z)" ) with ( patch.dict("daemon.handlers.acme._ISSUANCES", clear=True), patch.dict("daemon.handlers.acme._ISSUANCE_TASKS", clear=True), patch("daemon.handlers.acme._run_acme", return_value=output), patch("daemon.handlers.acme.refresh_state") as mock_refresh, ): result = _await_renew({"domain": "example.com"}) req = acme_mod._ISSUANCES[result["request_id"]] assert req.status == "skipped" assert req.steps[0].status == "done" mock_refresh.assert_not_called() def test_failure_marks_failed(self): with ( patch.dict("daemon.handlers.acme._ISSUANCES", clear=True), patch.dict("daemon.handlers.acme._ISSUANCE_TASKS", clear=True), patch( "daemon.handlers.acme._run_acme", side_effect=RuntimeError("acme.sh failed with exit code 1: boom"), ), ): result = _await_renew({"domain": "example.com"}) req = acme_mod._ISSUANCES[result["request_id"]] assert req.status == "failed" assert req.steps[0].status == "error" assert "boom" in req.steps[0].message def test_force_appends_flag(self): with ( patch.dict("daemon.handlers.acme._ISSUANCES", clear=True), patch.dict("daemon.handlers.acme._ISSUANCE_TASKS", clear=True), patch("daemon.handlers.acme._run_acme", return_value="ok") as mock_acme, patch("daemon.handlers.acme.refresh_state"), ): _await_renew({"domain": "example.com", "force": True}) renew_args = mock_acme.call_args_list[0].args[0] assert "--force" in renew_args def test_dedup_running_returns_existing(self): existing = IssueRequest(request_id="existing123", domain="example.com") with ( patch.dict("daemon.handlers.acme._ISSUANCES", clear=True), patch.dict("daemon.handlers.acme._ISSUANCE_TASKS", clear=True), patch("daemon.handlers.acme._run_acme") as mock_acme, ): acme_mod._ISSUANCES["existing123"] = existing result = renew_cert(None, {"domain": "example.com"}) assert result == { "request_id": "existing123", "domain": "example.com", "status": "existing", } mock_acme.assert_not_called() class TestGetRenewStatus: def test_missing_id(self): with pytest.raises(ValueError, match="'id' is required"): get_renew_status(None, None) with pytest.raises(ValueError, match="'id' is required"): get_renew_status(None, {}) def test_unknown_id(self): with ( patch.dict("daemon.handlers.acme._ISSUANCES", clear=True), pytest.raises(NotFoundError, match="not found"), ): get_renew_status(None, {"id": "unknown"}) def test_returns_request_dict(self): with ( patch.dict("daemon.handlers.acme._ISSUANCES", clear=True), patch.dict("daemon.handlers.acme._ISSUANCE_TASKS", clear=True), patch("daemon.handlers.acme._run_acme", return_value="Renewed"), patch("daemon.handlers.acme.refresh_state"), ): result = _await_renew({"domain": "example.com"}) status = get_renew_status(None, {"id": result["request_id"]}) assert status["request_id"] == result["request_id"] assert status["domain"] == "example.com" assert status["status"] == "completed" assert [s["name"] for s in status["steps"]] == ["renew", "deploy", "refresh"]