115 lines
5.3 KiB
Python
115 lines
5.3 KiB
Python
"""
|
||
بکاپِ زمانبندیشده: نگهداری، هوکِ بیرونی، هشدارِ 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])
|