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_churned_users_last_30_days(self):
        """Check if query returns correct user IDs who have not logged in in the last 30 days or never logged in."""
        with open("query.sql", "r") as file:
            query = file.read()

        # For testing, set CURRENT_DATE to 2023-06-15 (for deterministic results)
        # So last 30 days is since 2023-05-16
        # User 1: last login 2023-05-18 (active)
        # User 2: last login 2023-03-10 (churned)
        # User 3: last login 2023-06-01 (active)
        # User 4: last login 2023-04-10 (churned)
        # User 5: last login 2023-05-01 (churned)
        # Expected churned users: 2, 4, 5

        # Patch CURRENT_DATE to 2023-06-15 for test
        solution_query = """
        SELECT u.user_id
        FROM users u
        LEFT JOIN logins l ON u.user_id = l.user_id
        GROUP BY u.user_id
        HAVING MAX(l.login_date) < DATE '2023-06-15' - INTERVAL '30 days' OR MAX(l.login_date) IS NULL;
        """
        self.cursor.execute(solution_query)
        expected_rows = self.cursor.fetchall()
        expected_ids = set(row[0] for row in expected_rows)

        # Replace CURRENT_DATE with '2023-06-15' in user query for deterministic test
        user_query = query.replace('CURRENT_DATE', "DATE '2023-06-15'")
        self.cursor.execute(user_query)
        actual_rows = self.cursor.fetchall()
        actual_ids = set(row[0] for row in actual_rows)

        condition = actual_ids == expected_ids
        if not condition:
            missing = expected_ids - actual_ids
            extra = actual_ids - expected_ids
            msg = []
            if missing:
                msg.append(f"Missing user_ids: {sorted(missing)}")
            if extra:
                msg.append(f"Unexpected user_ids: {sorted(extra)}")
            failure_msg = ", ".join(msg)
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Query returns the correct set of churned user IDs.",
            failure_msg
        )

    def test_column_name(self):
        """Check if the result has the correct column name 'user_id'."""
        with open("query.sql", "r") as file:
            query = file.read()
        user_query = query.replace('CURRENT_DATE', "DATE '2023-06-15'")
        self.cursor.execute(user_query)
        columns = [desc[0] for desc in self.cursor.description]
        condition = columns == ['user_id']
        failure_msg = f"Result column should be named 'user_id', but got: {columns}"
        _dynamic_test(
            self,
            condition,
            "Result column is correctly named 'user_id'.",
            failure_msg
        )

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 SQL from the ground up with a focus on real-world product analytics. Learn to analyze sessionized data, measure retention and churn, track feature adoption, and perform A/B test analysis using advanced SQL techniques. This course is designed for aspiring product analysts with no prior SQL experience, blending engaging theory with hands-on challenges.

Learn how to structure, query, and analyze sessionized user data, a foundational skill for product analytics.

Dive into retention and churn metrics, learning how to measure and interpret user engagement over time.

Learn to track and analyze how users discover and adopt new product features using SQL.

Master advanced SQL techniques for analyzing A/B tests and using window functions for product metrics.

Challenge: Churn Detection

解答