Scheduled Postgres backup, restore check, backup warnings in health, payment result logging

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

statuslog/tests_backup.py

@@ -0,0 +1,114 @@
+"""
+بکاپِ زمان‌بندی‌شده: نگه‌داری، هوکِ بیرونی، هشدارِ health، و لاگِ پرداخت.
+خودِ pg_dump/pg_restore اینجا اجرا نمی‌شود (تست‌ها روی SQLite‌اند)؛ آن را یک اجرای دستی روی Postgresِ
+لپ‌تاپ سنجید (شرحش در docs/operations.md). آزموده‌نشده: زیر cron روی سرور.
+"""
+import datetime
+import json
+import tempfile
+from pathlib import Path
+from unittest import mock
+
+from django.test import SimpleTestCase, TestCase, override_settings
+
+from statuslog import backup
+
+
+def _touch(d: Path, stamp: str):
+    f = d / f'db-{stamp}.pgdump'
+    f.write_bytes(b'x')
+    return f
+
+
+class PruneTests(SimpleTestCase):
+    def test_keeps_last_7_days_and_4_weeks_only(self):
+        with tempfile.TemporaryDirectory() as t:
+            d = Path(t)
+            start = datetime.datetime(2026, 8, 1, 2, 0, 0)
+            for i in range(60):  # شصت روز، هر روز یک بکاپ
+                _touch(d, (start + datetime.timedelta(days=i)).strftime('%Y%m%d-%H%M%S'))
+            removed = backup._prune(d)
+            left = sorted(p.name for p in d.glob('db-*.pgdump'))
+            self.assertEqual(len(left) + len(removed), 60)
+            self.assertIn('db-20260929-020000.pgdump', left)   # آخرین
+            self.assertLessEqual(len(left), 7 + 4)
+            self.assertGreaterEqual(len(left), 7)
+
+    def test_two_backups_same_day_keep_latest(self):
+        with tempfile.TemporaryDirectory() as t:
+            d = Path(t)
+            _touch(d, '20261001-020000')
+            _touch(d, '20261001-140000')
+            backup._prune(d)
+            self.assertEqual([p.name for p in d.glob('db-*')], ['db-20261001-140000.pgdump'])
+
+
+class OffsiteTests(SimpleTestCase):
+    def test_not_configured_is_explicit(self):
+        with mock.patch.dict('os.environ', {'BACKUP_OFFSITE_CMD': ''}):
+            self.assertEqual(backup.run_offsite(Path('/x')), {'state': 'not_configured'})
+
+    def test_placeholder_required(self):
+        with mock.patch.dict('os.environ', {'BACKUP_OFFSITE_CMD': 'rsync a b'}):
+            self.assertEqual(backup.run_offsite(Path('/x'))['state'], 'failed')
+
+    def test_failing_command_is_reported_not_swallowed(self):
+        with mock.patch.dict('os.environ', {'BACKUP_OFFSITE_CMD': 'false {file}'}):
+            self.assertEqual(backup.run_offsite(Path('/x'))['state'], 'failed')
+
+    def test_ok(self):
+        with mock.patch.dict('os.environ', {'BACKUP_OFFSITE_CMD': 'true {file}'}):
+            self.assertEqual(backup.run_offsite(Path('/x'))['state'], 'ok')
+
+
+class HealthWarningTests(SimpleTestCase):
+    def _warn(self, status, **settings_kw):
+        from config import health
+        with tempfile.TemporaryDirectory() as t, override_settings(BACKUP_DIR=t, DEBUG=False, **settings_kw):
+            if status is not None:
+                (Path(t) / 'status.json').write_text(json.dumps(status))
+            with mock.patch.object(health.connection, 'vendor', 'postgresql'):
+                return health._backup_warnings()
+
+    def test_no_status_warns(self):
+        self.assertTrue(any('هیچ بکاپ' in w for w in self._warn(None)))
+
+    def test_stale_backup_warns(self):
+        old = (datetime.datetime.now().astimezone() - datetime.timedelta(hours=40)).isoformat()
+        self.assertTrue(any('ساعت پیش' in w for w in self._warn({'last_ok_at': old})))
+
+    def test_fully_healthy_has_no_warning(self):
+        now = datetime.datetime.now().astimezone().isoformat()
+        st = {'last_ok_at': now, 'offsite': {'state': 'ok'}, 'restore_check': {'ok': True, 'at': now}}
+        self.assertEqual(self._warn(st), [])
+
+    def test_offsite_missing_and_restore_untested_warn(self):
+        now = datetime.datetime.now().astimezone().isoformat()
+        w = self._warn({'last_ok_at': now, 'offsite': {'state': 'not_configured'}})
+        self.assertTrue(any('بیرون از این ماشین' in x for x in w))
+        self.assertTrue(any('آزموده نشده' in x for x in w))
+
+    def test_sqlite_dev_is_silent(self):
+        from config import health
+        self.assertEqual(health._backup_warnings(), [])
+
+
+class PaymentAuditLogTests(TestCase):
+    def test_unknown_authority_callback_is_404_and_logged(self):
+        with self.assertLogs('payment.views', level='WARNING') as cm:
+            r = self.client.get('/api/payment/zarinpal/callback?Authority=A000NOPE&Status=OK')
+        self.assertEqual(r.status_code, 404)
+        self.assertIn('no order', cm.output[0])
+
+    def test_ignored_duplicate_paid_result_is_logged_as_warning(self):
+        from payment.models import Order
+        from payment.services import apply_order_payment_result
+        o = Order.objects.create(user_id=1, amount=1000, idempotency_key='k1', gateway_authority='A1')
+        with self.assertLogs('payment.services', level='INFO') as cm:
+            apply_order_payment_result(order=o, gateway='zarinpal', gateway_ref_id='R1',
+                                       new_status='paid', event='t')
+        self.assertIn('ignored=False', cm.output[0])
+        with self.assertLogs('payment.services', level='INFO') as cm:
+            apply_order_payment_result(order=o, gateway='zarinpal', gateway_ref_id='R1',
+                                       new_status='paid', event='t')
+        self.assertIn('ignored=True', cm.output[0])