import unittest
import psycopg2
import pandas as pd
import re

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_single_row_three_columns(self):
        """Check that result has one row and three columns."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        condition = len(rows) == 1 and len(columns) == 3
        _dynamic_test(
            self,
            condition,
            "Query returns a single row with three columns.",
            f"Expected 1 row and 3 columns, got {len(rows)} rows and {len(columns)} columns."
        )

    def test_column_names(self):
        """Check that columns are named min_price, max_price, avg_price."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [normalize_text(desc[0]) for desc in self.cursor.description]
        expected = ["min_price", "max_price", "avg_price"]
        missing = [col for col in expected if col not in columns]
        condition = len(missing) == 0
        _dynamic_test(
            self,
            condition,
            "Returned columns are named min_price, max_price, and avg_price.",
            f"Missing columns: {', '.join(missing)}."
        )

    def test_min_price_value(self):
        """Check that min_price matches the lowest price in products."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        row = self.cursor.fetchone()
        col_idx = [desc[0] for desc in self.cursor.description].index("min_price")
        min_price = row[col_idx]
        self.cursor.execute("SELECT MIN(price) FROM products;")
        expected = self.cursor.fetchone()[0]
        condition = float(min_price) == float(expected)
        _dynamic_test(
            self,
            condition,
            "min_price value is correct.",
            f"min_price should be {expected}, got {min_price}."
        )

    def test_max_price_value(self):
        """Check that max_price matches the highest price in products."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        row = self.cursor.fetchone()
        col_idx = [desc[0] for desc in self.cursor.description].index("max_price")
        max_price = row[col_idx]
        self.cursor.execute("SELECT MAX(price) FROM products;")
        expected = self.cursor.fetchone()[0]
        condition = float(max_price) == float(expected)
        _dynamic_test(
            self,
            condition,
            "max_price value is correct.",
            f"max_price should be {expected}, got {max_price}."
        )

    def test_avg_price_value(self):
        """Check that avg_price matches the average price in products."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        row = self.cursor.fetchone()
        col_idx = [desc[0] for desc in self.cursor.description].index("avg_price")
        avg_price = row[col_idx]
        self.cursor.execute("SELECT AVG(price) FROM products;")
        expected = self.cursor.fetchone()[0]
        # Compare rounded to 2 decimals for floating point tolerance
        condition = round(float(avg_price), 2) == round(float(expected), 2)
        _dynamic_test(
            self,
            condition,
            "avg_price value is correct (rounded to 2 decimals).",
            f"avg_price should be {expected}, got {avg_price}."
        )

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: Analyze Price Distribution

解答