rspamd-mail-trainer-gitea/tests/test_cli.py

from contextlib import redirect_stderr, redirect_stdout
from io import StringIO
from pathlib import Path
import tempfile
import unittest

from rspamd_mail_trainer.cli import ScanStats, SearchOptions, iter_candidates, main, maybe_compile, parse_user_date


def write_mail(path: Path, sender: str, subject: str, body: str = "hello") -> None:
    path.write_text(
        "\n".join([
            f"From: {sender}",
            "To: johannes@example.org",
            f"Subject: {subject}",
            "Date: Fri, 12 Jun 2026 13:36:00 +0000",
            "Message-ID: <test@example.org>",
            "",
            body,
        ]),
        encoding="utf-8",
    )


class CliTests(unittest.TestCase):
    def test_maildir_sender_match(self) -> None:
        with tempfile.TemporaryDirectory() as td:
            root = Path(td) / "Maildir"
            cur = root / "cur"
            cur.mkdir(parents=True)
            write_mail(cur / "1:2,S", "no-reply@example.test", "Monthly report")
            opts = SearchOptions(
                from_re=maybe_compile("example.test", True, True),
                to_re=None,
                subject_re=None,
                any_re=None,
                after=None,
                before=None,
                max_size=1024 * 1024,
                include_seen_duplicates=False,
            )
            matches = list(iter_candidates([root], opts))
            self.assertEqual(len(matches), 1)
            self.assertEqual(matches[0][0].sender, "no-reply@example.test")

    def test_iter_candidates_populates_scan_stats(self) -> None:
        with tempfile.TemporaryDirectory() as td:
            root = Path(td) / "Maildir"
            cur = root / "cur"
            cur.mkdir(parents=True)
            write_mail(cur / "1:2,S", "no-reply@example.test", "Monthly report")
            opts = SearchOptions(
                from_re=maybe_compile("example.test", True, True),
                to_re=None,
                subject_re=None,
                any_re=None,
                after=None,
                before=None,
                max_size=1024 * 1024,
                include_seen_duplicates=False,
            )
            stats = ScanStats()
            matches = list(iter_candidates([root], opts, stats))
            self.assertEqual(len(matches), 1)
            self.assertEqual(stats.roots_seen, 1)
            self.assertEqual(stats.files_seen, 1)
            self.assertEqual(stats.messages_read, 1)
            self.assertEqual(stats.matches_found, 1)

    def test_main_logs_scan_progress_to_stderr(self) -> None:
        with tempfile.TemporaryDirectory() as td:
            root = Path(td) / "Maildir"
            cur = root / "cur"
            cur.mkdir(parents=True)
            write_mail(cur / "1:2,S", "no-reply@example.test", "Monthly report")
            stdout = StringIO()
            stderr = StringIO()

            with redirect_stdout(stdout), redirect_stderr(stderr):
                rc = main(["search", "--root", str(root), "--from", "example.test", "--literal"])

            self.assertEqual(rc, 0)
            self.assertIn("[1] maildir-file", stdout.getvalue())
            self.assertIn("INFO: Using 1 root(s)", stderr.getvalue())
            self.assertIn("INFO: Scanning mail root:", stderr.getvalue())
            self.assertIn("INFO: Scan finished:", stderr.getvalue())

    def test_quiet_suppresses_info_logging(self) -> None:
        with tempfile.TemporaryDirectory() as td:
            root = Path(td) / "Maildir"
            cur = root / "cur"
            cur.mkdir(parents=True)
            write_mail(cur / "1:2,S", "no-reply@example.test", "Monthly report")
            stdout = StringIO()
            stderr = StringIO()

            with redirect_stdout(stdout), redirect_stderr(stderr):
                rc = main(["search", "--quiet", "--root", str(root), "--from", "example.test", "--literal"])

            self.assertEqual(rc, 0)
            self.assertIn("[1] maildir-file", stdout.getvalue())
            self.assertNotIn("INFO:", stderr.getvalue())

    def test_check_reports_missing_rspamc_without_traceback(self) -> None:
        with tempfile.TemporaryDirectory() as td:
            root = Path(td) / "Maildir"
            cur = root / "cur"
            cur.mkdir(parents=True)
            write_mail(cur / "1:2,S", "no-reply@example.test", "Monthly report")
            stdout = StringIO()
            stderr = StringIO()

            with redirect_stdout(stdout), redirect_stderr(stderr):
                rc = main([
                    "check",
                    "--root",
                    str(root),
                    "--from",
                    "example.test",
                    "--literal",
                    "--rspamc",
                    "/no/such/rspamc",
                ])

            self.assertEqual(rc, 3)
            self.assertIn("returncode=127", stdout.getvalue())
            self.assertIn("Cannot run /no/such/rspamc", stderr.getvalue())

    def test_subject_match(self) -> None:
        with tempfile.TemporaryDirectory() as td:
            root = Path(td) / "Maildir"
            cur = root / "cur"
            cur.mkdir(parents=True)
            write_mail(cur / "2:2,S", "sender@example.org", "Invoice 123")
            opts = SearchOptions(
                from_re=None,
                to_re=None,
                subject_re=maybe_compile("Invoice", True, True),
                any_re=None,
                after=None,
                before=None,
                max_size=1024 * 1024,
                include_seen_duplicates=False,
            )
            matches = list(iter_candidates([root], opts))
            self.assertEqual(len(matches), 1)

    def test_parse_user_date_day_start(self) -> None:
        parsed = parse_user_date("2026-06-13")
        self.assertIsNotNone(parsed)
        assert parsed is not None
        self.assertEqual(parsed.hour, 0)
        self.assertEqual(parsed.minute, 0)


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