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_row_count_matches_sales(self):
        """Check that the query returns one row per sale."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT COUNT(*) FROM sales;")
        sales_count = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        result = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(result) == sales_count,
            "Returned row count matches sales table.",
            f"Expected {sales_count} rows, got {len(result)}."
        )

    def test_customer_name_combined_correctly(self):
        """Check that the customer_name column is correctly combined."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        result = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(result, columns=columns)
        self.cursor.execute("SELECT s.sale_id, c.first_name, c.last_name FROM sales s JOIN customers c ON s.customer_id = c.customer_id;")
        expected = self.cursor.fetchall()
        correct = True
        for sale_id, first, last in expected:
            customer_name = f"{first} {last}"
            # Find row in df with this customer_name
            if not (df['customer_name'] == customer_name).any():
                correct = False
                break
        _dynamic_test(
            self,
            correct,
            "Customer names are correctly combined.",
            "Some customer names are not combined correctly."
        )

    def test_product_name_matches(self):
        """Check that product_name matches the product for each sale."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        result = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(result, columns=columns)
        self.cursor.execute("SELECT s.sale_id, p.product_name FROM sales s JOIN products p ON s.product_id = p.product_id;")
        expected = self.cursor.fetchall()
        correct = True
        for sale_id, product_name in expected:
            if not (df['product_name'] == product_name).any():
                correct = False
                break
        _dynamic_test(
            self,
            correct,
            "Product names match for each sale.",
            "Some product names do not match the sale."
        )

    def test_quantity_matches_sales(self):
        """Check that quantity matches the value in the sales table for each sale."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        result = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(result, columns=columns)
        self.cursor.execute("SELECT c.first_name || ' ' || c.last_name AS customer_name, p.product_name, s.quantity FROM sales s JOIN customers c ON s.customer_id = c.customer_id JOIN products p ON s.product_id = p.product_id;")
        expected = self.cursor.fetchall()
        expected_df = pd.DataFrame(expected, columns=['customer_name', 'product_name', 'quantity'])
        # For each expected row, check if there is a matching row in df
        mismatch = False
        for _, row in expected_df.iterrows():
            mask = (
                (df['customer_name'] == row['customer_name']) &
                (df['product_name'] == row['product_name']) &
                (df['quantity'] == row['quantity'])
            )
            if not mask.any():
                mismatch = True
                break
        _dynamic_test(
            self,
            not mismatch,
            "Quantities match for each sale.",
            "Some quantities do not match the sales table."
        )

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 Purchase Details

Рішення