import unittest
import psycopg2
import pandas as pd
from decimal import Decimal

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 setUp(self):
        self.cursor.execute("ROLLBACK; BEGIN;")

    def test_columns_present(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 = [
            "performance_id",
            "employee_id",
            "review_date",
            "score",
            "name",
            "department",
            "transaction_amount",
            "avg_last_12_months",
            "is_spike"
        ]
        missing = [col for col in expected_columns if col not in columns]
        _dynamic_test(
            self,
            len(missing) == 0,
            "All required columns are present.",
            f"Missing columns: {', '.join(missing)}"
        )

    def test_avg_last_12_months_calculation(self):
        """Check avg_last_12_months is correct for at least one 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)
        # Pick Alice Johnson (employee_id=1)
        alice = df[df['employee_id'] == 1].sort_values('review_date')
        # Her two reviews: 2022-03-20 (score=8), 2023-03-18 (score=9)
        # First review: no prior 12 months, avg_last_12_months should be null
        first = alice.iloc[0]
        _dynamic_test(
            self,
            pd.isnull(first['avg_last_12_months']),
            "First transaction has null avg_last_12_months.",
            "First transaction should have null avg_last_12_months."
        )
        # Second review: prior 12 months includes only first review (score=8)
        second = alice.iloc[1]
        avg_val = second['avg_last_12_months']
        if isinstance(avg_val, float):
            diff = abs(avg_val - 8.0)
        elif isinstance(avg_val, Decimal):
            diff = abs(avg_val - Decimal('8.0'))
        else:
            diff = abs(float(avg_val) - 8.0)
        _dynamic_test(
            self,
            diff < 0.001,
            "Second transaction avg_last_12_months is correct.",
            f"Second transaction avg_last_12_months should be 8.0, got {second['avg_last_12_months']}"
        )

    def test_is_spike_flag(self):
        """Check is_spike is set correctly for all transactions."""
        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 Alice Johnson (employee_id=1):
        # 2022-03-20: avg_last_12_months=null, score=8, is_spike=0
        # 2023-03-18: avg_last_12_months=8, score=9, 9 < 16, so is_spike=0
        alice = df[df['employee_id'] == 1].sort_values('review_date')
        _dynamic_test(
            self,
            alice.iloc[0]['is_spike'] == 0,
            "First Alice transaction is not flagged as spike.",
            "First Alice transaction should not be flagged as spike."
        )
        _dynamic_test(
            self,
            alice.iloc[1]['is_spike'] == 0,
            "Second Alice transaction is not flagged as spike.",
            "Second Alice transaction should not be flagged as spike."
        )
        # For Bob Smith (employee_id=2):
        # 2021-08-01: avg_last_12_months=null, is_spike=0
        # 2022-08-01: avg=7, score=8, 8<14, is_spike=0
        # 2023-08-01: avg=(7+8)/2=7.5, score=7, 7<15, is_spike=0
        bob = df[df['employee_id'] == 2].sort_values('review_date')
        _dynamic_test(
            self,
            all(bob['is_spike'] == 0),
            "Bob Smith has no spikes, as expected.",
            "Bob Smith transactions should not be flagged as spikes."
        )

    def test_single_transaction_employee(self):
        """Check that employees with one transaction have null avg_last_12_months and is_spike=0."""
        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)
        # Frank Green (employee_id=6) has only one review
        frank = df[df['employee_id'] == 6]
        _dynamic_test(
            self,
            len(frank) == 1 and pd.isnull(frank.iloc[0]['avg_last_12_months']) and frank.iloc[0]['is_spike'] == 0,
            "Single transaction employee handled correctly.",
            "Single transaction employee avg_last_12_months should be null and is_spike should be 0."
        )

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: Detect Spending Spikes

Рішення

Challenge: Detect Spending Spikes

Рішення