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_selects_correct_columns(self):
        """Check if query selects the correct columns from products."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected_columns = ["product_id", "name", "price", "discount", "stock"]
        condition = set(columns) == set(expected_columns)
        _dynamic_test(
            self,
            condition,
            "Query selects the correct columns.",
            f"Query does not select the correct columns. Got: {columns}"
        )

    def test_returns_only_low_stock(self):
        """Check that only products with stock <= 50 are returned."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        if len(rows) == 0:
            condition = True  # Acceptable if no such products exist
        else:
            condition = all(row[4] <= 50 for row in rows)
        _dynamic_test(
            self,
            condition,
            "All returned products have stock <= 50.",
            "Query returns products with stock greater than 50."
        )

    def test_no_extra_rows(self):
        """Check that products with stock > 5 are not included in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        returned_ids = {row[0] for row in rows}
        self.cursor.execute("SELECT product_id FROM products WHERE stock > 50;")
        nonlow_ids = {row[0] for row in self.cursor.fetchall()}
        condition = returned_ids.isdisjoint(nonlow_ids)
        _dynamic_test(
            self,
            condition,
            "No products with stock > 50 are included.",
            "Query includes products with stock > 50."
        )


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: Find Low Stock Products

Solution

Challenge: Find Low Stock Products

Solution