"""Tests for daemon/handlers/acme.py — handler endpoint logic.""" import urllib.error from pathlib import Path from unittest.mock import MagicMock, patch import pytest from daemon.handlers.acme import ( _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, register_account, ) class TestGenerateSelfSigned: def test_generate_creates_files(self, tmp_path): with ( patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"), ): result = generate_self_signed(None, {"domain": "test.local"}) assert result["domain"] == "test.local" assert result["generated"] is True cert_dir = tmp_path / "acme" / "test.local" assert result["cert"] == str(cert_dir / "fullchain.cer") assert result["key"] == str(cert_dir / "test.local.key") assert (cert_dir / "fullchain.cer").is_file() assert (cert_dir / "test.local.key").is_file() def test_generate_idempotent_skips_existing(self, tmp_path): cert_dir = tmp_path / "acme" / "test.local" cert_dir.mkdir(parents=True) (cert_dir / "fullchain.cer").write_text("dummy-cert") (cert_dir / "test.local.key").write_text("dummy-key") with patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"): result = generate_self_signed(None, {"domain": "test.local"}) assert result["generated"] is False def test_generate_partial_existing(self, tmp_path): cert_dir = tmp_path / "acme" / "test.local" cert_dir.mkdir(parents=True) (cert_dir / "fullchain.cer").write_text("dummy-cert") # key missing -> should regenerate with patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"): 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._ACME_HOME", tmp_path / "acme"), patch("subprocess.run") as mock_run, ): def _create_files(*args, **kwargs): cert_dir = tmp_path / "acme" / "test.local" cert_dir.mkdir(parents=True, exist_ok=True) (cert_dir / "fullchain.cer").touch() (cert_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" cert_dir = tmp_path / "acme" / "test.local" if (cert_dir / "fullchain.cer").is_file(): assert cert_dir.is_dir() def test_generate_creates_directory(self, tmp_path): with patch("daemon.handlers.acme._ACME_HOME", tmp_path / "acme"): generate_self_signed(None, {"domain": "test.local"}) assert (tmp_path / "acme" / "test.local").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 192.168.1.1" ) with ( patch("socket.gethostbyname", return_value="192.168.1.1"), 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("socket.gethostbyname", return_value="192.168.1.1"), patch("subprocess.run", return_value=mock_result), ): passed, _ = _check_dns_public("example.com") assert passed is False 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()