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 the country table
        cls.cursor.execute("DROP TABLE IF EXISTS country;")
        cls.cursor.execute("""
            CREATE TABLE country (
                name VARCHAR(100),
                capital VARCHAR(100),
                continent VARCHAR(100)
            );
        """)
        # Insert sample data: European and non-European countries
        data = [
            ("France", "Paris", "Europe"),
            ("Germany", "Berlin", "Europe"),
            ("Spain", "Madrid", "Europe"),
            ("Italy", "Rome", "Europe"),
            ("Japan", "Tokyo", "Asia"),
            ("Mexico", "Mexico City", "North America"),
            ("United States", "Washington", "North America"),
            ("Thailand", "Bangkok", "Asia"),
            ("Paraguay", "Asunción", "South America"),
            ("Philippines", "Manila", "Asia"),
            ("New Zealand", "Wellington", "Oceania")
        ]
        cls.cursor.executemany(
            "INSERT INTO country (name, capital, continent) VALUES (%s, %s, %s);",
            data
        )
        cls.connection.commit()

    def get_query_result(self):
        with open("query.sql", "r") as f:
            query = f.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_columns_present_and_order(self):
        """Check that the output columns are exactly name, capital, continent in the correct order."""
        df = self.get_query_result()
        expected_columns = ['name', 'capital', 'continent']
        condition = list(df.columns) == expected_columns
        _dynamic_test(
            self,
            condition,
            "The output columns are correct and in the right order.",
            f"Expected columns {expected_columns}, but got {list(df.columns)}."
        )

    def test_no_european_countries(self):
        """Ensure that no row in the result has continent='Europe'."""
        df = self.get_query_result()
        has_europe = (df['continent'] == 'Europe').any()
        _dynamic_test(
            self,
            not has_europe,
            "No European countries are present in the result.",
            "There are countries from Europe in your result, but they should be excluded."
        )

    def test_all_non_european_countries_included(self):
        """Check that all non-European countries are included in the result."""
        # Reference data - must match the init script data for non-European countries
        all_countries = [
            {"name": "Japan", "capital": "Tokyo", "continent": "Asia"},
            {"name": "Mexico", "capital": "Mexico City", "continent": "North America"},
            {"name": "United States", "capital": "Washington", "continent": "North America"},
            {"name": "Thailand", "capital": "Bangkok", "continent": "Asia"},
            {"name": "Paraguay", "capital": "Asunción", "continent": "South America"},
            {"name": "Philippines", "capital": "Manila", "continent": "Asia"},
            {"name": "New Zealand", "capital": "Wellington", "continent": "Oceania"}
        ]
        df = self.get_query_result().copy()
        for col in ['name', 'capital', 'continent']:
            df[col] = df[col].astype(str).str.strip()
        missing = []
        for country in all_countries:
            mask = (
                (df['name'] == country['name']) &
                (df['capital'] == country['capital']) &
                (df['continent'] == country['continent'])
            )
            if not mask.any():
                missing.append(country)
        condition = len(missing) == 0
        failure_msg = f"Missing non-European countries: {missing}" if missing else ""
        _dynamic_test(
            self,
            condition,
            "All non-European countries are included in the result.",
            failure_msg
        )

    def test_sorted_by_capital(self):
        """Ensure that the result is sorted by capital (ascending)."""
        df = self.get_query_result()
        capitals = df['capital'].astype(str).tolist()
        condition = capitals == sorted(capitals)
        _dynamic_test(
            self,
            condition,
            "The result is sorted by capital in ascending order.",
            f"The capitals are not sorted: {capitals}"
        )

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):
    import re
    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: Non-Europe Capitals Sort

Oplossing