stack/tests/mail/test_droplet_lifecycle.py

"""Lifecycle coverage for mail/droplet.py — covers create, destroy, DNS, DKIM, seed, provision."""

from __future__ import annotations

import socket
from unittest.mock import MagicMock, patch

import pytest

from mail.droplet import (
    DOMAIN,
    HOSTNAME,
    _do_client,
    _get_ssh_key_ids,
    _ssh,
    apply_dns,
    attach_smarthost,
    down,
    export_dkim,
    provision,
    rotate_creds,
    seed_mailboxes,
    status,
    up,
    write_git_mailer_env,
)


class TestDoClient:
    @patch("mail.droplet.env", return_value="test-token")
    @patch("pydo.Client")
    def test_creates_client(self, mc_pydo, mc_env):
        _do_client()
        mc_pydo.assert_called_once_with(token="test-token")

    @patch("mail.droplet.env", return_value="")
    def test_raises_no_token(self, mc_env):
        with pytest.raises(RuntimeError, match="DIGITAL_OCEAN_PAT"):
            _do_client()


class TestGetSshKeyIds:
    def test_extracts_ids(self):
        client = MagicMock()
        client.ssh_keys.list.return_value = {"ssh_keys": [{"id": 1}, {"id": 2}]}
        assert _get_ssh_key_ids(client) == [1, 2]

    def test_empty(self):
        client = MagicMock()
        client.ssh_keys.list.return_value = {"ssh_keys": []}
        assert _get_ssh_key_ids(client) == []


class TestSsh:
    @patch("subprocess.run")
    def test_basic(self, mc_run):
        mc_run.return_value = MagicMock(returncode=0)
        _ssh("1.2.3.4", "echo", "hi")
        assert mc_run.called
        args = mc_run.call_args[0][0]
        assert "ssh" == args[0]
        assert "root@1.2.3.4" in args


class TestUpCreate:
    @patch("mail.droplet._do_client")
    @patch("mail.droplet._wait_for_active", return_value="9.8.7.6")
    @patch("mail.droplet._save_json")
    @patch("mail.droplet._cloud_init", return_value="#!/bin/bash\necho ok")
    @patch("mail.droplet._get_ssh_key_ids", return_value=[42])
    def test_creates_new(self, mc_keys, mc_init, mc_save, mc_wait, mc_client):
        client = mc_client.return_value
        client.droplets.list.return_value = {"droplets": []}
        client.droplets.create.return_value = {"droplet": {"id": 999}}

        result = up()
        assert result["id"] == 999
        assert result["public_ip"] == "9.8.7.6"
        client.droplets.create.assert_called_once()

    @patch("mail.droplet._do_client")
    @patch("mail.droplet._get_ssh_key_ids", return_value=[])
    def test_no_ssh_keys(self, mc_keys, mc_client):
        client = mc_client.return_value
        client.droplets.list.return_value = {"droplets": []}
        with pytest.raises(RuntimeError, match="No SSH keys"):
            up()


class TestDownConfirm:
    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("pathlib.Path.unlink")
    def test_destroy(self, mc_unlink, mc_discover, mc_client):
        mc_discover.return_value = {"id": 123, "name": HOSTNAME}
        client = mc_client.return_value
        down(confirm=True)
        client.droplets.destroy.assert_called_once_with(123)

    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    def test_refuse_without_confirm(self, mc_discover, mc_client):
        mc_discover.return_value = {"id": 123, "name": HOSTNAME}
        with pytest.raises(RuntimeError, match="refusing"):
            down(confirm=False)

    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("pathlib.Path.unlink")
    def test_destroy_exception(self, mc_unlink, mc_discover, mc_client):
        mc_discover.return_value = {"id": 123, "name": HOSTNAME}
        client = mc_client.return_value
        client.droplets.destroy.side_effect = Exception("already gone")
        down(confirm=True)


class TestStatusPTR:
    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("mail.droplet._save_json")
    @patch("socket.gethostbyaddr", return_value=(HOSTNAME, [], []))
    def test_ptr_match(self, mc_ptr, mc_save, mc_discover, mc_client):
        mc_discover.return_value = {
            "id": 1,
            "hostname": HOSTNAME,
            "public_ip": "1.2.3.4",
        }
        result = status()
        assert result["ptr_match"] is True

    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("mail.droplet._save_json")
    @patch("socket.gethostbyaddr", side_effect=socket.herror("no PTR"))
    def test_ptr_herror(self, mc_ptr, mc_save, mc_discover, mc_client):
        mc_discover.return_value = {
            "id": 1,
            "hostname": HOSTNAME,
            "public_ip": "1.2.3.4",
        }
        result = status()
        assert result["ptr"] is None
        assert result["ptr_match"] is False


class TestApplyDnsBody:
    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("httpx.Client")
    @patch("mail.cloudflare.ensure_mail_dns")
    def test_calls_ensure(self, mc_ensure, mc_http_cls, mc_discover, mc_client):
        mc_discover.return_value = {
            "hostname": HOSTNAME,
            "public_ip": "1.2.3.4",
        }
        http = MagicMock()
        http.__enter__ = MagicMock(return_value=http)
        http.__exit__ = MagicMock(return_value=False)
        mc_http_cls.return_value = http

        apply_dns()
        mc_ensure.assert_called_once()


class TestExportDkim:
    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("mail.droplet._ssh")
    @patch("httpx.Client")
    @patch("mail.cloudflare.ensure_dkim_record")
    def test_publishes(self, mc_dkim, mc_http_cls, mc_ssh, mc_discover, mc_client):
        mc_discover.return_value = {"public_ip": "1.2.3.4"}
        mc_ssh.return_value = MagicMock(returncode=0, stdout="v=DKIM1; p=abc")

        http = MagicMock()
        http.__enter__ = MagicMock(return_value=http)
        http.__exit__ = MagicMock(return_value=False)
        mc_http_cls.return_value = http

        export_dkim()
        mc_dkim.assert_called_once()

    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet", return_value=None)
    def test_no_droplet(self, mc_discover, mc_client):
        with pytest.raises(RuntimeError, match="no mail droplet"):
            export_dkim()

    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("mail.droplet._ssh")
    def test_ssh_fails(self, mc_ssh, mc_discover, mc_client):
        mc_discover.return_value = {"public_ip": "1.2.3.4"}
        mc_ssh.return_value = MagicMock(returncode=1, stdout="", stderr="err")
        with pytest.raises(RuntimeError, match="could not read DKIM"):
            export_dkim()


class TestAttachSmarhost:
    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("mail.postmark.ensure_postmark_server", return_value="srv-token")
    @patch("mail.postmark.ensure_postmark_domain", return_value={"ID": 1})
    @patch("mail.postmark.publish_postmark_dns")
    @patch("mail.postmark.verify_postmark_domain")
    @patch("httpx.Client")
    @patch("subprocess.run")
    @patch("mail.droplet._ssh")
    def test_full(
        self,
        mc_ssh,
        mc_sub,
        mc_http_cls,
        mc_verify,
        mc_publish,
        mc_domain,
        mc_server,
        mc_discover,
        mc_client,
    ):
        mc_discover.return_value = {"public_ip": "1.2.3.4"}
        mc_ssh.return_value = MagicMock(returncode=0)

        http = MagicMock()
        http.__enter__ = MagicMock(return_value=http)
        http.__exit__ = MagicMock(return_value=False)
        mc_http_cls.return_value = http

        attach_smarthost()
        mc_server.assert_called_once()
        mc_ssh.assert_called()


class TestSeedMailboxes:
    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("mail.droplet._load_json")
    @patch("mail.droplet._ssh")
    @patch("mail.droplet._save_json")
    def test_cached_sync(self, mc_save, mc_ssh, mc_load, mc_discover, mc_client):
        mc_discover.return_value = {"public_ip": "1.2.3.4"}
        mc_load.return_value = {"postmaster": "pw1"}
        mc_ssh.return_value = MagicMock(returncode=0)

        seed_mailboxes(addrs=(f"postmaster@{DOMAIN}",))
        assert mc_ssh.call_count == 1

    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("mail.droplet._load_json")
    @patch("mail.droplet._ssh")
    @patch("mail.droplet._save_json")
    def test_cached_create_fallback(
        self, mc_save, mc_ssh, mc_load, mc_discover, mc_client
    ):
        mc_discover.return_value = {"public_ip": "1.2.3.4"}
        mc_load.return_value = {"postmaster": "pw1"}
        mc_ssh.return_value = MagicMock(returncode=1)

        seed_mailboxes(addrs=(f"postmaster@{DOMAIN}",))
        assert mc_ssh.call_count == 3

    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("mail.droplet._load_json", return_value={})
    @patch("mail.droplet.rotate_creds")
    def test_no_cache_rotates(self, mc_rotate, mc_load, mc_discover, mc_client):
        mc_discover.return_value = {"public_ip": "1.2.3.4"}
        seed_mailboxes(addrs=(f"postmaster@{DOMAIN}",))
        mc_rotate.assert_called_once()

    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet", return_value=None)
    def test_no_droplet(self, mc_discover, mc_client):
        with pytest.raises(RuntimeError, match="no mail droplet"):
            seed_mailboxes()


class TestWriteGitMailerEnv:
    def test_writes_new(self, tmp_path):
        creds_file = tmp_path / "creds.json"
        creds_file.write_text('{"git": "pw123"}')
        droplet_file = tmp_path / "droplet.json"
        droplet_file.write_text(f'{{"hostname": "{HOSTNAME}"}}')
        env_file = tmp_path / "mailer.env"

        with (
            patch("mail.droplet.CREDS_JSON", creds_file),
            patch("mail.droplet.DROPLET_JSON", droplet_file),
            patch("mail.droplet.GIT_MAILER_ENV", env_file),
        ):
            result = write_git_mailer_env()
        assert result is True
        assert env_file.exists()

    def test_no_creds(self, tmp_path):
        creds_file = tmp_path / "nope.json"
        with patch("mail.droplet.CREDS_JSON", creds_file):
            result = write_git_mailer_env()
        assert result is False

    def test_no_git_creds(self, tmp_path):
        creds_file = tmp_path / "creds.json"
        creds_file.write_text('{"postmaster": "pw"}')
        droplet_file = tmp_path / "droplet.json"
        droplet_file.write_text("{}")
        with (
            patch("mail.droplet.CREDS_JSON", creds_file),
            patch("mail.droplet.DROPLET_JSON", droplet_file),
        ):
            result = write_git_mailer_env()
        assert result is False

    def test_unchanged(self, tmp_path):
        desired = (
            "GITEA__mailer__ENABLED=true\n"
            "GITEA__mailer__PROTOCOL=smtps\n"
            f"GITEA__mailer__SMTP_ADDR={HOSTNAME}\n"
            "GITEA__mailer__SMTP_PORT=465\n"
            f"GITEA__mailer__USER=git@{DOMAIN}\n"
            "GITEA__mailer__PASSWD=pw123\n"
            f"GITEA__mailer__FROM=git@{DOMAIN}\n"
        )
        creds_file = tmp_path / "creds.json"
        creds_file.write_text('{"git": "pw123"}')
        droplet_file = tmp_path / "droplet.json"
        droplet_file.write_text(f'{{"hostname": "{HOSTNAME}"}}')
        env_file = tmp_path / "mailer.env"
        env_file.write_text(desired)
        with (
            patch("mail.droplet.CREDS_JSON", creds_file),
            patch("mail.droplet.DROPLET_JSON", droplet_file),
            patch("mail.droplet.GIT_MAILER_ENV", env_file),
        ):
            result = write_git_mailer_env()
        assert result is False


class TestRotateCredsBody:
    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("mail.droplet._ssh")
    @patch("mail.droplet._load_json", return_value={})
    @patch("mail.droplet._save_json")
    @patch("pathlib.Path.chmod")
    def test_set_succeeds(
        self, mc_chmod, mc_save, mc_load, mc_ssh, mc_discover, mc_client
    ):
        mc_discover.return_value = {"public_ip": "1.2.3.4"}
        mc_ssh.return_value = MagicMock(returncode=0)
        pw = rotate_creds("git")
        assert len(pw) == 24
        assert mc_ssh.call_count == 1

    @patch("mail.droplet._do_client")
    @patch("mail.droplet.discover_droplet")
    @patch("mail.droplet._ssh")
    @patch("mail.droplet._load_json", return_value={})
    @patch("mail.droplet._save_json")
    @patch("pathlib.Path.chmod")
    def test_set_fails_then_create(
        self, mc_chmod, mc_save, mc_load, mc_ssh, mc_discover, mc_client
    ):
        mc_discover.return_value = {"public_ip": "1.2.3.4"}
        mc_ssh.side_effect = [
            MagicMock(returncode=1),
            MagicMock(returncode=0),
            MagicMock(returncode=0),
        ]
        pw = rotate_creds("user@sub.example.com")
        assert len(pw) == 24
        assert mc_ssh.call_count == 3


class TestProvision:
    @patch("mail.droplet.up")
    @patch("mail.droplet._load_json", return_value={"public_ip": "1.2.3.4"})
    @patch("mail.droplet._ssh")
    @patch("mail.droplet.apply_dns")
    @patch("mail.droplet.export_dkim")
    @patch("mail.droplet.seed_mailboxes")
    @patch("mail.droplet.attach_smarthost")
    @patch("mail.droplet.write_git_mailer_env", return_value=True)
    def test_happy(
        self, mc_git, mc_smart, mc_seed, mc_dkim, mc_dns, mc_ssh, mc_load, mc_up
    ):
        mc_ssh.return_value = MagicMock(returncode=0, stdout="v=DKIM1; p=abc")
        provision(wait_seconds=1)
        mc_up.assert_called_once()
        mc_dns.assert_called_once()
        mc_seed.assert_called_once()

    @patch("mail.droplet.up")
    @patch("mail.droplet._load_json", return_value={"public_ip": "1.2.3.4"})
    @patch("mail.droplet._ssh")
    @patch("mail.droplet.apply_dns")
    @patch("mail.droplet.export_dkim", side_effect=RuntimeError("no key"))
    @patch("mail.droplet.seed_mailboxes")
    @patch("mail.droplet.attach_smarthost")
    @patch("mail.droplet.write_git_mailer_env", return_value=False)
    def test_dkim_fails(
        self, mc_git, mc_smart, mc_seed, mc_dkim, mc_dns, mc_ssh, mc_load, mc_up
    ):
        mc_ssh.return_value = MagicMock(returncode=1, stdout="")
        provision(wait_seconds=0)
        mc_seed.assert_called_once()