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_column_names(self):
        """Check that all required columns are present in the query result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected_columns = ["sale_id", "sale_date", "employee_id", "region", "amount", "cumulative_sales"]
        missing_columns = [col for col in expected_columns if col not in columns]
        condition = len(missing_columns) == 0
        _dynamic_test(
            self,
            condition,
            "All required columns are present in the result.",
            f"Missing columns: {', '.join(missing_columns)}"
        )

    def test_cumulative_sales_correctness(self):
        """Check that cumulative_sales is calculated correctly per region and ordering."""
        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)
        # Prepare expected cumulative sum per region, ordered by sale_date, sale_id
        self.cursor.execute("SELECT sale_id, sale_date, employee_id, region, amount FROM sales")
        all_rows = self.cursor.fetchall()
        all_columns = [desc[0] for desc in self.cursor.description]
        sales_df = pd.DataFrame(all_rows, columns=all_columns)
        # Ensure 'amount' is numeric
        sales_df['amount'] = pd.to_numeric(sales_df['amount'])
        sales_df = sales_df.sort_values(["region", "sale_date", "sale_id"]).reset_index(drop=True)
        sales_df["cumulative_sales_expected"] = sales_df.groupby("region")['amount'].cumsum()
        # Compare result rows to expected
        merged = pd.merge(result_df, sales_df, on=["sale_id", "sale_date", "employee_id", "region", "amount"], how="left")
        # Check cumulative_sales matches
        condition = True
        mismatches = []
        for idx, row in merged.iterrows():
            user_val = float(row["cumulative_sales"])
            expected_val = float(row["cumulative_sales_expected"])
            if abs(user_val - expected_val) > 0.01:
                mismatches.append({
                    "sale_id": row["sale_id"],
                    "region": row["region"],
                    "expected": expected_val,
                    "user": user_val
                })
                condition = False
        if not condition:
            msg = "Cumulative sales mismatches:\\n" + "\\n".join(
                f"sale_id {m['sale_id']} (region {m['region']}): expected {m['expected']}, got {m['user']}" for m in mismatches)
        else:
            msg = ""
        _dynamic_test(
            self,
            condition,
            "Cumulative sales values are correct for each region.",
            msg
        )

    def test_ordering(self):
        """Check that results are ordered by region, sale_date, sale_id."""
        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)
        sorted_df = result_df.sort_values(["region", "sale_date", "sale_id"]).reset_index(drop=True)
        # Compare row by row
        condition = result_df.reset_index(drop=True).equals(sorted_df)
        _dynamic_test(
            self,
            condition,
            "Result is ordered by region, sale_date, and sale_id.",
            "Result is not ordered by region, sale_date, and sale_id."
        )

    def test_cumulative_sales_resets_per_region(self):
        """Check that cumulative_sales restarts from the first row for each region."""
        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)
        # Ensure 'amount' is numeric
        result_df['amount'] = pd.to_numeric(result_df['amount'])
        result_df['cumulative_sales'] = pd.to_numeric(result_df['cumulative_sales'])
        # For each region, first row cumulative_sales == amount
        group = result_df.groupby("region")
        resets = []
        for region, g in group:
            g_sorted = g.sort_values(["sale_date", "sale_id"]).reset_index(drop=True)
            if abs(float(g_sorted.iloc[0]["cumulative_sales"]) - float(g_sorted.iloc[0]["amount"])) > 0.01:
                resets.append(region)
        condition = len(resets) == 0
        _dynamic_test(
            self,
            condition,
            "Cumulative sales resets correctly for each region.",
            f"Cumulative sales does not reset for regions: {', '.join(resets)}"
        )

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):
    import re
    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 analyzing data with SQL window functions. This course guides you through the core concepts, syntax, and practical applications of window functions, empowering you to perform advanced analytics and reporting directly in SQL.

Explore the foundational concepts of window functions, their syntax, and how they differ from traditional aggregate functions. Learn to apply basic window functions for analytics.

Dive into ranking, percentiles, and analytic functions. Learn to use ROW_NUMBER, RANK, DENSE_RANK, and NTILE for advanced reporting.

Apply window functions to solve complex business problems, including advanced framing, multi-function queries, and real-world analytics.

Challenge: Cumulative Sales by Region

Рішення

Challenge: Cumulative Sales by Region

Рішення