"""Currency grant CLI regressions; runnable without third-party packages.""" from __future__ import annotations from contextlib import redirect_stdout import io import json from pathlib import Path import subprocess import sys import tempfile import unittest from unittest import mock TOOLS = Path(__file__).resolve().parents[1] sys.path.insert(0, str(TOOLS)) import dev_mail_grant class CurrencyMailGrantTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.output = Path(self.temporary.name) / "nested" / "grants.json" def args(self, *extra): return dev_mail_grant.parser().parse_args([ "grant", "--output", str(self.output), "--attachment", "4:0:123", *extra, ]) def run_grant(self, *extra): stream = io.StringIO() with redirect_stdout(stream): self.assertEqual(dev_mail_grant.grant(self.args(*extra)), 0) return json.loads(stream.getvalue()) def child(self, *extra): # -S removes site packages, proving the command can run without Crypto. return [sys.executable, "-S", str(TOOLS / "dev_mail_grant.py"), "grant", "--output", str(self.output), "--attachment", "4:0:123", *extra] def test_multiple_attachments_create_one_mail_and_preserve_duplicates(self): result = self.run_grant("--identity", "release-1", "--title", " 发放测试 ", "--body", " 正文 ", "--attachment", "3:0:5", "--attachment", "4:0:2", "--attachment", "12:0:1", "--attachment", "20:0:2147483647") value = dev_mail_grant.load_grants(self.output) self.assertEqual(value["version"], 1) self.assertEqual(len(value["grants"]), 1) self.assertEqual(value["grants"][0], result["grant"]) self.assertEqual(result["grant"]["title"], "发放测试") self.assertEqual(result["grant"]["body"], "正文") self.assertGreater(result["grant"]["sent_at"], 0) self.assertEqual(result["grant"]["rewards"], [ {"type": 4, "id": 0, "count": 123}, {"type": 3, "id": 0, "count": 5}, {"type": 4, "id": 0, "count": 2}, {"type": 12, "id": 0, "count": 1}, {"type": 20, "id": 0, "count": 2147483647}, ]) def test_identity_retry_keeps_original_timestamp_and_does_not_write(self): first = self.run_grant("--identity", "once") original = self.output.read_bytes() with mock.patch.object(dev_mail_grant, "atomic_json", side_effect=AssertionError("must not write")): second = self.run_grant("--identity", "once") self.assertTrue(first["created"]) self.assertFalse(second["created"]) self.assertEqual(first["grant"], second["grant"]) self.assertEqual(self.output.read_bytes(), original) def test_paid_jewelry_draw_tickets_and_large_gold_share_one_idempotent_mail(self): extra = ("--identity", "large-rewards", "--attachment", "2:0:100000000", "--attachment", "8:1000:100000000", "--attachment", "8:1104:100000000", "--attachment", "4:0:1000000000") first = self.run_grant(*extra) second = self.run_grant(*extra) self.assertTrue(first["created"]) self.assertFalse(second["created"]) self.assertEqual(first["grant"], second["grant"]) self.assertEqual(first["grant"]["rewards"][1:], [ {"type": 2, "id": 0, "count": 100000000}, {"type": 8, "id": 1000, "count": 100000000}, {"type": 8, "id": 1104, "count": 100000000}, {"type": 4, "id": 0, "count": 1000000000}, ]) def test_identity_conflict_is_rejected_without_changing_file(self): self.run_grant("--identity", "once") original = self.output.read_bytes() for extra in (("--title", "another"), ("--body", "another"), ("--attachment", "3:0:1")): with self.subTest(extra=extra), self.assertRaisesRegex(ValueError, "内容不同"): self.run_grant("--identity", "once", *extra) self.assertEqual(self.output.read_bytes(), original) def test_one_use_content_ticket_is_an_idempotent_single_item_attachment(self): first = self.run_grant("--identity", "full-moon-ticket", "--attachment", "19:450030:1") second = self.run_grant("--identity", "full-moon-ticket", "--attachment", "19:450030:1") self.assertTrue(first["created"]) self.assertFalse(second["created"]) self.assertEqual(first["grant"], second["grant"]) self.assertEqual(first["grant"]["rewards"][-1], {"type": 19, "id": 450030, "count": 1}) def test_omitted_identity_appends_distinct_grants(self): first, second = self.run_grant(), self.run_grant() self.assertNotEqual(first["grant"]["identity"], second["grant"]["identity"]) self.assertEqual(len(dev_mail_grant.load_grants(self.output)["grants"]), 2) def test_invalid_attachments_are_rejected(self): for attachment in ("8:1:1", "8:0:1", "8:1001:1", "8:1104:2147483648", "8:1000:0", "14:1000:1", "2:1:1", "19:450029:1", "19:450030:0", "19:450030:2", "4:1:1", "4:0:0", "4:0:-1", "4:0:2147483648", "4:0", "4:0:1:2", "4:0:true"): with self.subTest(attachment=attachment), self.assertRaises(SystemExit), redirect_stdout(io.StringIO()): with mock.patch("sys.stderr", new=io.StringIO()): self.args("--attachment", attachment) self.assertFalse(self.output.exists()) def test_invalid_title_body_or_identity_does_not_create_output(self): for extra in (("--title", " "), ("--title", "x" * 501), ("--body", "x" * 5001), ("--identity", " "), ("--identity", "x" * 501)): with self.subTest(extra=extra), self.assertRaises(ValueError): self.run_grant(*extra) self.assertFalse(self.output.exists()) def test_existing_invalid_schema_is_preserved(self): entry = {"identity": "one", "title": "title", "body": "body", "sent_at": 1, "rewards": [{"type": 4, "id": 0, "count": 1}]} invalid_values = [ [], {"version": True, "grants": []}, {"version": 1, "grants": [], "extra": 1}, {"version": "2.35.10", "mails": []}, {"version": 1, "grants": [entry, entry]}, {"version": 1, "grants": [{**entry, "sent_at": True}]}, {"version": 1, "grants": [{**entry, "extra": 1}]}, {"version": 1, "grants": [{**entry, "rewards": []}]}, {"version": 1, "grants": [{**entry, "rewards": [{"type": 4, "id": False, "count": 1}]}]}, {"version": 1, "grants": [{**entry, "rewards": [{"type": 4, "id": 0, "count": True}]}]}, ] self.output.parent.mkdir(parents=True) for value in invalid_values: original = json.dumps(value).encode() self.output.write_bytes(original) with self.subTest(value=value), self.assertRaises(ValueError): self.run_grant() self.assertEqual(self.output.read_bytes(), original) self.output.write_text("broken JSON", encoding="utf-8") with self.assertRaises(ValueError): self.run_grant() self.assertEqual(self.output.read_text(encoding="utf-8"), "broken JSON") def test_failed_atomic_replace_preserves_previous_file_and_cleans_temporary(self): self.run_grant("--identity", "one") original = self.output.read_bytes() with mock.patch.object(dev_mail_grant.os, "replace", side_effect=OSError("injected failure")): with self.assertRaises(OSError): self.run_grant("--identity", "two") self.assertEqual(self.output.read_bytes(), original) self.assertEqual(list(self.output.parent.glob("*.tmp")), []) def test_cli_uses_standard_library_and_reports_validation_errors(self): result = subprocess.run(self.child("--identity", "standalone"), capture_output=True, text=True, timeout=20) self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(json.loads(result.stdout)["created"]) result = subprocess.run(self.child("--identity", "standalone", "--title", "conflict"), capture_output=True, text=True, timeout=20) self.assertEqual(result.returncode, 1) self.assertIn("dev_tools:", result.stderr) def test_process_waits_for_lock_then_reads_latest_file(self): with dev_mail_grant.grant_file_lock(self.output): process = subprocess.Popen(self.child("--identity", "child"), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) try: with self.assertRaises(subprocess.TimeoutExpired): process.wait(timeout=0.5) entry = {"identity": "parent", "title": "title", "body": "body", "sent_at": 1, "rewards": [{"type": 4, "id": 0, "count": 1}]} dev_mail_grant.atomic_json(self.output, {"version": 1, "grants": [entry]}) except BaseException: process.kill() process.communicate() raise stdout, stderr = process.communicate(timeout=20) self.assertEqual(process.returncode, 0, stderr) self.assertTrue(json.loads(stdout)["created"]) self.assertEqual([item["identity"] for item in dev_mail_grant.load_grants(self.output)["grants"]], ["parent", "child"]) def test_parallel_processes_preserve_all_grants_and_shared_identity_once(self): identities = [f"child-{i}" for i in range(8)] + ["shared"] * 4 processes = [subprocess.Popen(self.child("--identity", identity), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) for identity in identities] try: results = [] for process in processes: stdout, stderr = process.communicate(timeout=30) self.assertEqual(process.returncode, 0, stderr) results.append(json.loads(stdout)) grants = dev_mail_grant.load_grants(self.output)["grants"] self.assertEqual({entry["identity"] for entry in grants}, set(identities)) self.assertEqual(len(grants), 9) self.assertEqual(sum(result["created"] for result in results), 9) finally: for process in processes: if process.poll() is None: process.kill() process.communicate() if __name__ == "__main__": unittest.main()