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_employees_with_no_assignments(self):
        """Returns all employees who have no rows in the assignments table."""
        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)

        # Employees with no assignments: Alice Johnson (1)
        expected = pd.DataFrame([
            {"employee_id": 1, "name": "Alice Johnson"}
        ])
        # Check if all expected rows are present
        missing_rows = []
        for _, expected_row in expected.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 employees are missing from the result:\\n{human_readable}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "All employees with no assignments are present in the result.",
            failure_msg
        )

    def test_does_not_include_employees_with_assignments(self):
        """Does not include employees who are assigned to at least 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)
        # Employees with assignments (should NOT be included)
        assigned_ids = [2,3,4,5,6,7,8]
        forbidden = result_df[result_df['employee_id'].isin(assigned_ids)]
        condition = forbidden.empty
        _dynamic_test(
            self,
            condition,
            "No employees with assignments are included in the result.",
            "Some employees with assignments are incorrectly included in the result."
        )

    def test_result_columns(self):
        """Returns correct columns: employee_id and name."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        condition = set(columns) == {"employee_id", "name"}
        _dynamic_test(
            self,
            condition,
            "Result contains only 'employee_id' and 'name' columns.",
            f"Result columns are incorrect: found {columns}."
        )

    def test_includes_managers_and_non_managers_with_no_assignments(self):
        """Includes both managers and non-managers if they have no project assignments."""
        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)
        # Alice Johnson is a manager (CEO, employee_id=1) and has no assignments
        condition = 1 in result_df['employee_id'].values
        _dynamic_test(
            self,
            condition,
            "Managers with no assignments are included.",
            "Managers with no assignments are missing from the result."
        )

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: Employees Without Projects

解答