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 test_all_rows_returned(self):
        """Check if the query returns all rows from employees."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == 10,
            "Query returns all 10 employee rows.",
            f"Query does not return all employee rows. Returned: {len(rows)}."
        )

    def test_bonus_nulls_replaced_with_500(self):
        """Check if bonus NULLs are replaced with 500."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)
        # Employees with NULL bonus in original data
        null_bonus_ids = [2,5,8,10]
        null_bonus_values = result_df[result_df['employee_id'].isin(null_bonus_ids)]['bonus'].tolist()
        condition = all(float(v) == 500 for v in null_bonus_values)
        _dynamic_test(
            self,
            condition,
            "All NULL bonus values correctly replaced with 500.",
            f"Bonus values for employees with NULLs not all replaced with 500: {null_bonus_values}"
        )

    def test_bonus_not_null_unchanged(self):
        """Check if original non-NULL bonus values are unchanged."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)
        # Employees with NOT NULL bonus in original data
        expected = {
            1: Decimal('1500.00'),
            3: Decimal('2000.00'),
            4: Decimal('1200.00'),
            6: Decimal('1800.00'),
            7: Decimal('1300.00'),
            9: Decimal('1100.00')
        }
        actual = result_df[result_df['employee_id'].isin(expected.keys())].set_index('employee_id')['bonus'].to_dict()
        correct = all(abs(actual[eid] - expected[eid]) < Decimal('0.001') for eid in expected)
        _dynamic_test(
            self,
            correct,
            "All non-NULL bonus values remain unchanged.",
            f"Some non-NULL bonus values were modified: {actual}"
        )

    def test_all_other_columns_unchanged(self):
        """Check if all other columns match original data."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)
        # Check that columns are present
        required_cols = set(['employee_id','name','department','manager_id','bonus'])
        col_cond = required_cols.issubset(set(result_df.columns))
        # Check that all non-bonus columns match original data
        # Original data
        orig = [
            (1, 'Alice Johnson', 'HR', None),
            (2, 'Bob Smith', 'IT', 1),
            (3, 'Carol White', 'Finance', 1),
            (4, 'David Brown', 'IT', 2),
            (5, 'Eve Black', 'HR', 1),
            (6, 'Frank Green', 'Finance', 3),
            (7, 'Grace Lee', 'IT', 2),
            (8, 'Hannah King', 'Marketing', None),
            (9, 'Ian Gray', 'Marketing', 8),
            (10, 'Jackie Miller', 'Finance', 3)
        ]
        match = True
        for eid, name, dept, mid in orig:
            row = result_df[result_df['employee_id']==eid]
            if row.empty:
                match = False
                break
            r = row.iloc[0]
            if r['name'] != name or r['department'] != dept or (pd.isna(r['manager_id']) and mid is not None) or (r['manager_id'] != mid and not (pd.isna(r['manager_id']) and mid is None)):
                match = False
                break
        _dynamic_test(
            self,
            col_cond and match,
            "All other columns are unchanged and present.",
            "Other columns were modified or missing."
        )


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

This beginner course focuses on the essential SQL functions used for data cleaning and transformation. Learners will practice handling missing values, formatting and standardizing data, and applying string, date, numeric, logical, and null handling functions to prepare datasets for analysis. The course provides hands-on challenges directly related to the most common data preparation tasks in SQL.

Challenge: Assign Default Bonuses

解答