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)