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_column_name(self):
        """Check if the result column is named total_inventory_value."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        _dynamic_test(
            self,
            'total_inventory_value' in [normalize_text(c) for c in columns],
            "The result column is named total_inventory_value.",
            "The result column must be named total_inventory_value."
        )

    def test_single_row(self):
        """Check if the query returns a single row."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == 1,
            "Query returns a single row.",
            f"Query should return a single row, but got {len(rows)} rows."
        )

    def test_total_inventory_value(self):
        """Check if the total inventory value is correctly calculated as sum(price * stock) for all products."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Calculate expected value
        self.cursor.execute("SELECT SUM(price * stock) FROM products;")
        expected = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        result = self.cursor.fetchone()[0]
        _dynamic_test(
            self,
            result == expected,
            "Total inventory value is correct.",
            f"Total inventory value is incorrect. Expected {expected}, got {result}."
        )

    def test_uses_sum_and_multiplication(self):
        """Check that the query uses SUM and multiplies price and stock."""
        with open("query.sql", "r") as file:
            query = file.read().lower().replace(' ', '')
        has_sum = 'sum(' in query
        has_multiplication = ('price*stock' in query) or ('stock*price' in query)
        _dynamic_test(
            self,
            has_sum and has_multiplication,
            "Query uses SUM and multiplies price and stock.",
            "Query must use SUM and multiply price and stock."
        )

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

This beginner course focuses on the essential SQL functions used for data cleaning and transformation. Learners will practice handling missing values, formatting and standardizing data, and applying string, date, numeric, logical, and null handling functions to prepare datasets for analysis. The course provides hands-on challenges directly related to the most common data preparation tasks in SQL.

Challenge: Calculate Inventory Value

解答