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_matches_performance(self):
        """Returns a row for every performance review in the performance table."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT COUNT(*) FROM performance;")
        expected_count = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        actual_count = len(self.cursor.fetchall())
        _dynamic_test(
            self,
            actual_count == expected_count,
            f"Returned {actual_count} rows, matching performance table.",
            f"Returned {actual_count} rows, but performance table has {expected_count} rows."
        )

    def test_columns_present(self):
        """Includes columns: employee_id, name, review_date, score, and score_difference."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected = {"employee_id", "name", "review_date", "score", "score_difference"}
        condition = expected.issubset(set(columns))
        _dynamic_test(
            self,
            condition,
            "All required columns are present.",
            f"Missing columns: {expected - set(columns)}"
        )

    def test_score_difference_logic(self):
        """For each employee, score_difference is the difference between current and previous review score, ordered by review_date. First review is NULL."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Get expected data
        self.cursor.execute("SELECT employee_id, review_date, score FROM performance ORDER BY employee_id, review_date;")
        perf = self.cursor.fetchall()
        expected = []
        from collections import defaultdict
        by_emp = defaultdict(list)
        for eid, date, score in perf:
            by_emp[eid].append((date, score))
        for eid in sorted(by_emp.keys()):
            reviews = sorted(by_emp[eid], key=lambda x: x[0])
            last = None
            for date, score in reviews:
                if last is None:
                    expected.append((eid, date, score, None))
                else:
                    expected.append((eid, date, score, score - last))
                last = score
        # Query user's result
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        # Map by employee_id and review_date
        idx_eid = columns.index("employee_id")
        idx_date = columns.index("review_date")
        idx_score = columns.index("score")
        idx_diff = columns.index("score_difference")
        result_map = {(r[idx_eid], r[idx_date]): (r[idx_score], r[idx_diff]) for r in rows}
        failed = []
        for eid, date, score, diff in expected:
            if (eid, date) not in result_map:
                failed.append(f"Missing row for employee {eid} on {date}")
                continue
            user_score, user_diff = result_map[(eid, date)]
            if user_score != score:
                failed.append(f"Wrong score for employee {eid} on {date}: got {user_score}, expected {score}")
            # Accept None or NULL for first diff
            if diff is None:
                if user_diff is not None:
                    failed.append(f"score_difference for employee {eid} on {date} should be NULL, got {user_diff}")
            else:
                if user_diff != diff:
                    failed.append(f"score_difference for employee {eid} on {date}: got {user_diff}, expected {diff}")
        _dynamic_test(
            self,
            len(failed) == 0,
            "score_difference is correct for all employees and reviews.",
            "\\n".join(failed)
        )

    def test_ordering(self):
        """Results are ordered by employee_id and review_date."""
        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]
        idx_eid = columns.index("employee_id")
        idx_date = columns.index("review_date")
        ordered = True
        last = None
        for row in rows:
            key = (row[idx_eid], row[idx_date])
            if last is not None:
                if key < last:
                    ordered = False
                    break
            last = key
        _dynamic_test(
            self,
            ordered,
            "Rows are ordered by employee_id and review_date.",
            "Rows are not ordered by employee_id and review_date."
        )

    def test_name_is_correct(self):
        """name column matches employees table for each employee_id."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT employee_id, name FROM employees;")
        emp_map = dict(self.cursor.fetchall())
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        idx_eid = columns.index("employee_id")
        idx_name = columns.index("name")
        fail = []
        for row in rows:
            eid = row[idx_eid]
            name = row[idx_name]
            if eid not in emp_map or emp_map[eid] != name:
                fail.append(f"name for employee_id {eid} should be '{emp_map.get(eid)}', got '{name}'")
        _dynamic_test(
            self,
            len(fail) == 0,
            "name column matches employees table for each employee_id.",
            "\\n".join(fail)
        )

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: Compare Performance Over Time

Рішення

Challenge: Compare Performance Over Time

Рішення