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 _get_user_query(self):
        with open("query.sql", "r") as file:
            return file.read()

    def test_columns_present(self):
        query = self._get_user_query()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required_columns = ["sale_id", "sale_date", "employee_id", "amount", "running_total"]
        missing = [col for col in required_columns if col not in columns]
        _dynamic_test(
            self,
            len(missing) == 0,
            "All required columns are present.",
            f"Missing columns: {', '.join(missing)}"
        )

    def test_result_ordering(self):
        query = self._get_user_query()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        if df.empty:
            _dynamic_test(self, False, "Result is not empty.", "Query returned no results.")
            return
        sorted_df = df.sort_values(by=["employee_id", "sale_date", "sale_id"]).reset_index(drop=True)
        is_sorted = df.reset_index(drop=True).equals(sorted_df)
        _dynamic_test(
            self,
            is_sorted,
            "Result is ordered by employee_id, sale_date, sale_id.",
            "Result is not ordered by employee_id, sale_date, sale_id."
        )

    def test_running_total_correct(self):
        query = self._get_user_query()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        error_msgs = []
        for emp_id, group in df.groupby("employee_id"):
            group = group.sort_values(["sale_date", "sale_id"])
            running_sum = 0
            for idx, row in group.iterrows():
                running_sum += float(row["amount"])
                actual = float(row["running_total"])
                if abs(running_sum - actual) > 0.01:
                    error_msgs.append(f"Employee {emp_id}, sale_id {row['sale_id']}: expected {running_sum}, got {actual}")
        _dynamic_test(
            self,
            len(error_msgs) == 0,
            "Running total is correct for all employees.",
            "\\n".join(error_msgs)
        )

    def test_running_total_non_decreasing(self):
        query = self._get_user_query()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        error_msgs = []
        for emp_id, group in df.groupby("employee_id"):
            group = group.sort_values(["sale_date", "sale_id"])
            prev = None
            for idx, row in group.iterrows():
                current = float(row["running_total"])
                if prev is not None and current < prev - 0.01:
                    error_msgs.append(f"Employee {emp_id}, sale_id {row['sale_id']}: running_total decreased from {prev} to {current}")
                prev = current
        _dynamic_test(
            self,
            len(error_msgs) == 0,
            "Running total never decreases for each employee.",
            "\\n".join(error_msgs)
        )

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: Calculate Running Total of Sales

Рішення

Challenge: Calculate Running Total of Sales

Рішення