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

A comprehensive course designed to help you master the CASE WHEN statement in SQL, from basic conditional logic to advanced use cases in data analysis and reporting. Through engaging theory and real-world tasks, you'll learn to write expressive, powerful queries that adapt to complex business requirements.

Learn the basics of the CASE WHEN statement, its syntax, and how to use it for simple conditional logic in SQL queries.

Delve deeper into complex CASE WHEN expressions, including nested conditions and multiple criteria.

Apply CASE WHEN in WHERE and ORDER BY clauses to filter and sort data conditionally.

Combine CASE WHEN with aggregate functions for advanced reporting and analytics.

Challenge: Categorize Sales Volume

Soluzione

Challenge: Categorize Sales Volume

Soluzione