import unittest
import psycopg2
import pandas as pd
import re

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_most_popular_genre_and_count(self):
        """Check if query returns the genre(s) with the highest number of borrowings and correct count."""
        with open("query.sql", "r") as file:
            query = file.read()

        # Calculate expected result
        solution_query = '''
            SELECT genre, borrow_count
            FROM (
                SELECT b.genre, COUNT(*) AS borrow_count
                FROM borrowings br
                JOIN books b ON br.book_id = b.book_id
                GROUP BY b.genre
            ) AS genre_counts
            WHERE borrow_count = (
                SELECT MAX(borrow_count)
                FROM (
                    SELECT COUNT(*) AS borrow_count
                    FROM borrowings br2
                    JOIN books b2 ON br2.book_id = b2.book_id
                    GROUP BY b2.genre
                ) AS counts
            );
        '''
        self.cursor.execute(solution_query)
        expected_rows = self.cursor.fetchall()
        expected_columns = [desc[0] for desc in self.cursor.description]
        expected_df = pd.DataFrame(expected_rows, columns=expected_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)

        # Check all expected rows are present
        missing_rows = []
        for _, expected_row in expected_df.iterrows():
            expected_dict = expected_row.to_dict()
            mask = (result_df[list(expected_dict.keys())] == pd.Series(expected_dict)).all(axis=1)
            if not mask.any():
                missing_rows.append(expected_dict)
        condition = len(missing_rows) == 0
        if not condition:
            human_readable = "\\n".join(str(r) for r in missing_rows)
            failure_msg = f"The following expected rows are missing:\\n{human_readable}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Query returns the correct genre(s) and borrow count.",
            failure_msg
        )

    def test_returns_only_genre_and_borrow_count_columns(self):
        """Check that only 'genre' and 'borrow_count' columns are returned."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected_columns = set(["genre", "borrow_count"])
        condition = set(columns) == expected_columns
        _dynamic_test(
            self,
            condition,
            "Query returns only 'genre' and 'borrow_count' columns.",
            f"Returned columns: {columns}. Only 'genre' and 'borrow_count' should be present."
        )

    def test_query_uses_subquery(self):
        """Check that the query uses at least one subquery."""
        with open("query.sql", "r") as file:
            query = file.read().lower()
        # Very basic check for presence of subquery by looking for SELECT inside parentheses
        # Remove comments for accurate check
        query_no_comments = re.sub(r'--.*', '', query)
        has_subquery = bool(re.search(r'\(\s*select\s', query_no_comments))
        _dynamic_test(
            self,
            has_subquery,
            "Query uses at least one subquery.",
            "Query does not use any subqueries."
        )

    def test_returns_all_ties(self):
        """Check that query returns all genres if there is a tie for the highest borrow count."""
        # Add borrowings to create a tie between two genres
        self.cursor.execute("INSERT INTO books (book_id, title, author, genre) VALUES (11, 'Tie Book', 'Author A', 'Science') ON CONFLICT (book_id) DO NOTHING;")
        self.cursor.execute("INSERT INTO borrowings (borrowing_id, member_id, book_id, borrow_date, return_date) VALUES (11, 1, 11, '2023-10-01', NULL) ON CONFLICT (borrowing_id) DO NOTHING;")
        self.connection.commit()
        with open("query.sql", "r") as file:
            query = file.read()
        # Get result after tie
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        genres = [row[0] for row in rows]
        borrow_counts = [row[1] for row in rows]
        # At least two genres should have the max borrow count
        max_count = max(borrow_counts)
        genres_with_max = [g for g, c in zip(genres, borrow_counts) if c == max_count]
        # There should be more than one genre with max count, and all results should have the max count
        condition = len(genres_with_max) > 1 and all(bc == max_count for bc in borrow_counts)
        _dynamic_test(
            self,
            condition,
            "Query returns all genres tied for highest borrow count.",
            f"Returned genres: {genres} with borrow counts: {borrow_counts}. Should include all ties."
        )

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

実際のシナリオを通じて、SQLにおけるサブクエリの記述と応用の技術を習得します。本モジュールでは、サブクエリの理論と実践、SELECT、WHERE、FROM句での利用、相関サブクエリについて解説します。

注文、顧客、商品に焦点を当てたeコマースデータセットを使用してサブクエリを探求します。サブクエリを活用してインサイトを抽出し、ビジネス上の課題を解決する方法を学びます。

チャレンジ：最も人気のジャンル

解答