The company needs to find the **minimum price** of a product from each category in order to organize a marketing campaign. They have assigned you this task!

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_column_aliases(self):
        """Check if column aliases category_name and min_price are used."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        _dynamic_test(
            self,
            "category_name" in columns and "min_price" in columns,
            "Both column aliases 'category_name' and 'min_price' are present.",
            "Column aliases 'category_name' and/or 'min_price' are missing."
        )

    def test_distinct_product_count(self):
        """Check that only categories with more than 5 distinct products appear."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)

        # Count distinct products per category from raw data
        self.cursor.execute('''
            SELECT category.name, COUNT(DISTINCT product.id) AS product_count
            FROM category
            JOIN product ON category.id = product.category_id
            GROUP BY category.name
        ''')
        prod_counts = dict(self.cursor.fetchall())
        result_categories = set(result_df['category_name'])
        failed = [cat for cat in result_categories if prod_counts.get(cat, 0) <= 5]
        _dynamic_test(
            self,
            len(failed) == 0,
            "All returned categories have more than 5 distinct products.",
            f"The following categories do not have more than 5 distinct products: {failed}"
        )

    def test_expected_result(self):
        """Check that the result matches the expected output."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)
        expected = pd.DataFrame([
            {"category_name": "Bakery", "min_price": "30.00"},
            {"category_name": "Beverages", "min_price": "20.00"},
            {"category_name": "Dairy", "min_price": "50.00"},
            {"category_name": "Frozen Foods", "min_price": "100.00"},
            {"category_name": "Fruits", "min_price": "70.00"},
            {"category_name": "Meat", "min_price": "350.00"},
            {"category_name": "Snacks", "min_price": "30.00"},
            {"category_name": "Vegetables", "min_price": "60.00"}
        ])
        # Compare both dataframes by sorting by category_name
        result_sorted = result_df.sort_values('category_name').reset_index(drop=True)
        expected_sorted = expected.sort_values('category_name').reset_index(drop=True)
        # Coerce min_price to string for comparison
        result_sorted['min_price'] = result_sorted['min_price'].apply(lambda x: f"{float(x):.2f}")
        condition = result_sorted.equals(expected_sorted)
        fail_msg = f"Expected result:\\n{expected_sorted}\\nBut got:\\n{result_sorted}" if not condition else ""
        _dynamic_test(
            self,
            condition,
            "Query returns the correct minimum price for each category.",
            fail_msg
        )

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 course is perfect for those who already have a basic understanding of SQL and want to delve into more advanced concepts to craft more powerful queries. Throughout the course, you will become familiar with data grouping and filtering grouped data. You will also learn how to work with multiple tables simultaneously, including how to combine them. Additionally, you will explore different types of table joins and how to apply them in practice.

Challenge: Get Minimum Price by Category

Challenge: Get Minimum Price by Category

Brief Instructions

解答