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

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: Books Borrowed in the Last Month

Рішення