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()
        # Drop and recreate country table
        cls.cursor.execute("""
            DROP TABLE IF EXISTS country;
            CREATE TABLE country (
                code CHAR(3) PRIMARY KEY,
                name VARCHAR(52) NOT NULL,
                continent VARCHAR(50),
                region VARCHAR(50) NOT NULL
            );
        """)
        # Insert test data
        cls.cursor.executemany(
            "INSERT INTO country (code, name, continent, region) VALUES (%s, %s, %s, %s);",
            [
                ("BLZ", "Belize", "North America", "Central America"),
                ("CAN", "Canada", "North America", "North America"),
                ("CRI", "Costa Rica", "North America", "Central America"),
                ("SLV", "El Salvador", "North America", "Central America"),
                ("GTM", "Guatemala", "North America", "Central America"),
                ("HND", "Honduras", "North America", "Central America"),
                ("MEX", "Mexico", "North America", "North America"),
                ("NIC", "Nicaragua", "North America", "Central America"),
                ("PAN", "Panama", "North America", "Central America"),
                ("USA", "United States", "North America", "North America"),
                ("BRA", "Brazil", "South America", "South America"),
                ("ARG", "Argentina", "South America", "South America"),
                ("FRA", "France", "Europe", "Western Europe"),
                ("CHN", "China", "Asia", "Eastern Asia")
            ]
        )
        cls.connection.commit()

    def test_selects_name_column(self):
        """Check if the query selects only the 'name' column."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        condition = columns == ["name"]
        _dynamic_test(
            self,
            condition,
            "Query selects only the 'name' column.",
            f"Query should select only the 'name' column but got: {columns}"
        )

    def test_correct_rows_returned(self):
        """Check if query returns only countries from 'Central America' or 'North America' regions."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_names = sorted([r[0].strip() for r in rows])
        expected_names = sorted([
            'Belize', 'Canada', 'Costa Rica', 'El Salvador', 'Guatemala',
            'Honduras', 'Mexico', 'Nicaragua', 'Panama', 'United States'
        ])
        _dynamic_test(
            self,
            result_names == expected_names,
            "Query returns only countries from 'Central America' or 'North America'.",
            f"Expected names: {expected_names}, got: {result_names}"
        )

    def test_or_operator_used(self):
        """Check that both 'Central America' and 'North America' regions are included in results and 'OR' is used."""
        with open("query.sql", "r") as file:
            query = file.read().lower().replace("\\n", " ")
        has_or = " or " in query
        has_central = "central america" in query
        has_north = "north america" in query
        condition = has_or and has_central and has_north
        _dynamic_test(
            self,
            condition,
            "Query uses OR operator and includes both regions.",
            "Query must use OR operator and include both 'Central America' and 'North America' in WHERE clause."
        )

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

Gain a solid understanding of SQL essentials tailored specifically for business intelligence. You will master querying, filtering, and aggregating business datasets through hands-on tasks designed to mirror real-world professional scenarios.

Challenge: America Regions Filter

解答