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_one_row_per_student(self):
        """Check that the query returns one row per student."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT COUNT(*) FROM students")
        student_count = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == student_count,
            "Query returns one row per student.",
            f"Query returns {len(rows)} rows, but there are {student_count} students."
        )

    def test_student_name_column(self):
        """Check that the student_name column contains the student's name."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        self.cursor.execute("SELECT student_id, name FROM students")
        expected = dict(self.cursor.fetchall())
        for idx, row in df.iterrows():
            # Find student_id by matching name
            name = row['student_name']
            _dynamic_test(
                self,
                name in expected.values(),
                f"Student name '{name}' is present in the student_name column.",
                f"Student name '{name}' is not a valid student from the students table."
            )

    def test_class_name_column_nullable(self):
        """Check that class_name is correct or NULL for students with no class."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        self.cursor.execute("SELECT student_id, class_id FROM students")
        student_classes = dict(self.cursor.fetchall())
        self.cursor.execute("SELECT class_id, class_name FROM classes")
        class_names = dict(self.cursor.fetchall())
        for idx, row in df.iterrows():
            student_name = row['student_name']
            self.cursor.execute("SELECT student_id FROM students WHERE name = %s", (student_name,))
            student_id = self.cursor.fetchone()[0]
            class_id = student_classes[student_id]
            if class_id is None:
                _dynamic_test(
                    self,
                    pd.isna(row['class_name']),
                    f"Student '{student_name}' has NULL class_name as expected.",
                    f"Student '{student_name}' should have NULL class_name, but got '{row['class_name']}'."
                )
            else:
                expected_class = class_names[class_id]
                _dynamic_test(
                    self,
                    row['class_name'] == expected_class,
                    f"Student '{student_name}' has correct class_name '{expected_class}'.",
                    f"Student '{student_name}' should have class_name '{expected_class}', but got '{row['class_name']}'."
                )

    def test_teacher_name_column_nullable(self):
        """Check that teacher_name is correct or NULL for students with no class or class with no teacher."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        self.cursor.execute("SELECT student_id, class_id FROM students")
        student_classes = dict(self.cursor.fetchall())
        self.cursor.execute("SELECT class_id, teacher_id FROM classes")
        class_teachers = dict(self.cursor.fetchall())
        self.cursor.execute("SELECT teacher_id, name FROM teachers")
        teacher_names = dict(self.cursor.fetchall())
        for idx, row in df.iterrows():
            student_name = row['student_name']
            self.cursor.execute("SELECT student_id FROM students WHERE name = %s", (student_name,))
            student_id = self.cursor.fetchone()[0]
            class_id = student_classes[student_id]
            if class_id is None:
                _dynamic_test(
                    self,
                    pd.isna(row['teacher_name']),
                    f"Student '{student_name}' has NULL teacher_name as expected (no class).",
                    f"Student '{student_name}' should have NULL teacher_name (no class), but got '{row['teacher_name']}'."
                )
            else:
                teacher_id = class_teachers[class_id]
                if teacher_id is None:
                    _dynamic_test(
                        self,
                        pd.isna(row['teacher_name']),
                        f"Student '{student_name}' has NULL teacher_name as expected (class has no teacher).",
                        f"Student '{student_name}' should have NULL teacher_name (class has no teacher), but got '{row['teacher_name']}'."
                    )
                else:
                    expected_teacher = teacher_names[teacher_id]
                    _dynamic_test(
                        self,
                        row['teacher_name'] == expected_teacher,
                        f"Student '{student_name}' has correct teacher_name '{expected_teacher}'.",
                        f"Student '{student_name}' should have teacher_name '{expected_teacher}', but got '{row['teacher_name']}'."
                    )

    def test_average_grade_column(self):
        """Check that average_grade is correct or NULL for students with no grades."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        self.cursor.execute("SELECT student_id FROM students")
        student_ids = [row[0] for row in self.cursor.fetchall()]
        for idx, row in df.iterrows():
            student_name = row['student_name']
            self.cursor.execute("SELECT student_id FROM students WHERE name = %s", (student_name,))
            student_id = self.cursor.fetchone()[0]
            self.cursor.execute("SELECT AVG(score) FROM grades WHERE student_id = %s", (student_id,))
            avg = self.cursor.fetchone()[0]
            if avg is None:
                _dynamic_test(
                    self,
                    pd.isna(row['average_grade']),
                    f"Student '{student_name}' has NULL average_grade as expected.",
                    f"Student '{student_name}' should have NULL average_grade, but got '{row['average_grade']}'."
                )
            else:
                # Allow up to 0.01 difference due to floating point
                _dynamic_test(
                    self,
                    abs(row['average_grade'] - avg) < 0.01,
                    f"Student '{student_name}' has correct average_grade {avg:.2f}.",
                    f"Student '{student_name}' should have average_grade {avg:.2f}, but got {row['average_grade']}."
                )

    def test_column_names(self):
        """Check that the correct columns are present: student_name, class_name, teacher_name, average_grade."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected_columns = set(["student_name", "class_name", "teacher_name", "average_grade"])
        _dynamic_test(
            self,
            set(columns) == expected_columns,
            "Query returns the correct column names.",
            f"Query columns are {columns}, but expected {expected_columns}."
        )

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: Student Performance Dashboard

解答