#!/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("alice@example.test\nbob@example.test\n") mod.fromAddr, mod.mailsecure = "ops@example.test", 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], ["alice@example.test", "bob@example.test"]) 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", "alice@example.test") 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": "alice@example.test"}, {"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", "ops@example.test", "", "admin@example.test", "test:1666") sent = [call.args[0] for call in smtp.send_message.call_args_list] self.assertEqual([message["To"] for message in sent], ["alice@example.test", "admin@example.test"]) 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", "ops@example.test", "", 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()