Gecko [ gtgates.com ]

Name	Size	Permission	Date
__init__.py	0 B	-rw-r--r--	2023-10-21 18:57
__init__.pyc	141 B	-rw-r--r--	2023-10-21 18:57
common.py	2.96 KB	-rw-r--r--	2023-10-21 18:57
common.pyc	4.42 KB	-rw-r--r--	2023-10-21 18:57
mock_backup.py	2.98 KB	-rw-r--r--	2023-10-21 18:57
mock_backup.pyc	4.07 KB	-rw-r--r--	2023-10-21 18:57
mock_sanity.py	3.54 KB	-rw-r--r--	2023-10-21 18:57
mock_sanity.pyc	3.36 KB	-rw-r--r--	2023-10-21 18:57

Rename

import random
import string

from hw_cpbackup.sanity import BackupSanity
from test.mock.common import S3_MOCKDATA

class MockBackupSanity(BackupSanity):
    """Mock BackupRoutine class"""

def __init__(self, s3_sesison=None):
        super(MockBackupSanity, self).__init__()
        self.alerted = False
        self.null_routed = False
        self.upload_complete = False
        self.dry_run = False
        self.no_alert = False
        self.s3_session = s3_sesison

@property
    def username(self):
        return 'mockuser'

@property
    def user_data(self):
        _user_data = []
        _user_data_obj = S3_MOCKDATA
        _user_data.append(_user_data_obj)
        return _user_data

@property
    def backup_archives(self):
        _backup_archives = ['%s.tar.gz' % self.username]
        for _ in range(10):
            _backup_archives.append('%s.tar.gz' % ''.join(
                random.SystemRandom().choice(string.ascii_letters.lower() + string.digits) for _ in range(8)))
        return _backup_archives

def user_archive(self, username):
        """Obtain archive for user"""
        for archive in self.backup_archives:
            if archive == username + '.tar.gz':
                return archive

def run(self, target_user=None):
        if self.user_data:
            for user_account in self.user_data:
                user = user_account['user']
                mock_s3_connection = self.s3_session
                if mock_s3_connection.bucket_exists:
                    if mock_s3_connection.storage_disabled:  # skip disabled users
                        continue
                    if not mock_s3_connection.backup_today:  # check for today backup
                        if not self.dry_run:
                            # mock_s3_connection.upload(archive=self.user_archive(username=user))
                            self.upload_complete = True
                    else:
                        self.upload_complete = True
                    if not mock_s3_connection.backup_keys:  # ensure backup entries after upload attempt
                        if not self.dry_run and not self.no_alert:
                            self.alerted = True
                        continue
                    if mock_s3_connection.backup_zero:  # if today backup size is zero, send alert
                        if not self.dry_run and not self.no_alert:
                            self.alerted = True
                        continue
                    if mock_s3_connection.exceeded_retention:  # if retention thresholds exceeded, remove excess
                        if not self.dry_run:
                            mock_s3_connection.enforce_retention()
                    if mock_s3_connection.exceeded_timestamp:  # if latest backup exceeds 72 hours
                        if not self.dry_run and not self.no_alert:
                            self.alerted = True
                        continue
                    if not mock_s3_connection.backup_today:  # if no backup for today
                        if not self.dry_run and not self.no_alert:
                            self.alerted = True
                        continue
                    if mock_s3_connection.dates_absent:  # if any backup absent prior 72 hours
                        continue
                else:
                    if not self.dry_run and not self.no_alert:
                        self.alerted = True

def reset(self):
        self.alerted = False
        self.null_routed = False
        self.upload_complete = False
        self.dry_run = False
        self.no_alert = False