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_first_use_each_feature(self):
        """Check if query returns correct first use date for each user-feature pair."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Run user query
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        # Expected result
        expected = [
            {'user_id': 1, 'feature_name': 'Search', 'first_use_date': pd.Timestamp('2024-05-01')},
            {'user_id': 1, 'feature_name': 'Upload', 'first_use_date': pd.Timestamp('2024-05-02')},
            {'user_id': 2, 'feature_name': 'Search', 'first_use_date': pd.Timestamp('2024-05-03')},
            {'user_id': 2, 'feature_name': 'Download', 'first_use_date': pd.Timestamp('2024-05-04')},
            {'user_id': 3, 'feature_name': 'Upload', 'first_use_date': pd.Timestamp('2024-05-06')},
            {'user_id': 3, 'feature_name': 'Download', 'first_use_date': pd.Timestamp('2024-05-08')},
            {'user_id': 4, 'feature_name': 'Search', 'first_use_date': pd.Timestamp('2024-05-10')},
            {'user_id': 4, 'feature_name': 'Download', 'first_use_date': pd.Timestamp('2024-05-11')},
            {'user_id': 4, 'feature_name': 'Upload', 'first_use_date': pd.Timestamp('2024-05-13')},
        ]
        expected_df = pd.DataFrame(expected)
        # Check that all expected rows are present
        missing_rows = []
        for _, exp_row in expected_df.iterrows():
            mask = (
                (result_df['user_id'] == exp_row['user_id']) &
                (result_df['feature_name'] == exp_row['feature_name']) &
                (pd.to_datetime(result_df['first_use_date']) == exp_row['first_use_date'])
            )
            if not mask.any():
                missing_rows.append(exp_row.to_dict())
        condition = len(missing_rows) == 0
        if not condition:
            failure_msg = f"The following expected rows are missing:\\n" + "\\n".join(str(r) for r in missing_rows)
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "All expected first use rows are present.",
            failure_msg
        )

    def test_no_extra_rows(self):
        """Check that no extra user-feature pairs are returned."""
        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)
        # Expected result count
        expected_count = 9
        condition = len(result_df) == expected_count
        _dynamic_test(
            self,
            condition,
            "No extra user-feature pairs returned.",
            f"Expected {expected_count} rows, got {len(result_df)}."
        )

    def test_columns_present(self):
        """Check that result has correct columns: user_id, feature_name, first_use_date."""
        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 = {'user_id', 'feature_name', 'first_use_date'}
        condition = set(columns) == expected_cols
        _dynamic_test(
            self,
            condition,
            "Result columns are correct.",
            f"Result columns are {columns}, but 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: Feature Adoption Metrics

Lösung

Challenge: Feature Adoption Metrics

Lösung