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_total_sessions_per_user(self):
        """Check that the first query returns correct total sessions per user."""
        with open("query.sql", "r") as file:
            query = file.read()
        queries = [q.strip() for q in query.split(';') if q.strip()]
        # First query: total sessions per user
        self.cursor.execute(queries[0])
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        # Expected results
        expected = pd.DataFrame([
            {'user_id': 1, 'total_sessions': 3},
            {'user_id': 2, 'total_sessions': 3},
            {'user_id': 3, 'total_sessions': 3},
        ])
        # Merge and check
        merged = pd.merge(result_df, expected, on=['user_id', 'total_sessions'], how='inner')
        condition = len(merged) == 3 and len(result_df) == 3
        _dynamic_test(
            self,
            condition,
            "Query returns correct total sessions per user.",
            f"Query did not return correct total sessions per user. Got: {result_df.to_dict('records')}"
        )

    def test_users_with_more_than_three_sessions(self):
        """Check that the second query returns only users with more than three sessions (should be none)."""
        with open("query.sql", "r") as file:
            query = file.read()
        queries = [q.strip() for q in query.split(';') if q.strip()]
        if len(queries) < 2:
            _dynamic_test(
                self,
                False,
                "Second query exists.",
                "Second query (for users with more than three sessions) is missing."
            )
            return
        self.cursor.execute(queries[1])
        rows = self.cursor.fetchall()
        condition = len(rows) == 0
        _dynamic_test(
            self,
            condition,
            "No users with more than three sessions returned as expected.",
            f"Unexpected users returned with more than three sessions: {rows}"
        )

    def test_no_users_with_three_or_fewer_sessions_in_second_query(self):
        """Ensure second query does not include users with three or fewer sessions."""
        with open("query.sql", "r") as file:
            query = file.read()
        queries = [q.strip() for q in query.split(';') if q.strip()]
        if len(queries) < 2:
            _dynamic_test(
                self,
                False,
                "Second query exists.",
                "Second query (for users with more than three sessions) is missing."
            )
            return
        self.cursor.execute(queries[1])
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        # All users have only 3 sessions, so result should be empty
        condition = result_df.empty
        _dynamic_test(
            self,
            condition,
            "Second query correctly excludes users with three or fewer sessions.",
            "Second query incorrectly includes users with three or fewer sessions."
        )

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: Querying Session Data

解答