test_additional_maintenance.py #1

#!/usr/bin/env python3
"""Remaining maintenance contracts with fake Perforce/SMTP, never real mail."""
import io
import os
from pathlib import Path
import smtplib
import subprocess
import tempfile
import unittest
from unittest.mock import Mock, patch

import test_maintenance_review as fixtures


class AdditionalMaintenance(unittest.TestCase):
    def test_pymail_address_list_and_secure_transport(self):
        with fixtures.module("pymail") as mod, tempfile.TemporaryDirectory() as tmp:
            body, addresses = Path(tmp) / "body", Path(tmp) / "addresses"
            body.write_text("maintenance diagnostic\n")
            addresses.write_text("[email protected]\[email protected]\n")
            mod.fromAddr, mod.mailsecure = "[email protected]", 1
            mod.mailUser, mod.mailPass = "test-user", "fixture-only-password"
            smtp = Mock()
            with patch.object(mod.smtplib, "SMTP", return_value=smtp):
                mod.main(["-t", str(addresses), "-s", "Diagnostic", "-i", str(body)])
            self.assertEqual(smtp.starttls.call_count, 2)
            self.assertEqual(smtp.login.call_count, 2)
            self.assertEqual([call.args[1] for call in smtp.sendmail.call_args_list], ["[email protected]", "[email protected]"])
            self.assertTrue(all("maintenance diagnostic" in call.args[2] for call in smtp.sendmail.call_args_list))

    def test_pymail_transport_failure_propagates(self):
        with fixtures.module("pymail") as mod, tempfile.TemporaryDirectory() as tmp:
            body = Path(tmp) / "body"
            body.write_text("test")
            with patch.object(mod.smtplib, "SMTP", side_effect=smtplib.SMTPException("denied")):
                with self.assertRaises(smtplib.SMTPException):
                    mod.mail(body, "subject", "[email protected]")

    def test_inactive_user_notifications_filter_protected_recent_and_invalid_accounts(self):
        with fixtures.module("email_pending_user_deletes") as mod:
            p4 = Mock()
            now = 2_000_000_000
            old = str(now - 60 * 86400)
            p4.run.return_value = [{"User": "alice", "Access": old, "Email": "[email protected]"},
                                  {"User": "p4admin", "Access": old},
                                  {"User": "recent", "Access": str(now)},
                                  {"User": "invalid", "Access": "never"}]
            smtp = Mock()
            smtp.__enter__ = Mock(return_value=smtp)
            smtp.__exit__ = Mock(return_value=False)
            with patch.object(mod.time, "time", return_value=now), patch.object(mod.smtplib, "SMTP", return_value=smtp):
                mod.warn_inactive_users(p4, 4, "smtp.example.test", "[email protected]", "", "[email protected]", "test:1666")
            sent = [call.args[0] for call in smtp.send_message.call_args_list]
            self.assertEqual([message["To"] for message in sent], ["[email protected]", "[email protected]"])
            self.assertTrue(all("alice" in message["Subject"] for message in sent))

    def test_inactive_user_query_failure_prevents_notifications(self):
        with fixtures.module("email_pending_user_deletes") as mod:
            p4 = Mock()
            p4.run.side_effect = fixtures.P4Exception("denied")
            with patch.object(mod.smtplib, "SMTP") as smtp:
                with self.assertRaises(fixtures.P4Exception):
                    mod.warn_inactive_users(p4, 4, "smtp.example.test", "[email protected]", "", None, "test:1666")
            smtp.assert_not_called()

    def test_password_reset_uses_stdin_and_propagates_rejection(self):
        class Input(io.StringIO):
            def close(self):
                self.saved = self.getvalue()
                super().close()
        with fixtures.module("setpass", ["1", "alice"]) as mod:
            mod.defaultpass = "fixture-secret"
            for status in (0, 7):
                process = Mock(stdin=Input(), stdout=io.StringIO(""), stderr=io.StringIO("denied"))
                process.wait.return_value = status
                with patch.object(mod, "Popen", return_value=process) as popen, patch.object(mod.time, "sleep"):
                    if status:
                        with self.assertRaises(SystemExit) as error:
                            mod.setpassword("alice")
                        self.assertEqual(error.exception.code, 1)
                    else:
                        mod.setpassword("alice")
                self.assertEqual(process.stdin.saved, "fixture-secret\nfixture-secret\n")
                self.assertNotIn("fixture-secret", str(popen.call_args.args))

    def test_mirror_groups_preserves_members_and_adds_requested_users(self):
        with fixtures.module("mirroraccess", ["1", "source", "alice", "bob"]) as mod:
            calls = []
            def command(args, **kwargs):
                calls.append((args, kwargs))
                if args[1] == "groups":
                    return subprocess.CompletedProcess(args, 0, "team\n")
                if "-o" in args:
                    return subprocess.CompletedProcess(args, 0, "Group: team\nUsers:\n\tsource\n")
                return subprocess.CompletedProcess(args, 0, "")
            with patch.object(fixtures.sdputils, "run_checked", side_effect=command):
                mod.main()
            self.assertEqual(calls[-1][1]["input"], "Group: team\nUsers:\n\tsource\n\talice\n\tbob\n")

    def test_mirror_query_failure_prevents_group_writes(self):
        with fixtures.module("mirroraccess", ["1", "source", "alice"]) as mod:
            with patch.object(fixtures.sdputils, "run_checked", side_effect=subprocess.CalledProcessError(7, ["p4"])) as run:
                with self.assertRaises(subprocess.CalledProcessError):
                    mod.main()
            self.assertEqual(run.call_count, 1)

    def test_update_changes_preserves_specs_and_stops_after_failed_read(self):
        for fail in (False, True):
            with self.subTest(fail=fail), tempfile.TemporaryDirectory() as tmp:
                listing = Path(tmp) / "changes"
                listing.write_text("41\n\n42\n")
                writes = []
                def command(args, **kwargs):
                    if "-o" in args:
                        if fail:
                            raise subprocess.CalledProcessError(7, args, stderr="denied")
                        return subprocess.CompletedProcess(args, 0, "Change: " + args[-1] + "\nDescription:\n\tkeep this\n")
                    writes.append(kwargs["input"])
                    return subprocess.CompletedProcess(args, 0, "")
                with patch.dict(os.environ, {"LOGS": tmp}), patch.object(fixtures.sdputils, "run_checked", side_effect=command):
                    if fail:
                        with self.assertRaises(SystemExit) as error:
                            with fixtures.module("update-changes", ["blue", str(listing)]):
                                pass
                        self.assertEqual(error.exception.code, 7)
                        self.assertEqual(writes, [])
                    else:
                        with fixtures.module("update-changes", ["blue", str(listing)]):
                            pass
                        self.assertEqual(writes, ["Change: 41\nDescription:\n\tkeep this\n", "Change: 42\nDescription:\n\tkeep this\n"])


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