import unittest
import psycopg2
import pandas as pd

class TestTask(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()
        # Create country table and insert sample data
        cls.cursor.execute("""
            DROP TABLE IF EXISTS country;
            CREATE TABLE country (
                code VARCHAR(3) PRIMARY KEY,
                name VARCHAR(50),
                continent VARCHAR(30)
            );
        """)
        cls.cursor.executemany(
            "INSERT INTO country (code, name, continent) VALUES (%s, %s, %s);",
            [
                ("USA", "United States", "North America"),
                ("CAN", "Canada", "North America"),
                ("BRA", "Brazil", "South America"),
                ("ARG", "Argentina", "South America"),
                ("FRA", "France", "Europe"),
                ("DEU", "Germany", "Europe"),
                ("AUS", "Australia", "Oceania"),
                ("ZAF", "South Africa", "Africa"),
                ("EGY", "Egypt", "Africa"),
                ("CHN", "China", "Asia"),
                ("JPN", "Japan", "Asia"),
                ("ATA", "Antarctica", "Antarctica")
            ]
        )
        cls.connection.commit()

    def test_query_runs_and_returns_results(self):
        """Check that the user query runs and returns at least one row."""
        try:
            with open("query.sql", "r") as file:
                query = file.read()
            self.cursor.execute(query)
            rows = self.cursor.fetchall()
            condition = len(rows) > 0
            failure_msg = "Query did not return any results."
        except Exception as e:
            condition = False
            failure_msg = f"Query could not be executed: {str(e)}"
        _dynamic_test(
            self,
            condition,
            "Query runs and returns at least one row.",
            failure_msg
        )

    def test_unique_continents_and_order(self):
        """Check that the query returns all unique continents in ascending order."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Reference solution
        solution_query = "SELECT DISTINCT continent FROM country ORDER BY continent;"
        self.cursor.execute(solution_query)
        expected_rows = self.cursor.fetchall()
        expected_continents = [row[0] for row in expected_rows]
        # User query
        self.cursor.execute(query)
        user_rows = self.cursor.fetchall()
        user_continents = [row[0] for row in user_rows]
        # Check all expected continents are present and in order
        condition = user_continents == expected_continents
        if not condition:
            failure_msg = (
                f"Expected continents in ascending order: {expected_continents}, but got: {user_continents}."
            )
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Returned all unique continents in ascending order.",
            failure_msg
        )

    def test_no_duplicates(self):
        """Check that there are no duplicate continents in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        continents = [row[0] for row in rows]
        unique_continents = set(continents)
        condition = len(continents) == len(unique_continents)
        if not condition:
            failure_msg = (
                f"Duplicate continents found in result: {continents}"
            )
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "No duplicate continents in the result.",
            failure_msg
        )

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: Find All Continents and Sort Them in Ascending Order

Solution