Gecko [ gtgates.com ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2025-12-01 20:27
__init__.py	0 B	-rw-r--r--	2025-12-01 20:27
common.py	2.96 KB	-rw-r--r--	2025-12-01 20:27
mock_backup.py	2.98 KB	-rw-r--r--	2025-12-01 20:27
mock_sanity.py	3.54 KB	-rw-r--r--	2025-12-01 20:27

Rename

import random
import string
from datetime import datetime, timedelta

from boto.s3.connection import Key, S3ResponseError

S3_MOCKDATA = {
    'user': 'mockuser',
    'retention': 45,
    's3': {
        's3id': ''.join(random.SystemRandom().choice(string.ascii_letters + string.digits) for _ in range(20)),
        's3key': ''.join(random.SystemRandom().choice(string.ascii_letters + string.digits) for _ in range(40)),
        's3endpoint': '127.0.0.1',  # not a real endpoint
    }
}

class MockBucket:
    """Mock Boto S3 Bucket class object"""

def __init__(self, **kwargs):
        self.name = kwargs.get('name')
        self.size = kwargs.get('key_size')
        self.start_date = kwargs.get('start_date')
        self.key_count = kwargs.get('key_count')

self._deleted_keys = []

def __iter__(self):
        return iter(self.get_all_keys())

@property
    def deleted_keys(self):
        return self._deleted_keys

def get_all_keys(self):
        keys = []
        for i in range(self.start_date, self.key_count):
            key_prefix = 'mockuser/mockuser'
            new_key = Key(name='%s.%s' % (key_prefix, datetime.strftime(
                datetime.today() - timedelta(days=i), '.%m.%d.tar.gz')))
            new_key.last_modified = datetime.strftime(
                datetime.today() - timedelta(days=i), '%Y-%m-%d')
            new_key.size = self.size
            if new_key.name not in self.deleted_keys:
                keys.append(new_key)
        return keys

def delete_key(self, key_name):
        self._deleted_keys.append(key_name)

class S3Session:
    """Mock Boto S3 connection session"""

def __init__(self, boto_exception=False, exclude='new', exists=True, start_date=0, key_count=10, key_size=1024):
        self.exclude = exclude
        self.exists = exists
        self.boto_exception = boto_exception
        self.start_date = start_date
        self.key_count = key_count
        self.key_size = key_size

self.created_bucket = False

def head_bucket(self, bucket_name=None):
        if self.exists and self.boto_exception:
            raise S3ResponseError(status=403, reason='Mock 403 Unauthorized')
        elif self.exists and not self.boto_exception:
            return MockBucket(start_date=self.start_date, key_count=self.key_count, key_size=self.key_size)
        elif not self.exists:
            raise S3ResponseError(status=404, reason='Mock 404 Not Found')

def lookup(self, bucket_name):
        to_exclude = None
        if self.exclude == 'new':
            to_exclude = ['shared-backups']
        elif self.exclude == 'old':
            to_exclude = ['shared_backups']
        elif self.exclude == 'all':
            to_exclude = ['shared_backups', 'shared-backups']
        if bucket_name in to_exclude:
            return None
        else:
            return MockBucket(start_date=self.start_date, key_count=self.key_count, key_size=self.key_size)

def create_bucket(self, bucket_name):
        self.created_bucket = True