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_all_employees_included(self):
        """Check that all employees are present in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT COUNT(*) FROM employees;")
        total_employees = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        result_rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(result_rows) == total_employees,
            "All employees are included in the result.",
            f"Expected {total_employees} rows, got {len(result_rows)}. Not all employees are included."
        )

    def test_columns_and_manager_names(self):
        """Check that the correct columns are present and manager names match."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        result_rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(result_rows, columns=columns)
        # Check columns
        has_employee = 'employee_name' in result_df.columns
        has_manager = 'manager_name' in result_df.columns
        # Check manager names
        self.cursor.execute("SELECT e.name AS employee_name, m.name AS manager_name FROM employees e LEFT JOIN employees m ON e.manager_id = m.employee_id;")
        solution_rows = self.cursor.fetchall()
        solution_df = pd.DataFrame(solution_rows, columns=['employee_name', 'manager_name'])
        match = result_df.equals(solution_df)
        _dynamic_test(
            self,
            has_employee and has_manager and match,
            "Result has correct columns and correct manager names.",
            "Either columns are missing or manager names do not match the expected output."
        )

    def test_ceo_has_null_manager(self):
        """Check that the CEO (Alice Johnson) has NULL as manager_name."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        alice_row = df[df['employee_name'] == 'Alice Johnson']
        is_null = pd.isnull(alice_row['manager_name'].values[0])
        _dynamic_test(
            self,
            is_null,
            "CEO (Alice Johnson) has NULL as manager_name.",
            "CEO (Alice Johnson) should have NULL as manager_name but does not."
        )

    def test_manager_names_correctness(self):
        """Check that each employee's manager_name matches the correct manager from employees 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]
        df = pd.DataFrame(rows, columns=columns)
        # Build a dict of employee_id to name
        self.cursor.execute("SELECT employee_id, name, manager_id FROM employees;")
        all_emps = self.cursor.fetchall()
        emp_dict = {row[0]: {'name': row[1], 'manager_id': row[2]} for row in all_emps}
        # For each row, check manager_name matches expected
        errors = []
        for idx, row in df.iterrows():
            emp_name = row['employee_name']
            manager_name = row['manager_name']
            # Find employee_id for this name
            emp_id = next((k for k, v in emp_dict.items() if v['name'] == emp_name), None)
            if emp_id is not None:
                manager_id = emp_dict[emp_id]['manager_id']
                if manager_id is None:
                    if pd.notnull(manager_name):
                        errors.append(f"{emp_name}: Expected NULL manager, got '{manager_name}'")
                else:
                    expected_manager_name = emp_dict[manager_id]['name']
                    if manager_name != expected_manager_name:
                        errors.append(f"{emp_name}: Expected '{expected_manager_name}', got '{manager_name}'")
        _dynamic_test(
            self,
            len(errors) == 0,
            "All manager names are correct for each employee.",
            "Manager name mismatches: " + "; ".join(errors)
        )


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: Manager-Employee Relationships

Lösung