import unittest
import sqlite3
import re


class TestTask(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.connection = sqlite3.connect(":memory:")
        cls.cursor = cls.connection.cursor()
        # Создаем таблицу sales и вставляем тестовые данные
        cls.cursor.execute("""
        CREATE TABLE sales (
            sale_id INTEGER PRIMARY KEY,
            product TEXT,
            quantity INTEGER
        );
        """)
        sales_data = [
            (1, "Product A", 5),    # Small
            (2, "Product B", 10),   # Small
            (3, "Product C", 20),   # Standard
            (4, "Product D", 50),   # Standard
            (5, "Product E", 100),  # Bulk
            (6, "Product F", 150)   # Bulk
        ]
        cls.cursor.executemany("INSERT INTO sales VALUES (?, ?, ?);", sales_data)
        cls.connection.commit()

    def test_volume_category_column_exists(self):
        """Check if the result includes 'volume_category' column."""
        with open("query.sql", "r") as file:
            query = file.read()
        try:
            self.cursor.execute(query)
            rows = self.cursor.fetchall()
            columns = [desc[0] for desc in self.cursor.description]
            condition = ("volume_category" in columns)
            success_msg = "Result set includes 'volume_category' column."
            failure_msg = f"'volume_category' column is missing. Got columns: {columns}"
        except Exception as e:
            condition = False
            success_msg = ""
            failure_msg = f"Query failed to execute: {e}"
        _dynamic_test(self, condition, success_msg, failure_msg)

    def test_volume_category_values(self):
        """Check if 'volume_category' values match quantity ranges."""
        with open("query.sql", "r") as file:
            query = file.read()
        try:
            self.cursor.execute(query)
            rows = self.cursor.fetchall()
            columns = [desc[0] for desc in self.cursor.description]

            if "quantity" not in columns or "volume_category" not in columns:
                condition = False
                success_msg = ""
                failure_msg = f"Result must contain 'quantity' and 'volume_category' columns. Got: {columns}"
            else:
                q_idx = columns.index("quantity")
                v_idx = columns.index("volume_category")

                misclassified = []
                for row in rows:
                    quantity = row[q_idx]
                    label = row[v_idx]
                    if quantity >= 100:
                        expected = "Bulk"
                    elif quantity >= 20:
                        expected = "Standard"
                    else:
                        expected = "Small"
                    if label != expected:
                        misclassified.append((quantity, label, expected))

                condition = (len(misclassified) == 0)
                success_msg = "All 'volume_category' values match the quantity ranges."
                failure_msg = (
                    f"Misclassified rows (quantity, got, expected): {misclassified}"
                    if misclassified else
                    "Unexpected error in classification."
                )
        except Exception as e:
            condition = False
            success_msg = ""
            failure_msg = f"Query failed to execute: {e}"
        _dynamic_test(self, condition, success_msg, failure_msg)

    def test_row_count_matches(self):
        """Check if the number of rows in the result matches the number of rows in 'sales' table."""
        with open("query.sql", "r") as file:
            query = file.read()
        try:
            self.cursor.execute("SELECT COUNT(*) FROM sales;")
            expected_count = self.cursor.fetchone()[0]

            self.cursor.execute(query)
            rows = self.cursor.fetchall()
            condition = (len(rows) == expected_count)
            success_msg = "Number of rows in the result matches the number of rows in 'sales' table."
            failure_msg = f"Expected {expected_count} rows, but got {len(rows)}."
        except Exception as e:
            condition = False
            success_msg = ""
            failure_msg = f"Query failed to execute: {e}"
        _dynamic_test(self, condition, success_msg, failure_msg)

    def test_all_original_columns_present(self):
        """Check if all original columns from 'sales' are present in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        try:
            # Оригинальные колонки таблицы sales
            self.cursor.execute("SELECT * FROM sales LIMIT 1;")
            orig_columns = [desc[0] for desc in self.cursor.description]

            # Колонки в результате query.sql
            self.cursor.execute(query)
            rows = self.cursor.fetchall()
            result_columns = [desc[0] for desc in self.cursor.description]

            missing = [c for c in orig_columns if c not in result_columns]
            condition = (len(missing) == 0)
            success_msg = "All original columns from 'sales' are present in the result."
            failure_msg = f"The following original column(s) are missing in result: {missing}"
        except Exception as e:
            condition = False
            success_msg = ""
            failure_msg = f"Query failed to execute: {e}"
        _dynamic_test(self, condition, success_msg, 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におけるCASE WHEN文を基礎的な条件分岐からデータ分析・レポーティングでの高度な活用例まで習得できる包括的なモジュールです。理論と実践的な課題を通じて、複雑なビジネス要件に対応する表現力豊かなクエリの作成方法を学びます。

CASE WHEN文の基本、構文、およびSQLクエリでの単純な条件付きロジックへの活用方法を学習します。

チャレンジ：販売量の分類

解答