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

Master the art of SQL joins with engaging, real-world examples and hands-on challenges. This course takes you from foundational join concepts to advanced multi-table queries, ensuring you can confidently analyze and combine data from multiple sources.

Explore the essential SQL join types—INNER, LEFT, RIGHT, and FULL OUTER JOIN—through practical scenarios and hands-on challenges.

Level up your SQL skills by mastering multi-table joins, self-joins, and advanced join conditions for complex data analysis.

Master the nuances of NULL handling in joins, optimize join performance, and tackle real-world data analysis challenges.

Challenge: Classes Without Teachers

解答