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

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 Inventory Value

Solution

Challenge: Calculate Inventory Value

Solution