import unittest
import psycopg2
import pandas as pd
from datetime import date

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 get_cutoff_date(self):
        self.cursor.execute("SELECT DATE '2021-01-02' - INTERVAL '6 months'")
        return self.cursor.fetchone()[0]

    def test_returns_recent_employees(self):
        """Check that only employees hired on or after the fixed cutoff date are returned."""
        with open("query.sql", "r") as file:
            query = file.read()
        cutoff_date = self.get_cutoff_date()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        if not result_df.empty:
            result_df['hire_date'] = pd.to_datetime(result_df['hire_date'])
        all_recent = result_df['hire_date'].ge(pd.to_datetime(cutoff_date)).all() if not result_df.empty else True
        _dynamic_test(
            self,
            all_recent,
            "All returned employees were hired on or after the cutoff date.",
            "Some returned employees were hired before the cutoff date."
        )

    def test_no_old_employees(self):
        """Check that no employees hired before cutoff are included."""
        with open("query.sql", "r") as file:
            query = file.read()
        cutoff_date = self.get_cutoff_date()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        if not result_df.empty:
            result_df['hire_date'] = pd.to_datetime(result_df['hire_date'])
        no_old = (~result_df['hire_date'].lt(pd.to_datetime(cutoff_date))).all() if not result_df.empty else True
        _dynamic_test(
            self,
            no_old,
            "No employees hired before the cutoff date are included.",
            "Some employees hired before the cutoff date are included."
        )

    def test_uses_subquery(self):
        """Check that a subquery is used in the WHERE clause."""
        with open("query.sql", "r") as file:
            query = file.read()
        uses_subquery = 'where' in query.lower() and '(' in query and 'select' in query.lower()
        _dynamic_test(
            self,
            uses_subquery,
            "Query uses a subquery in the WHERE clause.",
            "Query does not use a subquery in the WHERE clause."
        )

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

チャレンジ：最近採用された従業員

解答