Module harvester_e2e_tests.conftest
Functions
def check_depends(self, depends, item)-
Expand source code
def check_depends(self, depends, item): # monkey patch `DependencyManager.checkDepends` marker = item.get_closest_marker("dependency") if marker and marker.kwargs.get('param'): try: param_id = item.callspec.id depends = [f"{d}[{param_id}]" for d in depends] except AttributeError: pass # ref: https://github.com/RKrahl/pytest-dependency/issues/57#issuecomment-1000896418 if marker and marker.kwargs.get('any'): for depend in depends: try: self._check_depend([depend], item) except pytest.skip.Exception: continue else: return pytest.skip("%s depends on any of %s" % (item.name, ", ".join(depends))) else: self._check_depend(depends, item) def pytest_addoption(parser)-
Expand source code
def pytest_addoption(parser): with open('config.yml') as f: config_data = yaml.safe_load(f) parser.addoption( '--endpoint', action='store', default=config_data['endpoint'], help='Harvester API endpoint' ) parser.addoption( '--username', action='store', default=config_data['username'], help='Harvester username' ) parser.addoption( '--password', action='store', default=config_data['password'], help='Harvester password' ) parser.addoption( '--host-password', action='store', default=config_data['host-password'], help='Password to access Harvesrer node' ) parser.addoption( '--host-private-key', action='store', default=config_data['host-private-key'], help='private key to access Harvester node' ) parser.addoption( '--vlan-id', action='store', type=int, default=config_data['vlan-id'], help=('VLAN ID, if specified, will invoke the tests depended on ' 'external networking.') ) parser.addoption( '--vlan-nic', action='store', default=config_data['vlan-nic'], help='Physical NIC for VLAN. Default is "eth0"' ) parser.addoption( '--ip-pool-subnet', action='store', default=config_data['ip-pool-subnet'], help='Subnet of IP pool for load balancer' ) parser.addoption( '--ip-pool-start', action='store', default=config_data['ip-pool-start'], help='Start IP of IP pool for load balancer' ) parser.addoption( '--ip-pool-end', action='store', default=config_data['ip-pool-end'], help='End IP of IP pool for load balancer' ) parser.addoption( '--wait-timeout', action='store', type=int, default=config_data['wait-timeout'], help='Wait time for polling operations' ) parser.addoption( '--sleep-timeout', action='store', type=int, default=config_data['sleep-timeout'], help='Wait time for polling operations' ) parser.addoption( '--node-scripts-location', action='store', default=config_data['node-scripts-location'], help=('External scripts to power-off, power-up, and reboot a given ' 'Harvester node to facilitate the host-specific tests') ) parser.addoption( '--opensuse-image-url', action='store', default=config_data.get('opensuse-image-url'), help=('OpenSUSE image URL') ) parser.addoption( '--ubuntu-image-url', action='store', default=config_data.get('ubuntu-image-url'), help=('ubuntu image URL') ) parser.addoption( '--terraform-scripts-location', action='store', default=config_data['terraform-scripts-location'], help=('External scripts to create resources using terraform') ) parser.addoption( '--image-cache-url', action='store', default=config_data['image-cache-url'], help=('URL for the local images cache') ) parser.addoption( '--accessKeyId', action='store', default=config_data['accessKeyId'], help=('A user-id that uniquely identifies your account. ') ) parser.addoption( '--secretAccessKey', action='store', default=config_data['secretAccessKey'], help=('The password to your account') ) parser.addoption( '--bucketName', action='store', default=config_data['bucketName'], help=('Name of the bucket') ) parser.addoption( '--region', action='store', default=config_data['region'], help=('Region of the bucket') ) parser.addoption( '--s3-endpoint', action='store', default=config_data.get('s3-endpoint', ''), help=('S3 endpoint') ) parser.addoption( '--nfs-endpoint', action='store', default=config_data['nfs-endpoint'], help=('Endpoint for storing backup in nfs share') ) parser.addoption( '--rancher-endpoint', action='store', default=config_data.get('rancher-endpoint', None), help='Rancher API endpoint' ) parser.addoption( '--rancher-admin-password', action='store', default=config_data.get('rancher-admin-password', None), help='Rancher admin user password' ) parser.addoption( '--k8s-version', action='store', default=config_data.get('k8s-version'), help='K8s version to use for downstream cluster in Rancher integration tests' ) parser.addoption( '--rancher-cluster-wait-timeout', action='store', type=int, default=config_data['rancher-cluster-wait-timeout'], help='Wait time for polling Rancher cluster ready status' ) parser.addoption( '--nfs-mount-dir', action='store', default=config_data['nfs-mount-dir'], help=('mount directory for nfs share') ) parser.addoption( '--upgrade-prepare-dependence', action='store', default=config_data['upgrade-prepare-dependence'], help=('If true to prepare dependence') ) parser.addoption( '--upgrade-sc-replicas', action='store', default=config_data['upgrade-sc-replicas'], help=('default storage class replicas') ) parser.addoption( '--upgrade-target-version', action='store', default=config_data['upgrade-target-version'], help=('test target harvester version') ) parser.addoption( '--upgrade-iso-url', action='store', default=config_data['upgrade-iso-url'], help=('URL for specific iso') ) parser.addoption( '--upgrade-iso-checksum', action='store', default=config_data['upgrade-iso-checksum'], help=('URL for specific iso checksum') ) parser.addoption( '--upgrade-wait-timeout', action='store', default=config_data['upgrade-wait-timeout'], help=('Wait time for polling upgrade Harvester cluster completed status') ) parser.addoption( '--terraform-provider-harvester', action='store', default=config_data.get('terraform-provider-harvester'), help=('Version of Terraform Harvester Provider') ) parser.addoption( '--terraform-provider-rancher', action='store', default=config_data.get('terraform-provider-rancher'), help=('Version of Terraform Rancher Provider') ) parser.addoption( '--ubuntu-checksum', action='store', default=config_data.get('ubuntu-checksum'), help=('Checksum for ubuntu_image') ) parser.addoption( '--opensuse-checksum', action='store', default=config_data.get('opensuse-checksum'), help=('Checksum for opensuse_image') ) def pytest_collection_modifyitems(session, config, items)-
Expand source code
@pytest.hookimpl(hookwrapper=True) def pytest_collection_modifyitems(session, config, items): if config.getoption('--vlan-id') == -1: skip_public_network = pytest.mark.skip(reason=( 'VM not accessible because no VLAN setup with public routing')) for item in items: if 'public_network' in item.keywords: item.add_marker(skip_public_network) if (config.getoption('--nfs-endpoint') == '' and config.getoption('--accessKeyId') == ''): skip_backup = pytest.mark.skip(reason=( 'AWS credentials or NFS endpoint are not available')) for item in items: if 'backup' in item.keywords: item.add_marker(skip_backup) if config.getoption('--accessKeyId') == '': skip_backup = pytest.mark.skip(reason=( 'AWS credentials are not available')) for item in items: if 'backups3' in item.keywords: item.add_marker(skip_backup) if config.getoption('--nfs-endpoint') == '': skip_backup = pytest.mark.skip(reason=( 'NFS endpoint is not available')) for item in items: if 'backupnfs' in item.keywords: item.add_marker(skip_backup) if (not config.getoption('--rancher-endpoint') and not config.getoption('--rancher-admin-password')): skip_rancher_integration_external = pytest.mark.skip(reason=( 'Rancher endpoint and admin password are not specified')) for item in items: if 'rancher_integration_with_external_rancher' in item.keywords: item.add_marker(skip_rancher_integration_external) if not config.getoption("--delete-host", False): for item in items: if "delete_host" in item.keywords: item.add_marker(pytest.mark.skip(reason="Not configured to test host deletion.")) # legacy code above # ''' To enable the test select with `and depends` keyword, # to select test cases and it depended test cases. # ''' if "and depends" not in config.option.keyword: # DO nothing yield return all_items, old_keyword = items.copy(), config.option.keyword config.option.keyword = old_keyword.replace('and depends', '') yield scope_cls = { "session": pytest.Session, "package": pytest.Package, "module": pytest.Module, "class": pytest.Class } # ref: https://github.com/RKrahl/pytest-dependency/blob/0.5.1/pytest_dependency.py#L77 # named_items : dict[('dep-scope', 'dep-name'), list[(idx, 'test-item')]] # picked : list[(idx, 'test-item')] named_items, picked = dict(), list() for idx, item in enumerate(all_items): if item in items: picked.append((idx, item)) try: marker = item.get_closest_marker('dependency') scope = marker.kwargs.get('scope', 'module') node = item.getparent(scope_cls[scope]) named_items.setdefault((node, marker.kwargs['name']), []).append((idx, item)) except AttributeError: continue except KeyError: nodeid = item.nodeid.replace("::()::", "::") if scope not in ("session", "package"): shift = 2 if scope == "class" else 1 nodeid = nodeid.split("::", shift)[shift] named_items.setdefault((node, nodeid), []).append((idx, item)) def pick_depends(idx, item, items): picked = [] try: marker = item.get_closest_marker('dependency') scope = marker.kwargs.get('scope', 'module') node = item.getparent(scope_cls[scope]) for name in marker.kwargs['depends']: picked.extend(items.get((node, name), [])) except (AttributeError, KeyError): pass return picked depends = [] # list[(idx, 'test-item')] while picked: depends.append(picked.pop()) picked.extend(pick_depends(*depends[-1], named_items)) session.items = items = [it for idx, it in sorted(set(depends), key=lambda it: it[0])] # deselected.extend(t for ts in named_items.values() for i, t in ts if t not in items) deselected = [t for t in all_items if t not in items] # update to let the report shows correct counts config.pluginmanager.get_plugin('terminalreporter').stats['deselected'] = deselected def pytest_configure(config)-
Expand source code
def pytest_configure(config): # Register marker as the format (marker, (description)) markers = [ ("skip_version_if", "Mark test skipped when cluster version hit the condition"), ("skip_version_before", ( "mark test skipped when cluster version < provided version")), ("skip_version_after", ( "mark test skipped when cluster version >= provided version")), ('p0', ("mark the test's priority is p0")), ('p1', ("mark the test's priority is p1")), ('p2', ("mark the test's priority is p2")), ('hosts', ('{_r} host tests')), ('delete_host', ('{_r} host and will delete one of hosts')), ("negative", ("{_r} a negative tests")), ('keypairs', ("{_r} SSH keypairs tests")), ('images', ("{_r} image tests")), ("networks", ("{_r} vlan network tests")), ("volumes", ("{_r} volume tests")), ("virtualmachines", ("{_r} VM tests")), ("templates", ("{_r} VM template tests")), ("support_bundle", ("{_r} Support Bundle tests")), ("settings", ("{_r} settings tests")), ("upgrade", ("{_r} upgrade tests")), ("any_nodes", ("{_r} tests which could be ran on clushter with any nodes")), ("single_node", ("{_r} tests which could only be ran on cluster with single node")), ("three_nodes", ("{_r} tests which could only be ran on cluster with three nodes")), ('rancher', ("{_r} rancher integration tests")), ('rke1', ("{_r} rancher RKE1 tests")), ('rke2', ("{_r} rancher RKE2 tests")), ('terraform', ("{_r} terraform tests")), ('virtualmachines', ('{_r} virtualmachines tests')), ('backup_target', ('{_r} backup-target tests')), ('S3', ('{_r} backup-target tests with S3')), ('NFS', ('{_r} backup-target tests with NFS')) ] for m, msg in markers: related = 'mark the test is related to' config.addinivalue_line("markers", f"{m}:{msg.format(_r=related)}") def pytest_html_results_table_header(cells)-
Expand source code
def pytest_html_results_table_header(cells): cells.insert(1, '<th class="sortable time" data-column-type="time">EndTime</th>') def pytest_html_results_table_row(report, cells)-
Expand source code
def pytest_html_results_table_row(report, cells): cells.insert(1, f'<td class="col-time">{datetime.utcnow()}</td>')