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

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

解答