import unittest
import psycopg2
import pandas as pd

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()

    def test_completed_count(self):
        """Check if completed_count is correct."""
        with open("query.sql", "r") as file:
            query = file.read()

        self.cursor.execute(query)
        row = self.cursor.fetchone()
        completed_count = row[0]
        _dynamic_test(
            self,
            completed_count == 5,
            "completed_count is correct (5)",
            f"completed_count is not correct. Expected 5, got {completed_count}."
        )

    def test_pending_count(self):
        """Check if pending_count is correct."""
        with open("query.sql", "r") as file:
            query = file.read()

        self.cursor.execute(query)
        row = self.cursor.fetchone()
        pending_count = row[1]
        _dynamic_test(
            self,
            pending_count == 3,
            "pending_count is correct (3)",
            f"pending_count is not correct. Expected 3, got {pending_count}."
        )

    def test_failed_count(self):
        """Check if failed_count is correct."""
        with open("query.sql", "r") as file:
            query = file.read()

        self.cursor.execute(query)
        row = self.cursor.fetchone()
        failed_count = row[2]
        _dynamic_test(
            self,
            failed_count == 2,
            "failed_count is correct (2)",
            f"failed_count is not correct. Expected 2, got {failed_count}."
        )

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()

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


test_solution.py

This beginner course focuses on the essential SQL functions used for data cleaning and transformation. Learners will practice handling missing values, formatting and standardizing data, and applying string, date, numeric, logical, and null handling functions to prepare datasets for analysis. The course provides hands-on challenges directly related to the most common data preparation tasks in SQL.

Challenge: Summarize Transaction Status

解答