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_customers_included(self):
        """Check that all customers are included in the result."""
        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)
        self.cursor.execute("SELECT COUNT(*) FROM customers;")
        total_customers = self.cursor.fetchone()[0]
        customers_in_result = set(result_df['customer_id'].tolist())
        self.cursor.execute("SELECT customer_id FROM customers;")
        all_customers = set([r[0] for r in self.cursor.fetchall()])
        condition = all_customers.issubset(customers_in_result)
        _dynamic_test(
            self,
            condition,
            "All customers are included in the result.",
            "Some customers are missing from the result."
        )

    def test_customers_with_no_orders_show_null_amount(self):
        """Check that customers with no orders have NULL in amount column."""
        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)
        no_order_customers = [3, 5]
        filtered = result_df[result_df['customer_id'].isin(no_order_customers)]
        condition = filtered['amount'].isnull().all() and not filtered.empty
        _dynamic_test(
            self,
            condition,
            "Customers with no orders show NULL in amount column.",
            "Customers with no orders do not have NULL in amount column or are missing."
        )

    def test_multiple_orders_per_customer(self):
        """Check that each order of a customer appears as a separate row with correct amount."""
        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)
        alice_orders = result_df[result_df['customer_id'] == 1]
        expected_amounts = sorted([45.98, 0.00])
        actual_amounts = sorted([float(a) for a in alice_orders['amount'].dropna().tolist()])
        amounts_match = len(actual_amounts) == 2 and all(abs(a-b)<0.001 for a,b in zip(expected_amounts, actual_amounts))
        condition = amounts_match
        _dynamic_test(
            self,
            condition,
            "Multiple orders per customer are shown as separate rows with correct amounts.",
            "Multiple orders per customer are not shown correctly."
        )

    def test_result_sorted(self):
        """Check that the results are sorted by customer_id and then by order_id (NULL order_id last)."""
        with open("query.sql", "r") as file:
            query = file.read()
        # For the solution, order_id is not selected, so we need to reconstruct the sort order using available data.
        # We'll join with orders to get order_id for each row.
        # Get the expected result with order_id
        self.cursor.execute("""
            SELECT c.customer_id, c.name, o.amount, o.order_id
            FROM customers c
            LEFT JOIN orders o ON c.customer_id = o.customer_id
            ORDER BY c.customer_id, o.order_id
        """)
        expected_rows = self.cursor.fetchall()
        expected_df = pd.DataFrame(expected_rows, columns=["customer_id", "name", "amount", "order_id"])
        # Now get the user's result
        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)
        # Add order_id for matching (merge)
        merged = pd.merge(user_df, expected_df, on=["customer_id", "name", "amount"], how="left")
        # Now check if the user rows are in the same order as expected (by customer_id, order_id where order_id may be null)
        merged['order_id_for_sort'] = merged['order_id'].apply(lambda x: x if pd.notnull(x) else float('inf'))
        merged_expected = merged.sort_values(["customer_id", "order_id_for_sort"], kind="mergesort").reset_index(drop=True)
        # Compare the original user row order to the sorted order
        correct_order = list(merged.index) == list(merged_expected.index)
        _dynamic_test(
            self,
            correct_order,
            "Results are sorted by customer_id and then by order_id.",
            "Results are not sorted by customer_id and order_id."
        )

    def test_column_names(self):
        """Check that the columns are named customer_id, name, and amount."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected = set(["customer_id", "name", "amount"])
        condition = set(columns) == expected
        _dynamic_test(
            self,
            condition,
            "Column names are correct.",
            f"Column names are incorrect: found {columns}."
        )

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):
    import re
    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

Master the art of SQL joins with engaging, real-world examples and hands-on challenges. This course takes you from foundational join concepts to advanced multi-table queries, ensuring you can confidently analyze and combine data from multiple sources.

Explore the essential SQL join types—INNER, LEFT, RIGHT, and FULL OUTER JOIN—through practical scenarios and hands-on challenges.

Level up your SQL skills by mastering multi-table joins, self-joins, and advanced join conditions for complex data analysis.

Master the nuances of NULL handling in joins, optimize join performance, and tackle real-world data analysis challenges.

Challenge: List All Customers and Their Orders

Рішення

Challenge: List All Customers and Their Orders

Рішення