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_returns_each_product_id(self):
        """Check if query returns each product_id present in sales table."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT DISTINCT product_id FROM sales")
        expected_ids = set([row[0] for row in self.cursor.fetchall()])
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_ids = set([row[0] for row in rows])
        _dynamic_test(
            self,
            expected_ids == result_ids,
            "Query returns all product_id values from sales table.",
            f"Query does not return all product_id values. Missing: {expected_ids - result_ids}"
        )

    def test_includes_total_quantity_column(self):
        """Check if query includes total_quantity column with correct sums."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT product_id, SUM(quantity) AS total_quantity FROM sales GROUP BY product_id")
        expected = dict(self.cursor.fetchall())
        self.cursor.execute(query)
        result = dict(self.cursor.fetchall())
        condition = expected == result
        _dynamic_test(
            self,
            condition,
            "Query includes correct total quantity for each product_id.",
            f"Total quantity values are incorrect. Expected: {expected}, Got: {result}"
        )

    def test_ordered_by_total_quantity_desc(self):
        """Check if results are ordered by total_quantity descending."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        quantities = [row[1] for row in rows]
        condition = quantities == sorted(quantities, reverse=True)
        _dynamic_test(
            self,
            condition,
            "Results are ordered by total_quantity descending.",
            "Results are not ordered by total_quantity in descending order."
        )

    def test_uses_group_by_product_id(self):
        """Check if query uses GROUP BY product_id."""
        with open("query.sql", "r") as file:
            query = file.read()
        uses_group_by = "group by" in normalize_text(query) and "product_id" in normalize_text(query)
        _dynamic_test(
            self,
            uses_group_by,
            "Query uses GROUP BY product_id.",
            "Query does not use GROUP BY product_id."
        )

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: Bestselling Products

解答