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()

    def test_columns_and_ordering(self):
        """Check that the query extracts 'region' and 'capital' columns ordered by region, then by capital."""
        with open("query.sql", "r") as file:
            query = file.read()

        # Run user query
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)

        # Check columns
        expected_columns = ["region", "capital"]
        has_columns = list(result_df.columns) == expected_columns
        _dynamic_test(
            self,
            has_columns,
            "Query returns only 'region' and 'capital' columns in correct order.",
            f"Query must return only columns 'region' and 'capital' in order. Found: {list(result_df.columns)}"
        )

        # Check ordering
        is_ordered = result_df.equals(result_df.sort_values(["region", "capital"], kind="mergesort").reset_index(drop=True))
        _dynamic_test(
            self,
            is_ordered,
            "Rows are sorted by 'region' then by 'capital'.",
            "Rows are not sorted by 'region' then by 'capital'."
        )

    def test_all_rows_present(self):
        """Check that all expected (region, capital) pairs from the table are present in the output."""
        with open("query.sql", "r") as file:
            query = file.read()

        # Reference query
        sol_query = """
        SELECT region, capital FROM country
        """
        self.cursor.execute(sol_query)
        sol_rows = self.cursor.fetchall()
        sol_columns = [desc[0] for desc in self.cursor.description]
        sol_df = pd.DataFrame(sol_rows, columns=sol_columns)

        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)

        # Compare sets of tuples
        expected_set = set(tuple(x) for x in sol_df.values)
        result_set = set(tuple(x) for x in result_df.values)
        missing = expected_set - result_set
        extra = result_set - expected_set
        condition = len(missing) == 0 and len(extra) == 0
        fail_msg = ""
        if missing:
            fail_msg += f"Missing rows: {missing}. "
        if extra:
            fail_msg += f"Unexpected extra rows: {extra}."
        _dynamic_test(
            self,
            condition,
            "All expected (region, capital) pairs are present in the result.",
            fail_msg.strip()
        )

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

Learn the core fundamentals of SQL and relational databases. Understand how databases store and organize data, write basic SQL queries, work with tables and perform essential operations. Build a strong foundation for working with databases in backend development.

Challenge: Dual-Sort Query

解答