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 get_query_result(self):
        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]
        return pd.DataFrame(rows, columns=columns)

    def test_welcome_message_present(self):
        """Check that welcome_message column exists for all rows."""
        df = self.get_query_result()
        condition = 'welcome_message' in df.columns and df['welcome_message'].notnull().all()
        _dynamic_test(
            self,
            condition,
            "welcome_message column is present for all rows.",
            "welcome_message column is missing or contains NULL values."
        )

    def test_welcome_back_with_name(self):
        """Check 'Welcome back, [name]!' is returned for years_at_company > 1 and name is not NULL."""
        df = self.get_query_result()
        filtered = df[(df['years_at_company'] > 1) & (df['name'].notnull())]
        expected = [f"Welcome back, {row['name']}!" for _, row in filtered.iterrows()]
        actual = list(filtered['welcome_message'])
        condition = expected == actual
        _dynamic_test(
            self,
            condition,
            "Correct 'Welcome back, [name]!' for years > 1.",
            "Incorrect 'Welcome back, [name]!' for years > 1."
        )

    def test_welcome_with_name(self):
        """Check 'Welcome, [name]!' is returned for years_at_company <= 1 and name is not NULL."""
        df = self.get_query_result()
        filtered = df[(df['years_at_company'] <= 1) & (df['name'].notnull())]
        expected = [f"Welcome, {row['name']}!" for _, row in filtered.iterrows()]
        actual = list(filtered['welcome_message'])
        condition = expected == actual
        _dynamic_test(
            self,
            condition,
            "Correct 'Welcome, [name]!' for years <= 1.",
            "Incorrect 'Welcome, [name]!' for years <= 1."
        )

    def test_welcome_back_there(self):
        """Check 'Welcome back, there!' is returned for years_at_company > 1 and name is NULL."""
        # Insert a NULL name row for testing
        self.cursor.execute("INSERT INTO employees (id, name, department, salary, years_at_company) VALUES (100, NULL, 'Test', 50000, 2) ON CONFLICT DO NOTHING;")
        df = self.get_query_result()
        filtered = df[(df['years_at_company'] > 1) & (df['name'].isnull())]
        expected = ["Welcome back, there!" for _ in filtered.iterrows()]
        actual = list(filtered['welcome_message'])
        condition = expected == actual
        _dynamic_test(
            self,
            condition,
            "Correct 'Welcome back, there!' for years > 1 and name NULL.",
            "Incorrect 'Welcome back, there!' for years > 1 and name NULL."
        )

    def test_welcome_there(self):
        """Check 'Welcome, there!' is returned for years_at_company <= 1 and name is NULL."""
        # Insert a NULL name row for testing
        self.cursor.execute("INSERT INTO employees (id, name, department, salary, years_at_company) VALUES (101, NULL, 'Test', 50000, 1) ON CONFLICT DO NOTHING;")
        df = self.get_query_result()
        filtered = df[(df['years_at_company'] <= 1) & (df['name'].isnull())]
        expected = ["Welcome, there!" for _ in filtered.iterrows()]
        actual = list(filtered['welcome_message'])
        condition = expected == actual
        _dynamic_test(
            self,
            condition,
            "Correct 'Welcome, there!' for years <= 1 and name NULL.",
            "Incorrect 'Welcome, there!' for years <= 1 and name NULL."
        )

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: Custom Welcome Messages

解答