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_only_customers_with_orders(self):
        """Returns only customer names for those who have placed at least one order."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        names = set(r[0] for r in rows)
        expected_names = {"Alice Smith", "Bob Johnson", "Diana King"}
        missing = expected_names - names
        extra = names - expected_names
        condition = (missing == set()) and (extra == set())
        if not condition:
            failure_msg = f"Returned names: {names}. Expected: {expected_names}. Missing: {missing}. Extra: {extra}."
        else:
            failure_msg = ""
        _dynamic_test(self, condition, "Returned only customers with orders.", failure_msg)

    def test_no_customers_without_orders(self):
        """Does not include customers with no orders."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        names = set(r[0] for r in rows)
        not_expected = {"Charlie Lee", "Eve Adams"}
        forbidden = names & not_expected
        condition = len(forbidden) == 0
        if not condition:
            failure_msg = f"These customers should not appear: {forbidden}."
        else:
            failure_msg = ""
        _dynamic_test(self, condition, "No customers without orders are returned.", failure_msg)

    def test_inner_join_used(self):
        """The query uses INNER JOIN between customers and orders."""
        with open("query.sql", "r") as file:
            query = file.read()
        search = normalize_text(query)
        has_inner_join = ("inner join" in search) or ("join" in search and "left" not in search and "right" not in search and "full" not in search and "cross" not in search)
        _dynamic_test(self, has_inner_join, "INNER JOIN is used.", "Query does not use INNER JOIN between customers and orders.")

    def test_unique_customer_names(self):
        """The result includes each customer name only once, even if they have multiple orders."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        names = [r[0] for r in rows]
        unique_names = set(names)
        condition = len(names) == len(unique_names)
        if not condition:
            duplicates = [n for n in unique_names if names.count(n) > 1]
            failure_msg = f"Duplicate customer names found: {duplicates}"
        else:
            failure_msg = ""
        _dynamic_test(self, condition, "All customer names are unique in result.", failure_msg)


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結合の技術を習得します。このモジュールでは、基礎的な結合の概念から高度な複数テーブルのクエリまでを扱い、複数のデータソースからデータを自信を持って分析・統合できるようになります。

実践的なシナリオとハンズオンチャレンジを通じて、基本的なSQL結合タイプ（INNER、LEFT、RIGHT、FULL OUTER JOIN）を学習します。

チャレンジ：注文がある顧客を見つける

解答