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_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 = {"employee_id", "review_date", "sales_achieved", "previous_sales_achieved", "percentile"}
        _dynamic_test(
            self,
            set(columns) == expected,
            "All required columns are present.",
            f"Missing or extra columns. Expected: {expected}, Got: {set(columns)}"
        )

    def test_sales_department_only(self):
        """Check that only Sales department employees are included."""
        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)
        # Get all employee_ids in Sales
        self.cursor.execute("SELECT employee_id FROM employees WHERE department = 'Sales';")
        sales_ids = set(r[0] for r in self.cursor.fetchall())
        present_ids = set(df['employee_id'])
        _dynamic_test(
            self,
            present_ids.issubset(sales_ids) and len(present_ids) > 0,
            "Only Sales department employees are included.",
            f"Non-Sales employee(s) found: {present_ids - sales_ids}"
        )

    def test_previous_sales_achieved_lag(self):
        """Check that previous_sales_achieved is correct using LAG()."""
        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 each employee, check lag logic
        for eid in df['employee_id'].unique():
            emp_df = df[df['employee_id'] == eid].sort_values('review_date')
            prev = None
            for idx, row in emp_df.iterrows():
                if prev is None:
                    _dynamic_test(
                        self,
                        pd.isna(row['previous_sales_achieved']),
                        f"First review for employee {eid} has NULL previous_sales_achieved.",
                        f"First review for employee {eid} should have NULL previous_sales_achieved."
                    )
                else:
                    _dynamic_test(
                        self,
                        row['previous_sales_achieved'] == prev,
                        f"Correct previous_sales_achieved for employee {eid} at review {row['review_date']}.",
                        f"Incorrect previous_sales_achieved for employee {eid} at review {row['review_date']}. Expected {prev}, got {row['previous_sales_achieved']}"
                    )
                prev = row['sales_achieved']

    def test_percentile_rank_partition(self):
        """Check that percentile is calculated per review_date using PERCENT_RANK()."""
        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 date in df['review_date'].unique():
            subset = df[df['review_date'] == date]
            if len(subset) > 1:
                sorted_sales = subset.sort_values('sales_achieved')['sales_achieved'].tolist()
                expected_percentiles = []
                n = len(sorted_sales)
                for i, sale in enumerate(sorted_sales):
                    if n == 1:
                        expected_percentiles.append(0.0)
                    else:
                        expected_percentiles.append(i / (n - 1))
                # Compare with actual percentiles (sorted by sales_achieved)
                actual_percentiles = subset.sort_values('sales_achieved')['percentile'].tolist()
                # Allow for floating point tolerance
                for exp, act in zip(expected_percentiles, actual_percentiles):
                    _dynamic_test(
                        self,
                        abs(float(exp) - float(act)) < 0.01,
                        f"Percentile ranks for review_date {date} are correct.",
                        f"Percentile rank mismatch for review_date {date}. Expected {exp}, got {act}"
                    )

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 SQL from the ground up with a focus on business intelligence. Learn to write powerful queries for KPIs, implement business logic, analyze trends with window functions, design data models for BI tools, and apply reporting patterns. Each section blends engaging theory with hands-on, real-world challenges to prepare you for BI analyst roles.

Learn to use SQL aggregations to calculate and interpret key business metrics, such as revenue, customer counts, and averages, essential for BI analysis.

Implement business rules and conditional logic in SQL queries using CASE WHEN, enabling dynamic categorization and decision-making.

Analyze business trends and rankings over time using advanced SQL window functions, essential for BI trend analysis and leaderboards.

Challenge: Trend and Percentile Analysis

Рішення

Challenge: Trend and Percentile Analysis

Рішення