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_only_employees_with_multiple_projects(self):
        """Check result only includes employees assigned to more than one project."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)

        # Find employees with more than one project
        self.cursor.execute("""
            SELECT employee_id FROM employee_projects GROUP BY employee_id HAVING COUNT(project_id) > 1
        """)
        expected_ids = set(row[0] for row in self.cursor.fetchall())
        actual_ids = set(result_df['employee_id']) if not result_df.empty else set()
        condition = actual_ids == expected_ids
        missing = expected_ids - actual_ids
        extra = actual_ids - expected_ids
        msg = ""
        if missing:
            msg += f"Missing employees: {missing}. "
        if extra:
            msg += f"Unexpected employees: {extra}."
        _dynamic_test(
            self,
            condition,
            "Returned only employees assigned to more than one project.",
            f"Returned incorrect employees. {msg}"
        )

    def test_selects_correct_columns(self):
        """Check result includes only employee_id, first_name, last_name columns."""
        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 = {"employee_id", "first_name", "last_name"}
        condition = set(columns) == expected_columns
        _dynamic_test(
            self,
            condition,
            "Query selects correct columns.",
            f"Query columns are incorrect: {columns}"
        )

    def test_uses_in_with_nested_subquery(self):
        """Check that the query uses IN with a nested subquery in WHERE."""
        with open("query.sql", "r") as file:
            query = file.read().lower().replace('\\n', ' ')
        # Look for pattern: WHERE ... IN ( SELECT ... )
        uses_in = "where" in query and " in (" in query and "select" in query
        _dynamic_test(
            self,
            uses_in,
            "Query uses IN with a nested subquery in WHERE.",
            "Query does not use IN with a nested subquery in WHERE."
        )

    def test_query_works_with_initial_data(self):
        """Check query runs without errors and returns non-empty result for provided data."""
        with open("query.sql", "r") as file:
            query = file.read()
        try:
            self.cursor.execute(query)
            rows = self.cursor.fetchall()
            condition = isinstance(rows, list)
            _dynamic_test(
                self,
                condition,
                "Query runs successfully with initial data.",
                "Query failed to run with initial data."
            )
        except Exception as e:
            _dynamic_test(
                self,
                False,
                "Query runs successfully with initial data.",
                f"Query failed to run: {e}"
            )

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 writing and optimizing nested subqueries in SQL. This course explores advanced querying techniques using subqueries in WHERE, FROM, and SELECT clauses, and teaches best practices for performance optimization. Each section blends engaging theory with real-world challenges to solidify your understanding.

Explore how to use nested subqueries within the WHERE clause to filter data based on complex conditions and relationships.

Learn how to use nested subqueries in the FROM clause to create derived tables for complex data analysis.

Discover how to use nested subqueries in the SELECT clause to compute values for each row in your result set.

Master best practices and techniques for optimizing nested subqueries for performance and maintainability.

Challenge: Employees on Multiple Projects

解答