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_orders_included(self):
        """Check that all orders are present in the query result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT COUNT(*) FROM orders;")
        order_count = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == order_count,
            "All orders are included in the result.",
            f"Expected {order_count} orders, but got {len(rows)}."
        )

    def test_columns_present(self):
        """Check that required columns are present in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required = {"order_id", "order_date", "amount", "customer_name"}
        _dynamic_test(
            self,
            required.issubset(set(columns)),
            "All required columns are present.",
            f"Missing columns: {required - set(columns)}."
        )

    def test_customer_name_matches(self):
        """Check that customer name is correct or NULL for each order."""
        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)
        # Get correct mapping from orders to customer name
        self.cursor.execute("""
            SELECT o.order_id, c.name AS customer_name
            FROM orders o
            LEFT JOIN customers c ON o.customer_id = c.customer_id
        """)
        expected = dict(self.cursor.fetchall())
        mismatches = []
        for _, row in result_df.iterrows():
            oid = row['order_id']
            expected_name = expected[oid]
            actual_name = row['customer_name']
            if pd.isnull(expected_name):
                if not pd.isnull(actual_name):
                    mismatches.append((oid, actual_name, None))
            else:
                if actual_name != expected_name:
                    mismatches.append((oid, actual_name, expected_name))
        _dynamic_test(
            self,
            len(mismatches) == 0,
            "Customer names are correct or NULL where appropriate.",
            f"Mismatches found: {mismatches}"
        )

    def test_null_for_missing_customer(self):
        """Check that customer_name is NULL for orders with no matching customer."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Remove foreign key constraint to allow inserting an order with a non-existent customer
        self.cursor.execute("ALTER TABLE orders DROP CONSTRAINT IF EXISTS orders_customer_id_fkey;")
        self.connection.commit()
        self.cursor.execute("INSERT INTO orders (order_id, customer_id, order_date, amount) VALUES (99, 999, '2024-06-10', 100.00);")
        self.connection.commit()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        test_row = result_df[result_df['order_id'] == 99]
        cond = False
        if not test_row.empty:
            cond = pd.isnull(test_row.iloc[0]['customer_name'])
        _dynamic_test(
            self,
            cond,
            "Orders with missing customer have NULL for customer_name.",
            "Order with missing customer should have NULL for customer_name."
        )
        # Clean up
        self.cursor.execute("DELETE FROM orders WHERE order_id = 99;")
        self.connection.commit()
        # Restore the foreign key constraint if needed (no-op if not needed)
        #self.cursor.execute("ALTER TABLE orders ADD CONSTRAINT orders_customer_id_fkey FOREIGN KEY (customer_id) REFERENCES customers(customer_id);")
        #self.connection.commit()

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

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: Orders Without Customers

解答