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_correct_users_and_dates(self):
        """Check if only users who logged in exactly 30 days after signup are returned, with correct dates."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)

        # Compute expected results from init data
        self.cursor.execute('''
            SELECT u.user_id, u.signup_date, l.login_date
            FROM users u
            JOIN logins l ON l.user_id = u.user_id
            WHERE l.login_date = u.signup_date + INTERVAL '30 days'
        ''')
        expected_rows = self.cursor.fetchall()
        expected_df = pd.DataFrame(expected_rows, columns=["user_id", "signup_date", "login_date"])

        # Check that result matches expected
        missing = []
        for _, row in expected_df.iterrows():
            mask = (
                (result_df['user_id'] == row['user_id']) &
                (result_df['signup_date'] == row['signup_date']) &
                (result_df['login_date'] == row['login_date'])
            )
            if not mask.any():
                missing.append({"user_id": row['user_id'], "signup_date": row['signup_date'], "login_date": row['login_date']})
        extra = []
        for _, row in result_df.iterrows():
            mask = (
                (expected_df['user_id'] == row['user_id']) &
                (expected_df['signup_date'] == row['signup_date']) &
                (expected_df['login_date'] == row['login_date'])
            )
            if not mask.any():
                extra.append({"user_id": row['user_id'], "signup_date": row['signup_date'], "login_date": row['login_date']})
        condition = len(missing) == 0 and len(extra) == 0
        failure_msg = ""
        if missing:
            failure_msg += f"Missing expected rows: {missing}\\n"
        if extra:
            failure_msg += f"Unexpected extra rows: {extra}\\n"
        _dynamic_test(
            self,
            condition,
            "Returned users and login dates are correct for day 30 retention.",
            failure_msg
        )

    def test_only_exact_30_days(self):
        """Ensure no users are returned for logins that are not exactly 30 days after signup."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        for row in rows:
            user_id, signup_date, login_date = row
            delta = (login_date - signup_date).days
            _dynamic_test(
                self,
                delta == 30,
                f"User {user_id} returned for exactly 30 days after signup.",
                f"User {user_id} login_date is not exactly 30 days after signup (delta: {delta})."
            )

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: Retention Cohorts

解答