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_row_count(self):
        """Check that the query returns one row per employee."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT COUNT(*) FROM employees")
        expected_count = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == expected_count,
            "Query returns one row per employee.",
            f"Query returned {len(rows)} rows, but expected {expected_count}."
        )

    def test_earliest_salary_per_department(self):
        """Check that department_earliest_salary is correct for each employee."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        for dept in df['department'].unique():
            dept_df = df[df['department'] == dept]
            self.cursor.execute(
                """
                SELECT salary FROM employees WHERE department = %s ORDER BY hire_date ASC LIMIT 1
                """,
                (dept,)
            )
            expected_earliest = float(self.cursor.fetchone()[0])
            actual_earliest_set = set(dept_df['department_earliest_salary'].astype(float))
            _dynamic_test(
                self,
                actual_earliest_set == {expected_earliest},
                f"department_earliest_salary is correct for department {dept}.",
                f"department_earliest_salary is incorrect for department {dept}. Expected {expected_earliest}, found {actual_earliest_set}."
            )

    def test_salary_difference(self):
        """Check that salary_difference is correctly computed."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        for _, row in df.iterrows():
            salary = float(row['salary'])
            earliest = float(row['department_earliest_salary'])
            expected_diff = salary - earliest
            actual_diff = float(row['salary_difference'])
            _dynamic_test(
                self,
                abs(expected_diff - actual_diff) < 0.01,
                f"salary_difference is correct for employee_id {row['employee_id']}.",
                f"salary_difference is incorrect for employee_id {row['employee_id']}. Expected {expected_diff}, found {actual_diff}."
            )

    def test_window_function_used(self):
        """Check that a window function is used to determine earliest salary per department."""
        with open("query.sql", "r") as file:
            query = file.read().lower().replace('\\n', ' ')
        has_window = (
            'first_value' in query and
            'over' in query and
            'partition by' in query and
            'order by' in query
        )
        _dynamic_test(
            self,
            has_window,
            "Query uses a window function to determine earliest salary per department.",
            "Query does not use a window function to determine earliest salary per department."
        )

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: Salary Progression Analysis

Lösung

Challenge: Salary Progression Analysis

Lösung