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_single_row_three_columns(self):
        """Check that result has one row and three columns."""
        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]
        condition = len(rows) == 1 and len(columns) == 3
        _dynamic_test(
            self,
            condition,
            "Query returns a single row with three columns.",
            f"Expected 1 row and 3 columns, got {len(rows)} rows and {len(columns)} columns."
        )

    def test_column_names(self):
        """Check that columns are named min_price, max_price, avg_price."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [normalize_text(desc[0]) for desc in self.cursor.description]
        expected = ["min_price", "max_price", "avg_price"]
        missing = [col for col in expected if col not in columns]
        condition = len(missing) == 0
        _dynamic_test(
            self,
            condition,
            "Returned columns are named min_price, max_price, and avg_price.",
            f"Missing columns: {', '.join(missing)}."
        )

    def test_min_price_value(self):
        """Check that min_price matches the lowest price in products."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        row = self.cursor.fetchone()
        col_idx = [desc[0] for desc in self.cursor.description].index("min_price")
        min_price = row[col_idx]
        self.cursor.execute("SELECT MIN(price) FROM products;")
        expected = self.cursor.fetchone()[0]
        condition = float(min_price) == float(expected)
        _dynamic_test(
            self,
            condition,
            "min_price value is correct.",
            f"min_price should be {expected}, got {min_price}."
        )

    def test_max_price_value(self):
        """Check that max_price matches the highest price in products."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        row = self.cursor.fetchone()
        col_idx = [desc[0] for desc in self.cursor.description].index("max_price")
        max_price = row[col_idx]
        self.cursor.execute("SELECT MAX(price) FROM products;")
        expected = self.cursor.fetchone()[0]
        condition = float(max_price) == float(expected)
        _dynamic_test(
            self,
            condition,
            "max_price value is correct.",
            f"max_price should be {expected}, got {max_price}."
        )

    def test_avg_price_value(self):
        """Check that avg_price matches the average price in products."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        row = self.cursor.fetchone()
        col_idx = [desc[0] for desc in self.cursor.description].index("avg_price")
        avg_price = row[col_idx]
        self.cursor.execute("SELECT AVG(price) FROM products;")
        expected = self.cursor.fetchone()[0]
        # Compare rounded to 2 decimals for floating point tolerance
        condition = round(float(avg_price), 2) == round(float(expected), 2)
        _dynamic_test(
            self,
            condition,
            "avg_price value is correct (rounded to 2 decimals).",
            f"avg_price should be {expected}, got {avg_price}."
        )

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: Analyze Price Distribution

Solution

Challenge: Analyze Price Distribution

Solution