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

Master the use of built-in SQL functions to manipulate strings, dates, numbers, logical values, nulls, and JSON data. This course provides a hands-on approach to understanding and applying SQL functions in real-world scenarios, enhancing your data querying and transformation skills.

Explore SQL's powerful string functions for manipulating and analyzing textual data in databases.

Master SQL's date and time functions to work with temporal data effectively.

Learn to use SQL's numeric functions for calculations and data analysis.

Harness SQL's logical functions to control query flow and make decisions based on data.

Master handling of NULL values in SQL using specialized functions.

Learn to work with JSON data in SQL using specialized functions for querying and manipulation.

Challenge: Find Late Deliveries

Lösung

Challenge: Find Late Deliveries

Lösung