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_joins_on_customer_id(self):
        """Check if the query joins sales and customers on customer_id."""
        with open("query.sql", "r") as file:
            query = file.read()
        join_condition = "join customers c on s.customer_id = c.customer_id"
        join_condition_alt = "join customers c on (s.customer_id = c.customer_id)"
        normalized_query = normalize_text(query)
        condition = join_condition in normalized_query or join_condition_alt in normalized_query
        _dynamic_test(
            self,
            condition,
            "The query joins sales and customers on customer_id.",
            "The query does not join sales and customers on customer_id."
        )

    def test_groups_by_city(self):
        """Check if the query groups results by city."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized_query = normalize_text(query)
        condition = "group by c.city" in normalized_query or "group by city" in normalized_query
        _dynamic_test(
            self,
            condition,
            "The query groups results by city.",
            "The query does not group results by city."
        )

    def test_returns_cities_with_sales(self):
        """Check that the query returns each city present in customers table that has at least one sale."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        result_rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(result_rows, columns=columns)
        self.cursor.execute("""
            SELECT DISTINCT c.city
            FROM sales s
            JOIN customers c ON s.customer_id = c.customer_id
        """)
        expected_cities = set(row[0] for row in self.cursor.fetchall())
        result_cities = set(result_df['city'])
        missing = expected_cities - result_cities
        condition = len(missing) == 0
        _dynamic_test(
            self,
            condition,
            "The query returns all cities with at least one sale.",
            f"The following cities with sales are missing: {', '.join(missing)}" if missing else ""
        )

    def test_sales_count_column(self):
        """Check that the count of sales column is named sales_count and is correct."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        result_rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(result_rows, columns=columns)
        condition = 'sales_count' in result_df.columns
        _dynamic_test(
            self,
            condition,
            "The column with sales count is named sales_count.",
            "The column with sales count is not named sales_count."
        )
        # Now check correctness of sales_count values
        self.cursor.execute("""
            SELECT c.city, COUNT(s.sale_id) AS expected_count
            FROM sales s
            JOIN customers c ON s.customer_id = c.customer_id
            GROUP BY c.city
        """)
        expected = pd.DataFrame(self.cursor.fetchall(), columns=['city', 'expected_count'])
        merged = pd.merge(result_df, expected, on='city', how='outer', suffixes=('', '_expected'))
        merged['match'] = merged['sales_count'] == merged['expected_count']
        mismatch = merged[~merged['match']]
        condition = mismatch.empty
        fail_msg = "Sales count does not match expected for: " + ", ".join(mismatch['city']) if not mismatch.empty else ""
        _dynamic_test(
            self,
            condition,
            "The sales_count values are correct for each city.",
            fail_msg
        )

    def test_sales_count_uses_sale_id(self):
        """Check that the query counts sales using sale_id."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized_query = normalize_text(query)
        condition = "count(s.sale_id)" in normalized_query or "count(*)" in normalized_query
        _dynamic_test(
            self,
            condition,
            "The query counts sales using sale_id.",
            "The query does not count sales using sale_id."
        )

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

A comprehensive beginner-friendly course designed to teach SQL from the ground up, focusing on practical data analysis skills. Through engaging explanations, real-world examples, and hands-on challenges, learners will master querying, transforming, and analyzing data using SQL.

Learn the foundational concepts of relational databases and SQL. Explore how to retrieve, filter, and sort data from tables, setting the stage for deeper data analysis.

Dive into SQL's powerful aggregation features. Learn to summarize, group, and analyze data for deeper insights.

Unlock the power of combining data from multiple tables. Learn to perform different types of joins to answer complex business questions.

Master advanced SQL techniques for powerful data analysis, including subqueries, window functions, and data transformation.

Challenge: City-Based Sales Analysis

Рішення