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_rows_returned(self):
        """Check if query returns all rows from orders table."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT COUNT(*) FROM orders;")
        expected_count = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == expected_count,
            f"Returned all {expected_count} rows from orders table.",
            f"Returned {len(rows)} rows, expected {expected_count}."
        )

    def test_required_columns_present(self):
        """Check if output contains required columns."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required_columns = {"order_id", "customer_id", "order_date", "predicted_delivery_date"}
        missing = required_columns - set(columns)
        _dynamic_test(
            self,
            not missing,
            "All required columns are present.",
            f"Missing columns: {', '.join(missing)}."
        )

    def test_predicted_delivery_date_correct(self):
        """Check if predicted_delivery_date is exactly 5 days after order_date."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        incorrect = []
        for _, row in df.iterrows():
            order_date = pd.to_datetime(row["order_date"]).date()
            predicted = pd.to_datetime(row["predicted_delivery_date"]).date()
            if predicted != order_date + pd.Timedelta(days=5):
                incorrect.append({
                    "order_id": row["order_id"],
                    "order_date": str(order_date),
                    "predicted_delivery_date": str(predicted)
                })
        _dynamic_test(
            self,
            not incorrect,
            "All predicted_delivery_date values are exactly 5 days after order_date.",
            f"Incorrect prediction for: {incorrect}"
        )

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: Predict Delivery Dates

解答