import unittest
import psycopg2
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_initial_explain_present(self):
        """Check that the first EXPLAIN statement is present and analyzes the correct query."""
        with open("query.sql", "r") as file:
            content = file.read()
        explain_count = content.lower().count("explain")
        has_explain = explain_count >= 2
        has_initial_explain = normalize_text("explain\\nselect\\n    d.driver_id") in normalize_text(content)
        _dynamic_test(
            self,
            has_explain and has_initial_explain,
            "Initial EXPLAIN statement for the summary report query is present.",
            "Initial EXPLAIN statement analyzing the summary report query is missing."
        )

    def test_index_created(self):
        """Check that at least one index is created on a relevant join column."""
        with open("query.sql", "r") as file:
            content = normalize_text(file.read())
        # Look for index creation on rides.driver_id or payments.ride_id
        has_rides_driver_idx = "create index" in content and "on rides(driver_id)" in content
        has_payments_ride_idx = "create index" in content and "on payments(ride_id)" in content
        _dynamic_test(
            self,
            has_rides_driver_idx or has_payments_ride_idx,
            "At least one relevant index is created for the query joins.",
            "No relevant index for query joins was created."
        )

    def test_final_explain_present(self):
        """Check that the final EXPLAIN statement is present and analyzes the correct query after optimization."""
        with open("query.sql", "r") as file:
            content = file.read()
        explain_count = content.lower().count("explain")
        has_final_explain = explain_count >= 2
        # Check that the query after the index is the same as the first
        lines = [normalize_text(line) for line in content.splitlines() if line.strip()]
        explain_lines = [i for i, l in enumerate(lines) if l.startswith("explain")]
        # The final EXPLAIN should be after an index is created
        has_index_between = False
        if len(explain_lines) >= 2:
            between = lines[explain_lines[0]:explain_lines[1]]
            has_index_between = any("create index" in l for l in between)
        _dynamic_test(
            self,
            has_final_explain and has_index_between,
            "Final EXPLAIN statement analyzing the query after optimization is present.",
            "Final EXPLAIN statement after optimization is missing or not after index creation."
        )

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

Dive deep into the world of SQL query optimization by mastering the EXPLAIN statement and understanding how query planners work. This course blends engaging theory with hands-on, real-world tasks to help you analyze, interpret, and optimize SQL queries using EXPLAIN and related tools.

Learn the fundamentals of the EXPLAIN statement, how query planners work, and how to interpret basic query plans.

Delve into more complex query plans, including multi-table joins, subqueries, and the impact of indexes.

Apply your knowledge to complex, real-world scenarios involving large datasets, reporting queries, and performance troubleshooting.

Challenge: Optimizing Ride Summary Reports

Løsning