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_one_row_per_region(self):
        """Check that the result contains one row per unique region."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT DISTINCT region FROM sales")
        expected_regions = set(row[0] for row in self.cursor.fetchall())
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_regions = set(row[0] for row in rows)
        _dynamic_test(
            self,
            result_regions == expected_regions,
            "Result contains one row per unique region.",
            f"Result regions {result_regions} do not match expected regions {expected_regions}."
        )

    def test_highest_sale_amount_per_region(self):
        """Check that each row shows the highest sale amount for its region."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        self.cursor.execute("SELECT region, MAX(amount) FROM sales GROUP BY region")
        max_amounts = dict(self.cursor.fetchall())
        failed = []
        for row in rows:
            region, employee_id, amount = row
            if float(amount) != float(max_amounts[region]):
                failed.append((region, amount, max_amounts[region]))
        condition = len(failed) == 0
        if not condition:
            msg = "\\n".join([f"{r}: got {a}, expected {e}" for r, a, e in failed])
            failure_msg = f"Not all rows show the highest sale amount per region:\\n{msg}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Each row shows the highest sale amount for its region.",
            failure_msg
        )

    def test_only_one_row_per_region_even_on_tie(self):
        """Check that only one row per region is returned, even if there are ties."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        region_counts = {}
        for row in rows:
            region = row[0]
            region_counts[region] = region_counts.get(region, 0) + 1
        failed = [r for r, c in region_counts.items() if c != 1]
        condition = len(failed) == 0
        if not condition:
            failure_msg = f"Regions with more than one row: {failed}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Only one row per region is returned, even if there are ties.",
            failure_msg
        )

    def test_uses_row_number_window_function(self):
        """Check that the query uses ROW_NUMBER() OVER (PARTITION BY region ORDER BY amount DESC)."""
        with open("query.sql", "r") as file:
            query = file.read().lower().replace('\\n', ' ').replace('\\r', ' ')
        has_row_number = (
            "row_number" in query
            and "over" in query
            and "partition by" in query
            and "region" in query
            and "order by" in query
            and "amount" in query
            and "desc" in query
        )
        _dynamic_test(
            self,
            has_row_number,
            "Query uses ROW_NUMBER() OVER (PARTITION BY region ORDER BY amount DESC).",
            "Query does not use ROW_NUMBER() OVER (PARTITION BY region ORDER BY amount DESC)."
        )


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 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: Top Sales per Region

Lösung

Challenge: Top Sales per Region

Lösung