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 test_joins_on_indexed_columns(self):
        """Check if joins are performed only on indexed columns (primary or foreign keys)."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized_query = normalize_text(query)
        join_conditions = [
            "s.class_id = c.class_id",
            "c.teacher_id = t.teacher_id",
            "s.student_id = g.student_id"
        ]
        condition = all(cond in normalized_query for cond in join_conditions)
        _dynamic_test(
            self,
            condition,
            "All joins are performed on indexed columns.",
            "Not all joins are performed on indexed columns (primary/foreign keys)."
        )

    def test_inner_joins_used(self):
        """Check if all joins are INNER JOINs."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized_query = normalize_text(query)
        required_joins = [
            "inner join classes c on s.class_id = c.class_id",
            "inner join teachers t on c.teacher_id = t.teacher_id",
            "inner join grades g on s.student_id = g.student_id"
        ]
        condition = all(join in normalized_query for join in required_joins)
        _dynamic_test(
            self,
            condition,
            "All joins are INNER JOINs with correct conditions.",
            "Not all joins are INNER JOINs with correct conditions."
        )

    def test_table_aliases_used(self):
        """Check if table aliases are used for all tables."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized_query = normalize_text(query)
        aliases = ["students s", "classes c", "teachers t", "grades g"]
        condition = all(alias in normalized_query for alias in aliases)
        _dynamic_test(
            self,
            condition,
            "Table aliases are used for all tables.",
            "Table aliases are not used for all tables."
        )

    def test_score_filter(self):
        """Check if the query filters for score > 80 only."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized_query = normalize_text(query)
        condition = "g.score > 80" in normalized_query or "g.score>80" in normalized_query
        _dynamic_test(
            self,
            condition,
            "Query filters for scores greater than 80.",
            "Query does not filter for scores greater than 80."
        )

    def test_returned_columns(self):
        """Check if the query returns the required columns with correct aliases."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required_columns = [
            "student_id",
            "student_name",
            "class_name",
            "teacher_name",
            "subject",
            "score"
        ]
        condition = all(col in columns for col in required_columns)
        missing = [col for col in required_columns if col not in columns]
        _dynamic_test(
            self,
            condition,
            "Query returns all required columns.",
            f"Query is missing columns: {', '.join(missing)}" if missing else ""
        )

    def test_expected_rows_present(self):
        """Check if only rows with score > 80 are returned and join is correct."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Solution query
        solution_query = """
        SELECT
            s.student_id,
            s.name AS student_name,
            c.class_name,
            t.name AS teacher_name,
            g.subject,
            g.score
        FROM
            students s
            INNER JOIN classes c ON s.class_id = c.class_id
            INNER JOIN teachers t ON c.teacher_id = t.teacher_id
            INNER JOIN grades g ON s.student_id = g.student_id
        WHERE
            g.score > 80;
        """
        self.cursor.execute(solution_query)
        solution_rows = self.cursor.fetchall()
        solution_columns = [desc[0] for desc in self.cursor.description]
        solution_result_df = pd.DataFrame(solution_rows, columns=solution_columns)
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        # Find missing rows
        missing_rows = []
        for _, expected_row in solution_result_df.iterrows():
            expected_dict = expected_row.to_dict()
            mask = (result_df[list(expected_dict.keys())] == pd.Series(expected_dict)).all(axis=1)
            if not mask.any():
                missing_rows.append(expected_dict)
        condition = len(missing_rows) == 0
        if not condition:
            human_readable = "\\n".join(str(r) for r in missing_rows)
            failure_msg = f"The following expected rows are missing:\\n{human_readable}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "All expected rows are present in the result.",
            failure_msg
        )

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 module 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.

Challenge: Optimize a Join Query

Рішення