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_unassigned_employees(self):
        """Check that only employees not in employee_projects are returned."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Find expected employee_ids not in employee_projects
        self.cursor.execute("SELECT employee_id FROM employees")
        all_ids = set(row[0] for row in self.cursor.fetchall())
        self.cursor.execute("SELECT DISTINCT employee_id FROM employee_projects")
        assigned_ids = set(row[0] for row in self.cursor.fetchall())
        expected_ids = all_ids - assigned_ids
        # Run user query
        self.cursor.execute(query)
        result = self.cursor.fetchall()
        result_ids = set(row[0] for row in result)
        _dynamic_test(
            self,
            result_ids == expected_ids,
            "Returned only employees not assigned to any project.",
            f"The result does not match the set of unassigned employees. Expected IDs: {expected_ids}, got: {result_ids}."
        )

    def test_no_assigned_employee_in_result(self):
        """Check that no assigned employee appears in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT DISTINCT employee_id FROM employee_projects")
        assigned_ids = set(row[0] for row in self.cursor.fetchall())
        self.cursor.execute(query)
        result = self.cursor.fetchall()
        result_ids = set(row[0] for row in result)
        overlap = result_ids & assigned_ids
        _dynamic_test(
            self,
            len(overlap) == 0,
            "No assigned employees are returned.",
            f"The following assigned employee_ids are incorrectly included: {overlap}"
        )

    def test_selects_correct_columns(self):
        """Check that only employee_id, first_name, last_name are selected."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected_cols = ["employee_id", "first_name", "last_name"]
        _dynamic_test(
            self,
            columns == expected_cols,
            "Query selects only employee_id, first_name, last_name columns.",
            f"Query selects columns {columns}, but expected {expected_cols}."
        )

    def test_all_unassigned_employees_are_returned(self):
        """Check that all unassigned employees are present in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT employee_id, first_name, last_name FROM employees WHERE employee_id NOT IN (SELECT employee_id FROM employee_projects)")
        expected_rows = set(self.cursor.fetchall())
        self.cursor.execute(query)
        user_rows = set(self.cursor.fetchall())
        missing = expected_rows - user_rows
        _dynamic_test(
            self,
            len(missing) == 0,
            "All unassigned employees are present in the result.",
            f"The following unassigned employees are missing: {missing}"
        )

    def test_no_extra_rows(self):
        """Check that the query does not return extra rows."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT employee_id, first_name, last_name FROM employees WHERE employee_id NOT IN (SELECT employee_id FROM employee_projects)")
        expected_rows = set(self.cursor.fetchall())
        self.cursor.execute(query)
        user_rows = set(self.cursor.fetchall())
        extra = user_rows - expected_rows
        _dynamic_test(
            self,
            len(extra) == 0,
            "No extra rows returned.",
            f"The following extra rows are returned: {extra}"
        )

    @classmethod
    def tearDownClass(cls):
        cls.cursor.close()
        cls.connection.close()

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 Not in Any Project

解答