KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use check_values method in autotest

Best Python code snippet using autotest_python

sql.py

Source:sql.py

1#2# Licensed to the Apache Software Foundation (ASF) under one3# or more contributor license agreements.  See the NOTICE file4# distributed with this work for additional information5# regarding copyright ownership.  The ASF licenses this file6# to you under the Apache License, Version 2.0 (the7# "License"); you may not use this file except in compliance8# with the License.  You may obtain a copy of the License at9#10#   http://www.apache.org/licenses/LICENSE-2.011#12# Unless required by applicable law or agreed to in writing,13# software distributed under the License is distributed on an14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY15# KIND, either express or implied.  See the License for the16# specific language governing permissions and limitations17# under the License.18from typing import Any, Dict, Optional, Union19from airflow.exceptions import AirflowException20from airflow.operators.sql import BaseSQLOperator21def parse_boolean(val: str) -> Union[str, bool]:22    """Try to parse a string into boolean.23    Raises ValueError if the input is not a valid true- or false-like string value.24    """25    val = val.lower()26    if val in ('y', 'yes', 't', 'true', 'on', '1'):27        return True28    if val in ('n', 'no', 'f', 'false', 'off', '0'):29        return False30    raise ValueError(f"{val!r} is not a boolean-like string value")31def _get_failed_tests(checks):32    return [33        f"\tCheck: {check},\n\tCheck Values: {check_values}\n"34        for check, check_values in checks.items()35        if not check_values["success"]36    ]37class SQLColumnCheckOperator(BaseSQLOperator):38    """39    Performs one or more of the templated checks in the column_checks dictionary.40    Checks are performed on a per-column basis specified by the column_mapping.41    Each check can take one or more of the following options:42    - equal_to: an exact value to equal, cannot be used with other comparison options43    - greater_than: value that result should be strictly greater than44    - less_than: value that results should be strictly less than45    - geq_to: value that results should be greater than or equal to46    - leq_to: value that results should be less than or equal to47    - tolerance: the percentage that the result may be off from the expected value48    :param table: the table to run checks on49    :param column_mapping: the dictionary of columns and their associated checks, e.g.50    .. code-block:: python51        {52            "col_name": {53                "null_check": {54                    "equal_to": 0,55                },56                "min": {57                    "greater_than": 5,58                    "leq_to": 10,59                    "tolerance": 0.2,60                },61                "max": {"less_than": 1000, "geq_to": 10, "tolerance": 0.01},62            }63        }64    :param conn_id: the connection ID used to connect to the database65    :param database: name of database which overwrite the defined one in connection66    .. seealso::67        For more information on how to use this operator, take a look at the guide:68        :ref:`howto/operator:SQLColumnCheckOperator`69    """70    column_checks = {71        "null_check": "SUM(CASE WHEN column IS NULL THEN 1 ELSE 0 END) AS column_null_check",72        "distinct_check": "COUNT(DISTINCT(column)) AS column_distinct_check",73        "unique_check": "COUNT(column) - COUNT(DISTINCT(column)) AS column_unique_check",74        "min": "MIN(column) AS column_min",75        "max": "MAX(column) AS column_max",76    }77    def __init__(78        self,79        *,80        table: str,81        column_mapping: Dict[str, Dict[str, Any]],82        conn_id: Optional[str] = None,83        database: Optional[str] = None,84        **kwargs,85    ):86        super().__init__(conn_id=conn_id, database=database, **kwargs)87        for checks in column_mapping.values():88            for check, check_values in checks.items():89                self._column_mapping_validation(check, check_values)90        self.table = table91        self.column_mapping = column_mapping92        # OpenLineage needs a valid SQL query with the input/output table(s) to parse93        self.sql = f"SELECT * FROM {self.table};"94    def execute(self, context=None):95        hook = self.get_db_hook()96        failed_tests = []97        for column in self.column_mapping:98            checks = [*self.column_mapping[column]]99            checks_sql = ",".join([self.column_checks[check].replace("column", column) for check in checks])100            self.sql = f"SELECT {checks_sql} FROM {self.table};"101            records = hook.get_first(self.sql)102            if not records:103                raise AirflowException(f"The following query returned zero rows: {self.sql}")104            self.log.info(f"Record: {records}")105            for idx, result in enumerate(records):106                tolerance = self.column_mapping[column][checks[idx]].get("tolerance")107                self.column_mapping[column][checks[idx]]["result"] = result108                self.column_mapping[column][checks[idx]]["success"] = self._get_match(109                    self.column_mapping[column][checks[idx]], result, tolerance110                )111            failed_tests.extend(_get_failed_tests(self.column_mapping[column]))112        if failed_tests:113            raise AirflowException(114                f"Test failed.\nQuery:\n{self.sql}\nResults:\n{records!s}\n"115                "The following tests have failed:"116                f"\n{''.join(failed_tests)}"117            )118        self.log.info("All tests have passed")119    def _get_match(self, check_values, record, tolerance=None) -> bool:120        match_boolean = True121        if "geq_to" in check_values:122            if tolerance is not None:123                match_boolean = record >= check_values["geq_to"] * (1 - tolerance)124            else:125                match_boolean = record >= check_values["geq_to"]126        elif "greater_than" in check_values:127            if tolerance is not None:128                match_boolean = record > check_values["greater_than"] * (1 - tolerance)129            else:130                match_boolean = record > check_values["greater_than"]131        if "leq_to" in check_values:132            if tolerance is not None:133                match_boolean = record <= check_values["leq_to"] * (1 + tolerance) and match_boolean134            else:135                match_boolean = record <= check_values["leq_to"] and match_boolean136        elif "less_than" in check_values:137            if tolerance is not None:138                match_boolean = record < check_values["less_than"] * (1 + tolerance) and match_boolean139            else:140                match_boolean = record < check_values["less_than"] and match_boolean141        if "equal_to" in check_values:142            if tolerance is not None:143                match_boolean = (144                    check_values["equal_to"] * (1 - tolerance)145                    <= record146                    <= check_values["equal_to"] * (1 + tolerance)147                ) and match_boolean148            else:149                match_boolean = record == check_values["equal_to"] and match_boolean150        return match_boolean151    def _column_mapping_validation(self, check, check_values):152        if check not in self.column_checks:153            raise AirflowException(f"Invalid column check: {check}.")154        if (155            "greater_than" not in check_values156            and "geq_to" not in check_values157            and "less_than" not in check_values158            and "leq_to" not in check_values159            and "equal_to" not in check_values160        ):161            raise ValueError(162                "Please provide one or more of: less_than, leq_to, "163                "greater_than, geq_to, or equal_to in the check's dict."164            )165        if "greater_than" in check_values and "less_than" in check_values:166            if check_values["greater_than"] >= check_values["less_than"]:167                raise ValueError(168                    "greater_than should be strictly less than "169                    "less_than. Use geq_to or leq_to for "170                    "overlapping equality."171                )172        if "greater_than" in check_values and "leq_to" in check_values:173            if check_values["greater_than"] >= check_values["leq_to"]:174                raise ValueError(175                    "greater_than must be strictly less than leq_to. "176                    "Use geq_to with leq_to for overlapping equality."177                )178        if "geq_to" in check_values and "less_than" in check_values:179            if check_values["geq_to"] >= check_values["less_than"]:180                raise ValueError(181                    "geq_to should be strictly less than less_than. "182                    "Use leq_to with geq_to for overlapping equality."183                )184        if "geq_to" in check_values and "leq_to" in check_values:185            if check_values["geq_to"] > check_values["leq_to"]:186                raise ValueError("geq_to should be less than or equal to leq_to.")187        if "greater_than" in check_values and "geq_to" in check_values:188            raise ValueError("Only supply one of greater_than or geq_to.")189        if "less_than" in check_values and "leq_to" in check_values:190            raise ValueError("Only supply one of less_than or leq_to.")191        if (192            "greater_than" in check_values193            or "geq_to" in check_values194            or "less_than" in check_values195            or "leq_to" in check_values196        ) and "equal_to" in check_values:197            raise ValueError(198                "equal_to cannot be passed with a greater or less than "199                "function. To specify 'greater than or equal to' or "200                "'less than or equal to', use geq_to or leq_to."201            )202class SQLTableCheckOperator(BaseSQLOperator):203    """204    Performs one or more of the checks provided in the checks dictionary.205    Checks should be written to return a boolean result.206    :param table: the table to run checks on207    :param checks: the dictionary of checks, e.g.:208    .. code-block:: python209        {210            "row_count_check": {"check_statement": "COUNT(*) = 1000"},211            "column_sum_check": {"check_statement": "col_a + col_b < col_c"},212        }213    :param conn_id: the connection ID used to connect to the database214    :param database: name of database which overwrite the defined one in connection215    .. seealso::216        For more information on how to use this operator, take a look at the guide:217        :ref:`howto/operator:SQLTableCheckOperator`218    """219    sql_check_template = "CASE WHEN check_statement THEN 1 ELSE 0 END AS check_name"220    sql_min_template = "MIN(check_name)"221    def __init__(222        self,223        *,224        table: str,225        checks: Dict[str, Dict[str, Any]],226        conn_id: Optional[str] = None,227        database: Optional[str] = None,228        **kwargs,229    ):230        super().__init__(conn_id=conn_id, database=database, **kwargs)231        self.table = table232        self.checks = checks233        # OpenLineage needs a valid SQL query with the input/output table(s) to parse234        self.sql = f"SELECT * FROM {self.table};"235    def execute(self, context=None):236        hook = self.get_db_hook()237        check_names = [*self.checks]238        check_mins_sql = ",".join(239            self.sql_min_template.replace("check_name", check_name) for check_name in check_names240        )241        checks_sql = ",".join(242            [243                self.sql_check_template.replace("check_statement", value["check_statement"]).replace(244                    "check_name", check_name245                )246                for check_name, value in self.checks.items()247            ]248        )249        self.sql = f"SELECT {check_mins_sql} FROM (SELECT {checks_sql} FROM {self.table});"250        records = hook.get_first(self.sql)251        if not records:252            raise AirflowException(f"The following query returned zero rows: {self.sql}")253        self.log.info(f"Record: {records}")254        for check in self.checks.keys():255            for result in records:256                self.checks[check]["success"] = parse_boolean(str(result))257        failed_tests = _get_failed_tests(self.checks)258        if failed_tests:259            raise AirflowException(260                f"Test failed.\nQuery:\n{self.sql}\nResults:\n{records!s}\n"261                "The following tests have failed:"262                f"\n{', '.join(failed_tests)}"263            )...

98-validate-binary-search-tree.py

Source:98-validate-binary-search-tree.py

1# first solution2class Solution:3    def check_values(self, root, value, direction):4        print(root.val, value, direction)5        if direction == "right" and root.val <= value:6            return False7        elif direction == "left" and root.val >= value:8            return False9        if root.left and root.right:10            return (11                self.check_values(root.left, value, direction) and 12                self.check_values(root.right, value, direction)13            )14        elif root.left:15            return self.check_values(root.left, value, direction)16        elif root.right:17            return self.check_values(root.right, value, direction)18        else:19            return True20    21    def isValidBST(self, root: Optional[TreeNode]) -> bool:22        if root.left and root.right:23            return (24                self.check_values(root.left, root.val, "left") and 25                self.check_values(root.right, root.val, "right") and26                self.isValidBST(root.left) if root.val > root.left.val else False and27                self.isValidBST(root.right) if root.val < root.right.val else False28            )29        elif root.left:30            return (31                self.check_values(root.left, root.val, "left") and32                self.isValidBST(root.left) if root.val > root.left.val else False33            )34        elif root.right:35            return (36                self.check_values(root.right, root.val, "right") and37                self.isValidBST(root.right) if root.val < root.right.val else False38            )39        else:...

math.py

Source:math.py

1import civilpy2np = civilpy.np3solver = civilpy.sym4inf = civilpy.sym.oo5def equation_solver(eq, variable):6    return civilpy.sym.solve(eq, variable)7def arithmetic_growth(p_0, r, t):8    p_t = p_0 + (t * r)9    return p_t10def geometric_growth(p_0, r, t):11    p_t = p_0 * ((1 + r) ** t)12    return p_t13def find_global_extrema_order_3(f, x1, x2, variable='x'):14    x = civilpy.sym.symbols(variable)15    f_prime = civilpy.sym.diff(f)16    f_prime2 = civilpy.sym.diff(f_prime)17    print(f'The first derivative is \n{f_prime} the second is \n {f_prime2}\n')18    try:19        root1, root2 = civilpy.sym.solvers.solve(f_prime, x)20        print(f'The first root is {root1}, the second root is {root2}\n')21        for root in [root1, root2]:22            if f_prime2.subs(x, root) > 0:23                print(f'{root} is a local minimum\n')24            elif f_prime2.subs(x, root) < 0:25                print(f'{root} is a local maximum\n')26        inf_point = civilpy.sym.solvers.solve(f_prime2, x)27        print(f'The inflection point is {inf_point[0]}\n')28        check_values = {x1: 0, x2: 0, root1: 0, root2: 0}29        for key, value in check_values.items():30            check_values[key] = f.subs(x, key)31        global_min = min(check_values.keys(), key=(lambda k: check_values[k]))32        global_max = max(check_values.keys(), key=(lambda k: check_values[k]))33        print(f'Global min: {global_min}, Global max: {global_max}')34        return global_min, global_max, check_values35    except ValueError:...

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.