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_active_users_last_3_days(self):
        """Check that only users with sessions in the last 3 days are returned (including today)."""
        with open("query.sql", "r") as file:
            query = file.read()

        # Simulate CURRENT_DATE as 2024-06-03 (latest session date in dataset)
        # So the window is 2024-06-01, 2024-06-02, 2024-06-03 (inclusive)
        # All users have at least one session in this window
        expected_users = set([
            (1, 'Alice Johnson'),
            (2, 'Bob Smith'),
            (3, 'Charlie Lee')
        ])

        # Patch CURRENT_DATE to 2024-06-03 by setting the timezone for the session
        self.cursor.execute("SET TIME ZONE 'UTC';")
        self.cursor.execute("SET LOCAL timezone TO 'UTC';")
        # Run the user query
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        user_set = set((row[0], row[1]) for row in rows)

        missing = expected_users - user_set
        unexpected = user_set - expected_users
        condition = (missing == set()) and (unexpected == set())
        msg = ""
        if missing:
            msg += f"Missing users: {missing}. "
        if unexpected:
            msg += f"Unexpected users: {unexpected}."
        _dynamic_test(
            self,
            condition,
            "All and only active users in the last 3 days are returned.",
            msg
        )

    def test_no_duplicate_users(self):
        """Ensure each user appears only once, even with multiple sessions in the window."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        user_ids = [row[0] for row in rows]
        condition = len(user_ids) == len(set(user_ids))
        _dynamic_test(
            self,
            condition,
            "No duplicate users in result.",
            "Duplicate user IDs found in result."
        )

    def test_only_users_with_sessions_in_window(self):
        """If the window is changed to last 1 day (2024-06-03 only), only Bob Smith should be returned."""
        # Run a modified query for just 2024-06-03
        user_query = """
SELECT DISTINCT
  u.user_id,
  u.name
FROM users u
JOIN sessions s ON u.user_id = s.user_id
WHERE s.session_start >= DATE '2024-06-03' AND s.session_start < DATE '2024-06-04';
        """
        self.cursor.execute(user_query)
        rows = self.cursor.fetchall()
        expected = [(2, 'Bob Smith')]
        condition = sorted(rows) == sorted(expected)
        _dynamic_test(
            self,
            condition,
            "Correctly filters only users active on 2024-06-03.",
            f"Expected only Bob Smith, got: {rows}"
        )

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: Active User Metrics

解答