import unittest
import psycopg2
import pandas as pd
from datetime import date
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 setUp(self):
        # Reset tables for testing
        self.cursor.execute("DELETE FROM borrowings;")
        self.cursor.execute("DELETE FROM books;")
        self.cursor.execute("""
            INSERT INTO books (book_id, title, author, genre) VALUES
            (1, 'To Kill a Mockingbird', 'Harper Lee', 'Fiction'),
            (2, '1984', 'George Orwell', 'Dystopian'),
            (3, 'The Great Gatsby', 'F. Scott Fitzgerald', 'Classic'),
            (4, 'A Brief History of Time', 'Stephen Hawking', 'Science'),
            (5, 'The Hobbit', 'J.R.R. Tolkien', 'Fantasy'),
            (6, 'Pride and Prejudice', 'Jane Austen', 'Romance'),
            (7, 'The Catcher in the Rye', 'J.D. Salinger', 'Fiction'),
            (8, 'The Da Vinci Code', 'Dan Brown', 'Thriller'),
            (9, 'The Alchemist', 'Paulo Coelho', 'Adventure'),
            (10, 'The Art of War', 'Sun Tzu', 'Philosophy');
        """)
        self.cursor.execute("DELETE FROM members;")
        self.cursor.execute("""
            INSERT INTO members (member_id, name, join_date) VALUES
            (1, 'Alice Johnson', '2022-01-15'),
            (2, 'Bob Smith', '2022-03-22'),
            (3, 'Charlie Lee', '2022-05-10'),
            (4, 'Diana Evans', '2022-07-05'),
            (5, 'Ethan Brown', '2022-09-18');
        """)
        self.connection.commit()

    def _insert_borrowings(self, borrowings):
        for i, (member_id, book_id, borrow_date, return_date) in enumerate(borrowings, 1):
            if return_date is not None:
                self.cursor.execute(
                    """
                    INSERT INTO borrowings
                    (borrowing_id, member_id, book_id, borrow_date, return_date)
                    VALUES (%s, %s, %s, %s, %s)
                    """,
                    (i, member_id, book_id, borrow_date, return_date)
                )
            else:
                self.cursor.execute(
                    """
                    INSERT INTO borrowings
                    (borrowing_id, member_id, book_id, borrow_date, return_date)
                    VALUES (%s, %s, %s, %s, NULL)
                    """,
                    (i, member_id, book_id, borrow_date)
                )
        self.connection.commit()

    def test_returns_books_borrowed_on_or_after_cutoff(self):
        """Returns all books borrowed on or after July 1, 2023."""
        cutoff = date(2023, 7, 1)
        # Book 2 borrowed after cutoff, Book 5 borrowed before cutoff
        borrowings = [
            (1, 2, date(2023, 7, 10), date(2023, 7, 20)),  # on/after cutoff
            (2, 5, date(2023, 6, 15), date(2023, 6, 25)),  # before cutoff
        ]
        self._insert_borrowings(borrowings)

        with open("query.sql", "r") as file:
            query = file.read()

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

        condition = (2 in df["book_id"].values) and (5 not in df["book_id"].values)
        _dynamic_test(
            self,
            condition,
            "Returns only books borrowed on or after July 1, 2023.",
            "Did not return correct books for borrowings on or after July 1, 2023.",
        )

    def test_does_not_return_books_before_cutoff(self):
        """Does not return books borrowed only before July 1, 2023."""
        borrowings = [
            (1, 1, date(2023, 6, 1), date(2023, 6, 10)),
            (2, 2, date(2023, 5, 20), date(2023, 5, 30)),
        ]
        self._insert_borrowings(borrowings)

        with open("query.sql", "r") as file:
            query = file.read()

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

        condition = df.empty
        _dynamic_test(
            self,
            condition,
            "No books returned if all borrowings are before July 1, 2023.",
            "Books were incorrectly returned for borrowings before July 1, 2023.",
        )

    def test_uses_subquery(self):
        """Uses a subquery to filter books based on borrowings on or after July 1, 2023."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized = normalize_text(query)
        has_subquery = "select" in normalized and "in (" in normalized
        _dynamic_test(
            self,
            has_subquery,
            "Query uses a subquery.",
            "Query does not use a subquery.",
        )

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コマースデータセットを使用してサブクエリを探求します。サブクエリを活用してインサイトを抽出し、ビジネス上の課題を解決する方法を学びます。

チャレンジ：先月に借りられた本

解答