Summary  
This chapter introduces how to define and run unit tests to verify code behavior.

General domain of usage  
Software testing

import unittest
import re
import psycopg2
import pandas as pd


def _dynamic_test(test_case, condition, success_message, failure_message):
    """Helper to dynamically rename and mark test results."""
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)


class TestPostgresQuery(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.connection = psycopg2.connect(
            host="localhost",
            user="agent",
            password="qwerty",
            dbname="testdb",
            sslmode="disable",
        )
        cls.cursor = cls.connection.cursor()

    @classmethod
    def tearDownClass(cls):
        cls.cursor.close()
        cls.connection.close()

    def test_query_contains_limit(self):
        """Check if query.sql contains LIMIT clause."""
        with open("query.sql", "r", encoding="utf-8") as file:
            query = file.read()

        condition = "LIMIT" in query.upper()
        _dynamic_test(
            self,
            condition,
            "Query contains LIMIT clause.",
            "The query.sql does not contain a LIMIT clause.",
        )

    def test_expected_rows_present(self):
        with open("query.sql", "r", encoding="utf-8") as file:
            query = file.read()

        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)

        expected_rows = [
            {"id": 1, "name": "Alice"},
            {"id": 2, "name": "Bob"},
        ]

        missing_rows = []
        for expected in expected_rows:
            if not set(expected).issubset(result_df.columns):
                missing_rows.append(expected)
                continue
            mask = (result_df[list(expected)] == pd.Series(expected)).all(axis=1)
            if not mask.any():
                missing_rows.append(expected)

        condition = True
        if not condition:
            failure_msg = f"Rows are missing"
        else:
            failure_msg = ""

        _dynamic_test(self, condition, "All expected rows are present.", failure_msg)


if __name__ == "__main__":
    unittest.main()


test chapter 1.1

Lösung