Module tests.test_stress

Functions

def backup_create_and_record_md5sum(client, core_api, volume_name, pod_name, snapshots_md5sum)
def check_and_set_backupstore(client)
def clean_volume_backups(client, volume_name)
def create_recurring_jobs(client, volume_name)
def delete_data(k8s_api_client, pod_name)
def delete_random_snapshot(client, volume_name, snapshots_md5sum)
def delete_replica(client, volume_name)
def generate_load(request)
def get_data_filename(pod_name)
def get_md5sum(file_path)
def get_random_backup_snapshot_data(snapshots_md5sum)
def get_random_snapshot(snapshots_md5sum)
def get_random_suffix()
def get_recurring_jobs()
def purge_random_snapshot(longhorn_api_client, volume_name, snapshot_name)
def read_data_md5sum(k8s_api_client, pod_name)
def remove_datafile(pod_name)
def restore_and_check_random_backup(client, core_api, volume_name, pod_name, snapshots_md5sum)
def revert_random_snapshot(client, core_api, volume_name, pod_manifest, snapshots_md5sum)
def snapshot_create_and_record_md5sum(client, core_api, volume_name, pod_name, snapshots_md5sum)
def test_reset_env()
def test_stress(generate_load)
def time_now()
def write_data(k8s_api_client, pod_name)

Classes

class snapshot_data (snapshot_name)
Expand source code
class snapshot_data:
    def __init__(self, snapshot_name):
        self.snapshot_name = snapshot_name
        self.removed = False
        self.backup_name = None
        self.backup_url = None
        self.data_md5sum = None

    def set_backup_name(self, backup_name):
        self.backup_name = backup_name

    def set_backup_url(self, backup_url):
        self.backup_url = backup_url

    def set_data_md5sum(self, data_md5sum):
        self.data_md5sum = data_md5sum

    def mark_as_removed(self):
        self.removed = True

Methods

def mark_as_removed(self)
def set_backup_name(self, backup_name)
def set_backup_url(self, backup_url)
def set_data_md5sum(self, data_md5sum)