import unittest
import psycopg2
import pandas as pd
import re

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):
        # Restore all tables to known state before each test
        self.cursor.execute("DELETE FROM assignments;")
        self.cursor.execute("DELETE FROM projects;")
        self.cursor.execute("DELETE FROM employees;")

        self.cursor.execute("""
            INSERT INTO employees (employee_id, name, manager_id)
            VALUES
                (1, 'Alice Johnson', NULL),
                (2, 'Bob Smith', 1),
                (3, 'Carol White', 1),
                (4, 'David Brown', 2),
                (5, 'Eve Black', 2),
                (6, 'Frank Green', 3),
                (7, 'Grace Lee', 3),
                (8, 'Hannah Adams', 1);
        """)

        self.cursor.execute("DELETE FROM departments;")
        self.cursor.execute("""
            INSERT INTO departments (department_id, department_name)
            VALUES
                (1, 'Engineering'),
                (2, 'HR'),
                (3, 'Finance'),
                (4, 'Marketing');
        """)

        self.cursor.execute("""
            INSERT INTO projects (project_id, project_name, department_id)
            VALUES
                (1, 'Website Redesign', 1),
                (2, 'Recruitment Drive', 2),
                (3, 'Budget Planning', 3),
                (4, 'Product Launch', 4);
        """)

        self.cursor.execute("""
            INSERT INTO assignments (assignment_id, employee_id, project_id)
            VALUES
                (1, 2, 1),
                (2, 4, 1),
                (3, 5, 4),
                (4, 2, 4),
                (5, 3, 2),
                (6, 6, 2),
                (7, 7, 2),
                (8, 8, 3);
        """)
        self.connection.commit()

    def test_all_assignments_present(self):
        """Check if all assignments are present with correct joins."""
        with open("query.sql", "r") as file:
            query = file.read()

        solution_query = """
            SELECT
                assignments.assignment_id,
                employees.name AS employee_name,
                projects.project_name,
                departments.department_name
            FROM assignments
            JOIN employees ON assignments.employee_id = employees.employee_id
            JOIN projects ON assignments.project_id = projects.project_id
            JOIN departments ON projects.department_id = departments.department_id
        """

        self.cursor.execute(solution_query)
        solution_rows = self.cursor.fetchall()
        solution_columns = [desc[0] for desc in self.cursor.description]
        solution_df = pd.DataFrame(solution_rows, columns=solution_columns)

        self.cursor.execute(query)
        user_rows = self.cursor.fetchall()
        user_columns = [desc[0] for desc in self.cursor.description]
        user_df = pd.DataFrame(user_rows, columns=user_columns)

        # Normalize both dataframes safely (no applymap)
        solution_df = normalize_df(solution_df)
        user_df = normalize_df(user_df)

        # Compare by expected rows (order-independent)
        expected_cols = list(solution_df.columns)
        try:
            user_df_subset = user_df[expected_cols]
        except KeyError:
            _dynamic_test(
                self,
                False,
                "All expected assignment rows are present with correct joins.",
                "User query does not output the required columns for row comparison."
            )
            return

        expected_set = rows_as_set(solution_df, expected_cols)
        actual_set = rows_as_set(user_df_subset, expected_cols)

        missing = expected_set - actual_set
        condition = len(missing) == 0

        failure_msg = ""
        if not condition:
            failure_msg = "The following expected rows are missing:\\n" + "\\n".join(sorted(missing))

        _dynamic_test(
            self,
            condition,
            "All expected assignment rows are present with correct joins.",
            failure_msg
        )

    def test_output_columns(self):
        """Check output columns are correct and in any order."""
        with open("query.sql", "r") as file:
            query = file.read()

        self.cursor.execute(query)
        user_columns = set([normalize_text(desc[0]) for desc in self.cursor.description])

        expected_columns = set([
            "assignment_id",
            "employee_name",
            "project_name",
            "department_name"
        ])

        condition = expected_columns == user_columns
        missing = expected_columns - user_columns
        extra = user_columns - expected_columns

        failure_msg = ""
        if not condition:
            failure_msg = f"Output columns mismatch. Missing: {missing}. Extra: {extra}."

        _dynamic_test(
            self,
            condition,
            "Output includes assignment_id, employee_name, project_name, department_name.",
            failure_msg
        )

    def test_inner_joins_only(self):
        """Check only assignments with valid links in all tables are returned (inner join behavior)."""
        self.cursor.execute("DELETE FROM assignments;")
        self.connection.commit()

        try:
            self.cursor.execute("ALTER TABLE assignments DISABLE TRIGGER ALL;")
            self.cursor.execute(
                "INSERT INTO assignments (assignment_id, employee_id, project_id) VALUES (99, 100, 200);"
            )
            self.connection.commit()

            with open("query.sql", "r") as file:
                query = file.read()

            self.cursor.execute(query)
            rows = self.cursor.fetchall()

            condition = len(rows) == 0
            _dynamic_test(
                self,
                condition,
                "Query returns only assignments with valid links in all tables.",
                "Query returns assignments with missing links; should only return valid ones."
            )
        finally:
            self.cursor.execute("DELETE FROM assignments WHERE assignment_id = 99;")
            self.cursor.execute("ALTER TABLE assignments ENABLE TRIGGER ALL;")
            self.connection.commit()

        # Restore assignments for other tests
        self.cursor.execute("DELETE FROM assignments;")
        self.cursor.execute("""
            INSERT INTO assignments (assignment_id, employee_id, project_id)
            VALUES
                (1, 2, 1),
                (2, 4, 1),
                (3, 5, 4),
                (4, 2, 4),
                (5, 3, 2),
                (6, 6, 2),
                (7, 7, 2),
                (8, 8, 3);
        """)
        self.connection.commit()

    def test_joins_on_correct_columns(self):
        """Check joins use correct columns by removing a join and expecting empty output."""
        self.cursor.execute("DELETE FROM assignments;")
        self.cursor.execute("DELETE FROM employees;")
        self.connection.commit()

        with open("query.sql", "r") as file:
            query = file.read()

        self.cursor.execute(query)
        rows = self.cursor.fetchall()

        condition = len(rows) == 0
        _dynamic_test(
            self,
            condition,
            "Joins are performed on correct columns (employees missing, so output empty).",
            "Query does not join on correct columns: should return no rows if employees table is empty."
        )

        # Restore employees and assignments for other tests
        self.cursor.execute("DELETE FROM employees;")
        self.cursor.execute("""
            INSERT INTO employees (employee_id, name, manager_id)
            VALUES
                (1, 'Alice Johnson', NULL),
                (2, 'Bob Smith', 1),
                (3, 'Carol White', 1),
                (4, 'David Brown', 2),
                (5, 'Eve Black', 2),
                (6, 'Frank Green', 3),
                (7, 'Grace Lee', 3),
                (8, 'Hannah Adams', 1);
        """)
        self.cursor.execute("DELETE FROM assignments;")
        self.cursor.execute("""
            INSERT INTO assignments (assignment_id, employee_id, project_id)
            VALUES
                (1, 2, 1),
                (2, 4, 1),
                (3, 5, 4),
                (4, 2, 4),
                (5, 3, 2),
                (6, 6, 2),
                (7, 7, 2),
                (8, 8, 3);
        """)
        self.connection.commit()


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: str) -> str:
    text = str(text).lower()
    text = re.sub(r"\s{2,}", " ", text)
    text = re.sub(r"\s*([,:?])\s*", r"\1 ", text)
    return text.strip()


def normalize_df(df: pd.DataFrame) -> pd.DataFrame:
    """
    Safe replacement for DataFrame.applymap(normalize_text).
    Works even if applymap is unavailable.
    """
    # normalize column names too (helps with weird casing/spaces)
    df = df.copy()
    df.columns = [normalize_text(c) for c in df.columns]

    # cast to string then normalize each column via Series.map
    for col in df.columns:
        df[col] = df[col].astype(str).map(normalize_text)
    return df


def rows_as_set(df: pd.DataFrame, cols: list[str]) -> set[str]:
    """
    Convert rows to a set of canonical strings so comparisons are order-independent.
    """
    tmp = df[cols].copy()
    # ensure stable column order
    tmp = tmp[sorted(tmp.columns)]
    return set(tmp.apply(lambda r: "|".join(r.values.tolist()), axis=1))


if __name__ == "__main__":
    unittest.main()


test_solution.py

実践的で現実世界の例やハンズオンチャレンジを通じて、SQL結合の技術を習得します。本コースでは、基礎的な結合の概念から高度な多テーブルクエリまでを網羅し、複数のデータソースからデータを自信を持って分析・統合できる力を身につけます。

実践的なシナリオとハンズオンチャレンジを通じて、主要なSQL結合タイプ（INNER、LEFT、RIGHT、FULL OUTER JOIN）を探求します。

複数テーブルの結合、自己結合、および複雑なデータ分析のための高度な結合条件を習得し、SQLスキルを向上させます。

結合におけるNULL処理の詳細を習得し、結合パフォーマンスを最適化し、実際のデータ分析課題に対応します。

チャレンジ：複雑なアサインメントマッチング

解答