test_generate_rnr_features.py 文件源码

def _delete_test_cluster(self, cluster_id):
        self.logger.info("Attempting to clean up the test cluster that was spun up for the unit test: %s" % cluster_id)
        config = load_config()
        url, user_id, password = get_rnr_credentials(config)

        response = requests.get('%s/v1/solr_clusters/%s' % (url, cluster_id),
                                auth=(user_id, password),
                                headers={'x-global-transaction-id': 'Rishavs app',
                                         'Content-type': 'application/json'})
        response_text = json.dumps(response.json(), indent=4, sort_keys=True)
        if response.status_code == 200:
            self.logger.info('Found a test cluster that needs cleanup: %s' % response_text)
            response = requests.delete('%s/v1/solr_clusters/%s' % (url, cluster_id),
                                       auth=(user_id, password),
                                       headers={'x-global-transaction-id': 'Rishavs app',
                                                'Content-type': 'application/json'})
            response.raise_for_status()
            self.logger.info("Successfully deleted test cluster: %s" % cluster_id)
        else:
            self.logger.info('No cleanup required for cluster id: %s (got response: %s)' % (cluster_id, response_text))