import unittest
import pandas as pd
import psycopg2

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 test_returns_two_rows(self):
        """Check if exactly 2 rows are returned."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == 2,
            "Query returns exactly 2 rows.",
            f"Query returned {len(rows)} rows instead of 2."
        )

    def test_columns_present(self):
        """Check if each row contains member_id, name, and borrow_count columns."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected_cols = {"member_id", "name", "borrow_count"}
        _dynamic_test(
            self,
            expected_cols.issubset(set(columns)),
            "All required columns are present.",
            f"Missing columns: {expected_cols - set(columns)}."
        )

    def test_correct_top_members(self):
        """Check if the returned members are those with the highest borrow count."""
        # Compute borrow counts for each member
        solution_query = """
            SELECT m.member_id, m.name, borrow_count
            FROM (
                SELECT member_id, COUNT(*) AS borrow_count
                FROM borrowings
                GROUP BY member_id
            ) AS member_borrow_counts
            JOIN members m ON m.member_id = member_borrow_counts.member_id
            ORDER BY borrow_count DESC, m.member_id ASC
            LIMIT 2;
        """
        self.cursor.execute(solution_query)
        expected_rows = self.cursor.fetchall()
        expected_df = pd.DataFrame(expected_rows, columns=["member_id", "name", "borrow_count"])

        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        user_rows = self.cursor.fetchall()
        user_df = pd.DataFrame(user_rows, columns=[desc[0] for desc in self.cursor.description])

        # Compare dataframes (ignore column order)
        merged = pd.merge(expected_df, user_df, on=["member_id", "name", "borrow_count"], how="outer", indicator=True)
        missing = merged[merged["_merge"] != "both"]
        condition = missing.empty and len(user_df) == 2
        if not condition:
            failure_msg = f"Returned rows do not match expected top 2 members. Difference:\\n{missing}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Returned members are the correct top 2 by borrow count.",
            failure_msg
        )

    def test_ordering(self):
        """Check if results are ordered by borrow_count DESC, member_id ASC."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        df = pd.DataFrame(rows, columns=[desc[0] for desc in self.cursor.description])
        # Check ordering
        correct_order = df.sort_values(by=["borrow_count", "member_id"], ascending=[False, True])
        condition = df.reset_index(drop=True).equals(correct_order.reset_index(drop=True))
        _dynamic_test(
            self,
            condition,
            "Rows are ordered by borrow_count DESC, member_id ASC.",
            "Rows are not ordered by borrow_count DESC, member_id ASC."
        )


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

Master the art of writing and applying subqueries in SQL through real-world scenarios. This module covers the theory and practice of subqueries, including their use in SELECT, WHERE, and FROM clauses, as well as correlated subqueries. 

Explore subqueries using an e-commerce dataset, focusing on orders, customers, and products. Learn to extract insights and solve business problems using subqueries.

Challenge: Most Active Members

Lösung