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_all_sales_records_present(self):
        """Check that all sales records are present and paired with correct customer names."""
        with open("query.sql", "r") as file:
            query = file.read()
        solution_query = '''
        SELECT customers.first_name, customers.last_name, sales.quantity
        FROM customers JOIN sales ON customers.customer_id = sales.customer_id
        '''
        self.cursor.execute(solution_query)
        solution_rows = self.cursor.fetchall()
        solution_columns = [desc[0] for desc in self.cursor.description]
        solution_df = pd.DataFrame(solution_rows, columns=solution_columns)

        self.cursor.execute(query)
        user_rows = self.cursor.fetchall()
        user_columns = [desc[0] for desc in self.cursor.description]
        user_df = pd.DataFrame(user_rows, columns=user_columns)

        missing_rows = []
        for _, expected_row in solution_df.iterrows():
            mask = (
                (user_df['first_name'] == expected_row['first_name']) &
                (user_df['last_name'] == expected_row['last_name']) &
                (user_df['quantity'] == expected_row['quantity'])
            )
            if not mask.any():
                missing_rows.append(expected_row.to_dict())
        condition = len(missing_rows) == 0
        if not condition:
            missing_str = '\\n'.join(str(row) for row in missing_rows)
            failure_msg = f"The following expected rows are missing from your result:\\n{missing_str}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "All sales records are present and paired with correct customer names.",
            failure_msg
        )

    def test_no_extra_rows(self):
        """Check that only sales with a matching customer record are included."""
        with open("query.sql", "r") as file:
            query = file.read()
        solution_query = '''
        SELECT customers.first_name, customers.last_name, sales.quantity
        FROM customers JOIN sales ON customers.customer_id = sales.customer_id
        '''
        self.cursor.execute(solution_query)
        solution_rows = self.cursor.fetchall()
        solution_columns = [desc[0] for desc in self.cursor.description]
        solution_df = pd.DataFrame(solution_rows, columns=solution_columns)

        self.cursor.execute(query)
        user_rows = self.cursor.fetchall()
        user_columns = [desc[0] for desc in self.cursor.description]
        user_df = pd.DataFrame(user_rows, columns=user_columns)

        extra_rows = []
        for _, user_row in user_df.iterrows():
            mask = (
                (solution_df['first_name'] == user_row['first_name']) &
                (solution_df['last_name'] == user_row['last_name']) &
                (solution_df['quantity'] == user_row['quantity'])
            )
            if not mask.any():
                extra_rows.append(user_row.to_dict())
        condition = len(extra_rows) == 0
        if not condition:
            extra_str = '\\n'.join(str(row) for row in extra_rows)
            failure_msg = f"Your result contains extra/unexpected rows:\\n{extra_str}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "No extra/unexpected rows are present in the result.",
            failure_msg
        )

    def test_quantity_matches_sales_table(self):
        """Check that the quantity value for each sale matches the sales table."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        user_rows = self.cursor.fetchall()
        user_columns = [desc[0] for desc in self.cursor.description]
        user_df = pd.DataFrame(user_rows, columns=user_columns)

        self.cursor.execute("SELECT sale_id, customer_id, quantity FROM sales")
        sales_rows = self.cursor.fetchall()
        sales_df = pd.DataFrame(sales_rows, columns=["sale_id", "customer_id", "quantity"])
        self.cursor.execute("SELECT customer_id, first_name, last_name FROM customers")
        cust_rows = self.cursor.fetchall()
        cust_df = pd.DataFrame(cust_rows, columns=["customer_id", "first_name", "last_name"])
        merged = pd.merge(sales_df, cust_df, how="inner", on="customer_id")

        mismatch = []
        for _, row in merged.iterrows():
            mask = (
                (user_df['first_name'] == row['first_name']) &
                (user_df['last_name'] == row['last_name']) &
                (user_df['quantity'] == row['quantity'])
            )
            if not mask.any():
                mismatch.append({
                    "first_name": row['first_name'],
                    "last_name": row['last_name'],
                    "quantity": row['quantity']
                })
        condition = len(mismatch) == 0
        if not condition:
            mismatch_str = '\\n'.join(str(row) for row in mismatch)
            failure_msg = f"The following (name, quantity) pairs are missing or incorrect:\\n{mismatch_str}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Quantity value for each sale matches the sales table.",
            failure_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

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: Customer Sales Report

Рішення

Challenge: Customer Sales Report

Рішення