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_avg_session_duration_per_device(self):
        """Check that the query calculates correct average session duration per device_type in minutes."""
        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)

        # Calculate expected averages
        # desktop: session_ids 1 (60), 4 (60), 6 (60), 8 (60) => avg = 60
        # mobile: session_ids 2 (90), 5 (60), 7 (60) => avg = (90+60+60)/3 = 70
        # tablet: session_ids 3 (60), 9 (60) => avg = 60
        expected = {
            'desktop': 60.0,
            'mobile': 70.0,
            'tablet': 60.0
        }
        # Check all expected device types are present
        missing_types = [d for d in expected if d not in result_df['device_type'].values]
        condition = len(missing_types) == 0
        _dynamic_test(
            self,
            condition,
            "All expected device_type values are present.",
            f"Missing device_type(s): {', '.join(missing_types)}"
        )
        # Check average values (allowing small float error)
        for device_type, exp_avg in expected.items():
            df_row = result_df[result_df['device_type'] == device_type]
            if not df_row.empty:
                avg_val = float(df_row.iloc[0]['avg_session_duration_minutes'])
                condition = abs(avg_val - exp_avg) < 0.0001
                _dynamic_test(
                    self,
                    condition,
                    f"Average session duration for {device_type} is correct.",
                    f"Average session duration for {device_type} is {avg_val}, expected {exp_avg}."
                )

    def test_only_expected_columns(self):
        """Check that only device_type and avg_session_duration_minutes columns are present."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected_cols = {'device_type', 'avg_session_duration_minutes'}
        condition = set(columns) == expected_cols
        _dynamic_test(
            self,
            condition,
            "Result contains only the expected columns.",
            f"Result columns are {columns}, expected {expected_cols}."
        )

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: Session Duration Insights

解答