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_all_rows_selected(self):
        """Check if query selects all rows from products table."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT COUNT(*) FROM products;")
        expected_count = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == expected_count,
            "Query selects all rows from products table.",
            f"Query does not select all rows. Expected {expected_count}, got {len(rows)}."
        )

    def test_column_names(self):
        """Check if output includes the required columns."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        actual_columns = [desc[0] for desc in self.cursor.description]
        required_columns = {"product_id", "product_name"}
        # Look for a column that is uppercase and one that is substring
        has_upper = any("upper" in col or "upper" in normalize_text(col) for col in actual_columns)
        has_first3 = any("first3" in col or "first3" in normalize_text(col) for col in actual_columns)
        condition = required_columns.issubset(set(actual_columns)) and has_upper and has_first3
        _dynamic_test(
            self,
            condition,
            "Output includes all required columns.",
            "Output does not include all required columns: product_id, product_name, uppercase and first 3 letters columns."
        )

    def test_uppercase_column(self):
        """Check if the uppercase column displays product names in all capital letters."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        # Find the uppercase column
        upper_col_idx = None
        for idx, col in enumerate(columns):
            if "upper" in col or "upper" in normalize_text(col):
                upper_col_idx = idx
                break
        self.cursor.execute("SELECT product_name FROM products;")
        product_names = [row[0] for row in self.cursor.fetchall()]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        # Find matching uppercase values
        condition = True
        for i, row in enumerate(rows):
            if upper_col_idx is not None:
                expected = product_names[i].upper()
                actual = row[upper_col_idx]
                if actual != expected:
                    condition = False
                    break
        _dynamic_test(
            self,
            condition,
            "Uppercase column displays product names in all capital letters.",
            "Uppercase column does not display all product names in uppercase."
        )

    def test_first3_column(self):
        """Check if the first 3 letters column displays only the first three characters of each product name."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        # Find the first3 column
        first3_col_idx = None
        for idx, col in enumerate(columns):
            if "first3" in col or "first3" in normalize_text(col):
                first3_col_idx = idx
                break
        self.cursor.execute("SELECT product_name FROM products;")
        product_names = [row[0] for row in self.cursor.fetchall()]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        condition = True
        for i, row in enumerate(rows):
            if first3_col_idx is not None:
                expected = product_names[i][:3]
                actual = row[first3_col_idx]
                if actual != expected:
                    condition = False
                    break
        _dynamic_test(
            self,
            condition,
            "First 3 letters column displays correct substring for all products.",
            "First 3 letters column does not display correct substring for all products."
        )


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

A comprehensive beginner-friendly course designed to teach SQL from the ground up, focusing on practical data analysis skills. Through engaging explanations, real-world examples, and hands-on challenges, learners will master querying, transforming, and analyzing data using SQL.

Learn the foundational concepts of relational databases and SQL. Explore how to retrieve, filter, and sort data from tables, setting the stage for deeper data analysis.

Dive into SQL's powerful aggregation features. Learn to summarize, group, and analyze data for deeper insights.

Unlock the power of combining data from multiple tables. Learn to perform different types of joins to answer complex business questions.

Master advanced SQL techniques for powerful data analysis, including subqueries, window functions, and data transformation.

Challenge: Product Name Formatting

解答