import unittest
import psycopg2
import pandas as pd
import re

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 fix_query_cast_date(self, query):
        # Ensure both base and recursive terms cast as DATE
        # Replace 'report_date + INTERVAL ...' with 'CAST(report_date + INTERVAL ... AS DATE)'
        query = re.sub(r"(report_date \+ INTERVAL '[^']+')", r"CAST(\1 AS DATE)", query)
        return query

    def test_all_dates_inclusive(self):
        """Check that all dates between 2024-07-01 and 2024-07-05 are present, inclusive."""
        with open("query.sql", "r") as file:
            query = file.read()
        query = self.fix_query_cast_date(query)
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_dates = sorted([str(r[0]) for r in rows])
        expected_dates = [
            '2024-07-01',
            '2024-07-02',
            '2024-07-03',
            '2024-07-04',
            '2024-07-05'
        ]
        condition = result_dates == expected_dates
        _dynamic_test(
            self,
            condition,
            "All dates between 2024-07-01 and 2024-07-05 are present.",
            f"Expected dates: {expected_dates}, but got: {result_dates}"
        )

    def test_no_dates_skipped(self):
        """Check that no dates are skipped in the generated range."""
        with open("query.sql", "r") as file:
            query = file.read()
        query = self.fix_query_cast_date(query)
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_dates = sorted([str(r[0]) for r in rows])
        for i in range(len(result_dates) - 1):
            prev = pd.to_datetime(result_dates[i])
            curr = pd.to_datetime(result_dates[i+1])
            if (curr - prev).days != 1:
                condition = False
                break
        else:
            condition = True
        _dynamic_test(
            self,
            condition,
            "No dates are skipped in the generated date range.",
            "There is a gap between two consecutive dates in the result."
        )

    def test_end_date_included(self):
        """Check that the end date is included in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        query = self.fix_query_cast_date(query)
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_dates = [str(r[0]) for r in rows]
        condition = '2024-07-05' in result_dates
        _dynamic_test(
            self,
            condition,
            "End date 2024-07-05 is included in the result.",
            "End date 2024-07-05 is missing from the result."
        )

    def test_generic_date_range(self):
        """Check that the query works for a different valid date range (2024-08-10 to 2024-08-13)."""
        with open("query.sql", "r") as file:
            query = file.read()
        query = re.sub(r"DATE '2024-07-01'", "DATE '2024-08-10'", query)
        query = re.sub(r"DATE '2024-07-05'", "DATE '2024-08-13'", query)
        query = self.fix_query_cast_date(query)
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_dates = sorted([str(r[0]) for r in rows])
        expected_dates = [
            '2024-08-10',
            '2024-08-11',
            '2024-08-12',
            '2024-08-13'
        ]
        condition = result_dates == expected_dates
        _dynamic_test(
            self,
            condition,
            "Query works for a different valid date range.",
            f"Expected: {expected_dates}, Got: {result_dates}"
        )

    def test_single_date(self):
        """Check that the query works for a single date range (start = end)."""
        with open("query.sql", "r") as file:
            query = file.read()
        query = re.sub(r"DATE '2024-07-01'", "DATE '2024-09-15'", query)
        query = re.sub(r"DATE '2024-07-05'", "DATE '2024-09-15'", query)
        query = self.fix_query_cast_date(query)
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_dates = [str(r[0]) for r in rows]
        expected_dates = ['2024-09-15']
        condition = result_dates == expected_dates
        _dynamic_test(
            self,
            condition,
            "Query works for single date where start = end.",
            f"Expected: {expected_dates}, Got: {result_dates}"
        )


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

A comprehensive course designed to help you master Common Table Expressions (CTEs) in SQL, from foundational concepts to advanced real-world applications. Explore recursive queries, modular query design, and performance considerations through engaging theory and practical, scenario-based tasks.

Explore the basics of Common Table Expressions, their syntax, and how they can simplify SQL queries. Practice using CTEs in real-world scenarios such as reporting and data transformation.

Delve into recursive CTEs for handling hierarchical data, such as organizational charts and bill of materials. Practice building recursive queries to solve real-world problems.

Master advanced CTE techniques, including combining CTEs with aggregates, optimizing performance, and using CTEs for complex reporting.

Challenge: Generating Date Ranges

解答