import unittest
import psycopg2
import pandas as pd

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_all_products_returned(self):
        """Check that all products are returned by the query."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        self.cursor.execute("SELECT COUNT(*) FROM products;")
        expected_count = self.cursor.fetchone()[0]
        _dynamic_test(
            self,
            len(rows) == expected_count,
            "All products are returned.",
            f"Expected {expected_count} products, got {len(rows)}."
        )

    def test_discounted_price_column_present(self):
        """Check that the discounted_price column exists in the result."""
        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,
            "discounted_price" in columns,
            "discounted_price column exists.",
            "discounted_price column is missing."
        )

    def test_discounted_price_calculation(self):
        """Check that discounted_price is calculated as price - discount for each product."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        self.cursor.execute("SELECT product_id, price, discount FROM products;")
        prod_rows = self.cursor.fetchall()
        for product_id, price, discount in prod_rows:
            expected_value = round(float(price) - float(discount), 2)
            actual_value = df[df["product_id"] == product_id]["discounted_price"].iloc[0]
            condition = abs(float(actual_value) - expected_value) < 0.01
            _dynamic_test(
                self,
                condition,
                f"discounted_price correct for product_id {product_id}.",
                f"discounted_price incorrect for product_id {product_id}: expected {expected_value}, got {actual_value}."
            )

    def test_discounted_price_rounded(self):
        """Check that discounted_price is rounded to two decimal places for all products."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        not_rounded = []
        for idx, v in df["discounted_price"].items():
            # Check if value has at most two decimal places
            str_v = f"{v:.2f}"
            if "." in str_v:
                decimals = str_v.split(".")[1]
                if len(decimals) != 2:
                    not_rounded.append((idx, v))
        condition = len(not_rounded) == 0
        _dynamic_test(
            self,
            condition,
            "All discounted_price values are rounded to two decimal places.",
            f"Some discounted_price values are not rounded to two decimals: {not_rounded}"
        )

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

Master the use of built-in SQL functions to manipulate strings, dates, numbers, logical values, nulls, and JSON data. This course provides a hands-on approach to understanding and applying SQL functions in real-world scenarios, enhancing your data querying and transformation skills.

Explore SQL's powerful string functions for manipulating and analyzing textual data in databases.

Master SQL's date and time functions to work with temporal data effectively.

Learn to use SQL's numeric functions for calculations and data analysis.

Harness SQL's logical functions to control query flow and make decisions based on data.

Master handling of NULL values in SQL using specialized functions.

Learn to work with JSON data in SQL using specialized functions for querying and manipulation.

Challenge: Calculate Discounted Prices

Solution

Challenge: Calculate Discounted Prices

Solution