112 lines
4.7 KiB
Python
112 lines
4.7 KiB
Python
from datetime import datetime, timedelta
|
|
from unittest.mock import Mock, patch
|
|
|
|
import jingrow
|
|
from jingrow.tests.utils import FrappeTestCase
|
|
|
|
from jcloude.jcloude.audit import BackupRecordCheck, OffsiteBackupCheck
|
|
from jcloude.jcloude.pagetype.agent_job.agent_job import AgentJob
|
|
from jcloude.jcloude.pagetype.jcloude_settings.test_jcloude_settings import (
|
|
create_test_jcloude_settings,
|
|
)
|
|
from jcloude.jcloude.pagetype.site.test_site import create_test_site
|
|
from jcloude.jcloude.pagetype.site_activity.site_activity import log_site_activity
|
|
from jcloude.jcloude.pagetype.site_backup.test_site_backup import create_test_site_backup
|
|
from jcloude.jcloude.pagetype.telegram_message.telegram_message import TelegramMessage
|
|
|
|
|
|
@patch.object(TelegramMessage, "enqueue", new=Mock())
|
|
@patch.object(AgentJob, "enqueue_http_request", new=Mock())
|
|
class TestBackupRecordCheck(FrappeTestCase):
|
|
def tearDown(self):
|
|
jingrow.db.rollback()
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
|
|
self.yesterday = jingrow.utils.now_datetime().date() - timedelta(days=1)
|
|
self._2_hrs_before_yesterday = datetime.combine(self.yesterday, datetime.min.time()) - timedelta(
|
|
hours=2
|
|
)
|
|
|
|
def test_audit_will_fail_if_backup_older_than_interval(self):
|
|
create_test_jcloude_settings()
|
|
site = create_test_site(creation=self._2_hrs_before_yesterday)
|
|
create_test_site_backup(site.name, creation=self._2_hrs_before_yesterday + timedelta(hours=1))
|
|
BackupRecordCheck()
|
|
audit_log = jingrow.get_last_pg("Audit Log", {"audit_type": BackupRecordCheck.audit_type})
|
|
self.assertEqual(audit_log.status, "Failure")
|
|
|
|
def test_audit_succeeds_when_backup_in_interval_exists(self):
|
|
create_test_jcloude_settings()
|
|
site = create_test_site(creation=self._2_hrs_before_yesterday)
|
|
|
|
create_test_site_backup(
|
|
site.name,
|
|
creation=self._2_hrs_before_yesterday + timedelta(hours=3),
|
|
)
|
|
BackupRecordCheck()
|
|
audit_log = jingrow.get_last_pg("Audit Log", {"audit_type": BackupRecordCheck.audit_type})
|
|
self.assertEqual(audit_log.status, "Success")
|
|
|
|
def test_audit_log_is_created(self):
|
|
create_test_jcloude_settings()
|
|
site = create_test_site(creation=self._2_hrs_before_yesterday)
|
|
create_test_site_backup(site.name, creation=self.yesterday)
|
|
audit_logs_before = jingrow.db.count("Audit Log", {"audit_type": BackupRecordCheck.audit_type})
|
|
BackupRecordCheck()
|
|
audit_logs_after = jingrow.db.count("Audit Log", {"audit_type": BackupRecordCheck.audit_type})
|
|
self.assertGreater(audit_logs_after, audit_logs_before)
|
|
|
|
def test_sites_created_within_interval_are_ignored(self):
|
|
create_test_jcloude_settings()
|
|
create_test_site()
|
|
# no backup
|
|
BackupRecordCheck()
|
|
|
|
audit_log = jingrow.get_last_pg("Audit Log", {"audit_type": BackupRecordCheck.audit_type})
|
|
self.assertEqual(audit_log.status, "Success")
|
|
|
|
def test_sites_that_were_recently_activated_are_ignored(self):
|
|
create_test_jcloude_settings()
|
|
site = create_test_site(creation=self._2_hrs_before_yesterday)
|
|
act = log_site_activity(site.name, "Activate Site")
|
|
act.db_set("creation", self._2_hrs_before_yesterday + timedelta(hours=24))
|
|
BackupRecordCheck()
|
|
audit_log = jingrow.get_last_pg("Audit Log", {"audit_type": BackupRecordCheck.audit_type})
|
|
self.assertEqual(audit_log.status, "Success")
|
|
|
|
|
|
@patch.object(TelegramMessage, "enqueue", new=Mock())
|
|
@patch.object(AgentJob, "enqueue_http_request", new=Mock())
|
|
class TestOffsiteBackupCheck(FrappeTestCase):
|
|
def tearDown(self):
|
|
jingrow.db.rollback()
|
|
|
|
def test_audit_succeeds_when_all_remote_files_are_in_remote(self):
|
|
create_test_jcloude_settings()
|
|
site = create_test_site()
|
|
site_backup = create_test_site_backup(site.name)
|
|
jingrow.db.set_value("Remote File", site_backup.remote_database_file, "file_path", "remote_file1")
|
|
jingrow.db.set_value("Remote File", site_backup.remote_public_file, "file_path", "remote_file2")
|
|
jingrow.db.set_value("Remote File", site_backup.remote_private_file, "file_path", "remote_file3")
|
|
with patch.object(
|
|
OffsiteBackupCheck,
|
|
"_get_all_files_in_s3",
|
|
new=lambda x: ["remote_file1", "remote_file2", "remote_file3"],
|
|
):
|
|
OffsiteBackupCheck()
|
|
audit_log = jingrow.get_last_pg("Audit Log", {"audit_type": OffsiteBackupCheck.audit_type})
|
|
self.assertEqual(audit_log.status, "Success")
|
|
|
|
def test_audit_fails_when_all_remote_files_not_in_remote(self):
|
|
create_test_jcloude_settings()
|
|
site = create_test_site()
|
|
# 3 remote files are created here
|
|
site_backup = create_test_site_backup(site.name)
|
|
jingrow.db.set_value("Remote File", site_backup.remote_database_file, "file_path", "remote_file1")
|
|
with patch.object(OffsiteBackupCheck, "_get_all_files_in_s3", new=lambda x: ["remote_file1"]):
|
|
OffsiteBackupCheck()
|
|
audit_log = jingrow.get_last_pg("Audit Log", {"audit_type": OffsiteBackupCheck.audit_type})
|
|
self.assertEqual(audit_log.status, "Failure")
|