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_only_null_manager_id(self):
        """Check that only employees with null manager_id are returned."""
        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)
        condition = df['manager_id'].isnull().all() and len(df) > 0
        _dynamic_test(
            self,
            condition,
            "All returned employees have null manager_id.",
            "Some returned employees have a non-null manager_id, or result is empty."
        )

    def test_all_columns_included(self):
        """Check that all columns from employees table are included."""
        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", "name", "department", "manager_id", "bonus"]
        condition = set(expected_columns) == set(columns)
        _dynamic_test(
            self,
            condition,
            "All columns from employees table are included.",
            f"Missing or extra columns in result: {set(columns) ^ set(expected_columns)}"
        )

    def test_returns_correct_employees(self):
        """Check that correct employees are returned when manager_id is null."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Get expected employees
        self.cursor.execute("SELECT employee_id FROM employees WHERE manager_id IS NULL")
        expected_ids = set(row[0] for row in self.cursor.fetchall())
        # Run user query
        self.cursor.execute(query)
        result_ids = set(row[0] for row in self.cursor.fetchall())
        condition = result_ids == expected_ids
        _dynamic_test(
            self,
            condition,
            "Returned employees match those with null manager_id.",
            f"Returned employee IDs do not match: {expected_ids ^ result_ids}"
        )

    def test_works_with_additional_nulls(self):
        """Check that query works if more employees with null manager_id are added."""
        # Add a test employee with null manager_id
        self.cursor.execute("""
            INSERT INTO employees (employee_id, name, department, manager_id, bonus)
            VALUES (999, 'Test User', 'Test', NULL, 999.99)
        """)
        self.connection.commit()
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        employee_ids = [row[0] for row in rows]
        condition = 999 in employee_ids
        _dynamic_test(
            self,
            condition,
            "Query includes newly added employee with null manager_id.",
            "Query does not include newly added employee with null manager_id."
        )
        # Clean up
        self.cursor.execute("DELETE FROM employees WHERE employee_id = 999")
        self.connection.commit()
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 use of built-in SQL functions to manipulate strings, dates, numbers, logical values, nulls, and JSON data. This course provides a hands-on approach to understanding and applying SQL functions in real-world scenarios, enhancing your data querying and transformation skills.

Explore SQL's powerful string functions for manipulating and analyzing textual data in databases.

Master SQL's date and time functions to work with temporal data effectively.

Learn to use SQL's numeric functions for calculations and data analysis.

Harness SQL's logical functions to control query flow and make decisions based on data.

Master handling of NULL values in SQL using specialized functions.

Learn to work with JSON data in SQL using specialized functions for querying and manipulation.

Challenge: Find Employees Without Managers

Solution