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

組み込みSQL関数を活用して、文字列、日付、数値、論理値、NULL、JSONデータを操作する方法を習得します。本モジュールでは、実践的なアプローチでSQL関数の理解と応用を深め、現実のシナリオにおけるデータクエリおよび変換スキルを向上させます。

データベース内のテキストデータを操作および分析するための、SQLの強力な文字列関数を学習します。

チャレンジ：デフォルトボーナスの割り当て

解答