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_columns_present(self):
        """Check if all required columns including salary_band are present in query result."""
        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 = [
            "id", "name", "department", "salary", "years_at_company", "salary_band"
        ]
        missing_columns = [col for col in expected_columns if col not in columns]
        _dynamic_test(
            self,
            not missing_columns,
            "All required columns including salary_band are present.",
            f"Missing columns: {', '.join(missing_columns)}"
        )

    def test_senior_band(self):
        """Check if salary_band is 'Senior' for salary >= 90000."""
        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)
        senior_rows = df[df['salary'] >= 90000]
        incorrect = senior_rows[senior_rows['salary_band'] != 'Senior']
        _dynamic_test(
            self,
            incorrect.empty,
            "All employees with salary >= 90000 are labeled 'Senior'.",
            f"Incorrect salary_band for ids: {incorrect['id'].tolist()} (expected 'Senior')"
        )

    def test_mid_band(self):
        """Check if salary_band is 'Mid' for 60000 <= salary < 90000."""
        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)
        mid_rows = df[(df['salary'] >= 60000) & (df['salary'] < 90000)]
        incorrect = mid_rows[mid_rows['salary_band'] != 'Mid']
        _dynamic_test(
            self,
            incorrect.empty,
            "All employees with 60000 <= salary < 90000 are labeled 'Mid'.",
            f"Incorrect salary_band for ids: {incorrect['id'].tolist()} (expected 'Mid')"
        )

    def test_junior_band(self):
        """Check if salary_band is 'Junior' for salary < 60000."""
        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)
        junior_rows = df[df['salary'] < 60000]
        incorrect = junior_rows[junior_rows['salary_band'] != 'Junior']
        _dynamic_test(
            self,
            incorrect.empty,
            "All employees with salary < 60000 are labeled 'Junior'.",
            f"Incorrect salary_band for ids: {incorrect['id'].tolist()} (expected 'Junior')"
        )

    def test_salary_band_for_all_rows(self):
        """Check that salary_band is not NULL for any employee."""
        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)
        null_bands = df[df['salary_band'].isnull()]
        _dynamic_test(
            self,
            null_bands.empty,
            "salary_band is assigned for every employee.",
            f"salary_band is NULL for ids: {null_bands['id'].tolist()}"
        )

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

A comprehensive course designed to help you master the CASE WHEN statement in SQL, from basic conditional logic to advanced use cases in data analysis and reporting. Through engaging theory and real-world tasks, you'll learn to write expressive, powerful queries that adapt to complex business requirements.

Learn the basics of the CASE WHEN statement, its syntax, and how to use it for simple conditional logic in SQL queries.

Delve deeper into complex CASE WHEN expressions, including nested conditions and multiple criteria.

Apply CASE WHEN in WHERE and ORDER BY clauses to filter and sort data conditionally.

Combine CASE WHEN with aggregate functions for advanced reporting and analytics.

Challenge: Departmental Salary Bands

解答