How to use error_group method in grail

Best Python code snippet using grail_python

cloud_error_processing.py

Source:cloud_error_processing.py

1import os2import requests3import sys4from google.cloud import firestore5from googleapiclient.discovery import build6from google.cloud import secretmanager7# Add our library utils to the path8sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))9from lib.constants import ISSUES_API_URL10from lib.error_logger import ErrorLogger11class GithubIssueHandler(ErrorLogger):12    """ Handles posting issues to github using the personal access token stored in SecretsManager. """13    def __init__(self, gcs_project_name):14        self._username = "automation"  # This is ignored because we use an access token.15        self._gcs_project_name = gcs_project_name16        self._password = self._get_github_token()17    def _error_group_to_github_issue(self, error_group):18        title = "Automated error report"19        body = error_group["representative"]["message"]20        return {"title": title, "body": body}21    def _get_github_token(self):22        client = secretmanager.SecretManagerServiceClient()23        name = client.secret_version_path(self._gcs_project_name, "github-token", "latest")24        response = client.access_secret_version(name)25        return response.payload.data26    def post_error_to_github(self, error_group):27        """ Returns the issue url if successfully posted, else raises a ConnectionError exception """28        session = requests.Session()29        session.auth = (self._username, self._password)30        response = session.post(ISSUES_API_URL, json=self._error_group_to_github_issue(error_group))31        if response.status_code != 201:32            self.log_error("Could not create github issue.", status_code=response.status_code)33            raise ConnectionError()34        return response.json()["html_url"]35def register_new_errors(gcs_project_name):36    """ If new error groups are reported, log an issue on github with the details """37    service = build("clouderrorreporting", "v1beta1")38    errors_past_day = (39        service.projects()40        .groupStats()41        .list(projectName="projects/{}".format(gcs_project_name), timeRange_period="PERIOD_1_DAY")42        .execute()43    )44    gh_issue_handler = GithubIssueHandler(gcs_project_name)45    db = firestore.Client()46    errors_db = db.collection("errors")47    for error_group in errors_past_day["errorGroupStats"]:48        group_id = error_group["group"]["groupId"]49        if int(error_group["count"]) < 2:50            # Don't add one-off errors to the db51            continue52        doc = errors_db.document(group_id)53        if not doc.get().exists and error_group["group"]["resolutionStatus"] == "OPEN":54            try:55                error_group["group"]["trackingIssues"] = [56                    {"url": gh_issue_handler.post_error_to_github(error_group)}57                ]58                error_group["group"]["resolutionStatus"] = "ACKNOWLEDGED"59            except ConnectionError:60                # Could not create an issue61                # Don't add it to our known errors db, we can retry on the next scheduled job.62                continue63            # Now set it to acknowledged and link to the issue in Cloud Error Reporting.64            service.projects().groups().update(65                name=error_group["group"]["name"], body=error_group["group"]66            ).execute()67        doc.set(error_group)68if __name__ == "__main__":69    import os...

utils.py

Source:utils.py

1import numpy2def check_results_adequacy(results, error_group, check_nan=True):3    error_group.add_message("invalid_results")4    error_group.invalid_results.clear()5    def anynan(a):6        return numpy.any(numpy.isnan(a))7    if results is None:8        return None9    if results.data is None:10        error_group.invalid_results("Results do not include information on test data")11    elif not results.data.domain.has_discrete_class:12        error_group.invalid_results("Discrete outcome variable is required")13    elif check_nan and (14        anynan(results.actual)15        or anynan(results.predicted)16        or (results.probabilities is not None and anynan(results.probabilities))17    ):18        error_group.invalid_results("Results contains invalid values")19    else:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.