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_returns_all_classes(self):
        """Check that all classes are returned regardless of teacher assignment."""
        with open("query.sql", "r") as file:
            query = file.read()

        self.cursor.execute("SELECT COUNT(*) FROM classes;")
        expected_count = self.cursor.fetchone()[0]

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

        _dynamic_test(
            self,
            actual_count == expected_count,
            f"Returned all {expected_count} classes.",
            f"Returned {actual_count} classes, expected {expected_count}."
        )

    def test_teacher_name_for_assigned(self):
        """Check that teacher name is correct for assigned classes."""
        with open("query.sql", "r") as file:
            query = file.read()

        # Get expected pairs (class_id, teacher_name)
        self.cursor.execute("""
            SELECT c.class_id, t.name FROM classes c
            JOIN teachers t ON c.teacher_id = t.teacher_id
        """)
        expected = {row[0]: row[1] for row in self.cursor.fetchall()}

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

        failures = []
        for class_id, teacher_name in expected.items():
            match = df[(df['class_id']==class_id) & (df['teacher_name']==teacher_name)]
            if match.empty:
                failures.append(f"class_id {class_id} should have teacher '{teacher_name}'")
        condition = len(failures) == 0
        _dynamic_test(
            self,
            condition,
            "Classes with assigned teachers display correct teacher name.",
            "\\n".join(failures)
        )

    def test_unassigned_label_for_null_teacher(self):
        """Check that 'Unassigned' is shown for classes without a teacher."""
        with open("query.sql", "r") as file:
            query = file.read()

        self.cursor.execute("SELECT class_id FROM classes WHERE teacher_id IS NULL;")
        unassigned_ids = [row[0] for row in self.cursor.fetchall()]

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

        failures = []
        for class_id in unassigned_ids:
            match = df[(df['class_id']==class_id) & (df['teacher_name']=='Unassigned')]
            if match.empty:
                failures.append(f"class_id {class_id} should have teacher_name 'Unassigned'")
        condition = len(failures) == 0
        _dynamic_test(
            self,
            condition,
            "All unassigned classes display 'Unassigned' as teacher name.",
            "\\n".join(failures)
        )

    def test_columns(self):
        """Check that the correct columns are returned: class_id, class_name, teacher_name."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected = ["class_id", "class_name", "teacher_name"]
        condition = columns == expected
        _dynamic_test(
            self,
            condition,
            "Returned columns are correct.",
            f"Returned columns: {columns}, expected: {expected}."
        )

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

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

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

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

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

チャレンジ：教師のいないクラス

解答