import unittest
import psycopg2
import pandas as pd
from datetime import timedelta

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_only_late_orders(self):
        """Check that only orders shipped more than 3 days after order_date are returned."""
        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)
        if not result_df.empty:
            # Check that all returned rows have shipped_at > order_date + 3 days
            result_df['order_date'] = pd.to_datetime(result_df['order_date'])
            result_df['shipped_at'] = pd.to_datetime(result_df['shipped_at'])
            late_mask = result_df['shipped_at'] > (result_df['order_date'] + pd.Timedelta(days=3))
            condition = late_mask.all()
            failure_msg = "Some returned rows do not have shipped_at more than 3 days after order_date."
        else:
            # Acceptable if no rows, as no late orders in base data
            condition = True
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Only orders shipped more than 3 days after order_date are returned.",
            failure_msg or "Some rows were returned, but no orders should be late based on the data."
        )

    def test_no_early_orders_in_result(self):
        """Check that no orders shipped within 3 days 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)
        if not result_df.empty:
            result_df['order_date'] = pd.to_datetime(result_df['order_date'])
            result_df['shipped_at'] = pd.to_datetime(result_df['shipped_at'])
            # Any row where shipped_at <= order_date + 3 days is an error
            early_mask = result_df['shipped_at'] <= (result_df['order_date'] + pd.Timedelta(days=3))
            condition = (~early_mask).all()
            failure_msg = "Early orders (not late) were included in the result."
        else:
            condition = True
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "No early orders are included in the result.",
            failure_msg
        )

    def test_query_works_with_late_data(self):
        """Check that query correctly returns late orders if present."""
        # Insert a late order
        self.cursor.execute("INSERT INTO orders (order_id, customer_id, order_date, delivery_time, shipped_at) VALUES (11, 111, '2024-05-01', '10:00:00', '2024-05-05 12:00:00')")
        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)
        # There should be one row, order_id 11
        condition = (result_df.shape[0] >= 1) and (11 in result_df['order_id'].values)
        _dynamic_test(
            self,
            condition,
            "Late order is returned correctly.",
            "Late order was not returned as expected."
        )

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

This beginner course focuses on the essential SQL functions used for data cleaning and transformation. Learners will practice handling missing values, formatting and standardizing data, and applying string, date, numeric, logical, and null handling functions to prepare datasets for analysis. The course provides hands-on challenges directly related to the most common data preparation tasks in SQL.

Challenge: Find Late Deliveries

解答