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_joins_on_customer_id(self):
        """Check if the query joins sales and customers on customer_id."""
        with open("query.sql", "r") as file:
            query = file.read()
        join_condition = "join customers c on s.customer_id = c.customer_id"
        join_condition_alt = "join customers c on (s.customer_id = c.customer_id)"
        normalized_query = normalize_text(query)
        condition = join_condition in normalized_query or join_condition_alt in normalized_query
        _dynamic_test(
            self,
            condition,
            "The query joins sales and customers on customer_id.",
            "The query does not join sales and customers on customer_id."
        )

    def test_groups_by_city(self):
        """Check if the query groups results by city."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized_query = normalize_text(query)
        condition = "group by c.city" in normalized_query or "group by city" in normalized_query
        _dynamic_test(
            self,
            condition,
            "The query groups results by city.",
            "The query does not group results by city."
        )

    def test_returns_cities_with_sales(self):
        """Check that the query returns each city present in customers table that has at least one sale."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        result_rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(result_rows, columns=columns)
        self.cursor.execute("""
            SELECT DISTINCT c.city
            FROM sales s
            JOIN customers c ON s.customer_id = c.customer_id
        """)
        expected_cities = set(row[0] for row in self.cursor.fetchall())
        result_cities = set(result_df['city'])
        missing = expected_cities - result_cities
        condition = len(missing) == 0
        _dynamic_test(
            self,
            condition,
            "The query returns all cities with at least one sale.",
            f"The following cities with sales are missing: {', '.join(missing)}" if missing else ""
        )

    def test_sales_count_column(self):
        """Check that the count of sales column is named sales_count and is correct."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        result_rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(result_rows, columns=columns)
        condition = 'sales_count' in result_df.columns
        _dynamic_test(
            self,
            condition,
            "The column with sales count is named sales_count.",
            "The column with sales count is not named sales_count."
        )
        # Now check correctness of sales_count values
        self.cursor.execute("""
            SELECT c.city, COUNT(s.sale_id) AS expected_count
            FROM sales s
            JOIN customers c ON s.customer_id = c.customer_id
            GROUP BY c.city
        """)
        expected = pd.DataFrame(self.cursor.fetchall(), columns=['city', 'expected_count'])
        merged = pd.merge(result_df, expected, on='city', how='outer', suffixes=('', '_expected'))
        merged['match'] = merged['sales_count'] == merged['expected_count']
        mismatch = merged[~merged['match']]
        condition = mismatch.empty
        fail_msg = "Sales count does not match expected for: " + ", ".join(mismatch['city']) if not mismatch.empty else ""
        _dynamic_test(
            self,
            condition,
            "The sales_count values are correct for each city.",
            fail_msg
        )

    def test_sales_count_uses_sale_id(self):
        """Check that the query counts sales using sale_id."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized_query = normalize_text(query)
        condition = "count(s.sale_id)" in normalized_query or "count(*)" in normalized_query
        _dynamic_test(
            self,
            condition,
            "The query counts sales using sale_id.",
            "The query does not count sales using sale_id."
        )

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を用いたデータのクエリ、変換、分析を習得します。

リレーショナルデータベースとSQLの基礎概念を学習します。テーブルからデータを取得、フィルタリング、並べ替える方法を探求し、より高度なデータ分析の基礎を築きます。

SQLの強力な集計機能を深く学びます。データを要約し、グループ化し、より深い洞察のために分析する方法を習得します。

複数のテーブルからデータを組み合わせる力を引き出します。さまざまな種類の結合を実行し、複雑なビジネス上の質問に答える方法を学びます。

サブクエリ、ウィンドウ関数、データ変換などを含む高度なSQL技術を習得し、強力なデータ分析を実現します。

チャレンジ：都市別売上分析

解答