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_monthly_adoption_rates_upload(self):
        """Check if the query returns correct monthly adoption rates for 'Upload' feature."""
        with open("query.sql", "r") as file:
            query = file.read()

        solution_query = '''
        SELECT
          DATE_TRUNC('month', event_date) AS month_start,
          COUNT(DISTINCT user_id) AS users_adopted,
          (COUNT(DISTINCT user_id) * 100.0 / (SELECT COUNT(*) FROM users)) AS adoption_rate_percent
        FROM feature_events
        WHERE feature_name = 'Upload'
        GROUP BY month_start
        ORDER BY month_start;
        '''
        self.cursor.execute(solution_query)
        sol_rows = self.cursor.fetchall()
        sol_columns = [desc[0] for desc in self.cursor.description]
        sol_df = pd.DataFrame(sol_rows, columns=sol_columns)

        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)

        # Check that all expected rows are present
        missing_rows = []
        for _, expected_row in sol_df.iterrows():
            expected_dict = expected_row.to_dict()
            # Compare float columns with tolerance
            mask = (
                (df['month_start'] == expected_dict['month_start']) &
                (df['users_adopted'] == expected_dict['users_adopted']) &
                (abs(df['adoption_rate_percent'] - expected_dict['adoption_rate_percent']) < 0.001)
            )
            if not mask.any():
                missing_rows.append(expected_dict)
        condition = len(missing_rows) == 0
        if not condition:
            human_readable = "\\n".join(str(r) for r in missing_rows)
            failure_msg = f"The following expected rows are missing:\\n{human_readable}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Correct monthly adoption rates for 'Upload' feature are returned.",
            failure_msg
        )

    def test_only_upload_feature(self):
        """Check that only 'Upload' feature is included in the result."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        # Check there are no adoption rates for other features
        # Get all events for 'Upload' and get month_start values
        self.cursor.execute("""
            SELECT DISTINCT DATE_TRUNC('month', event_date) AS month_start
            FROM feature_events
            WHERE feature_name = 'Upload'
        """)
        upload_months = set(row[0] for row in self.cursor.fetchall())
        # All rows in result should correspond to these months
        result_months = set(df['month_start'])
        condition = result_months.issubset(upload_months)
        _dynamic_test(
            self,
            condition,
            "Result contains only months with 'Upload' events.",
            "Result contains months without 'Upload' events."
        )

    def test_column_names(self):
        """Check that column names are as required."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required = ["month_start", "users_adopted", "adoption_rate_percent"]
        condition = columns == required
        _dynamic_test(
            self,
            condition,
            "Column names are correct.",
            f"Column names are incorrect. Got: {columns}"
        )

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: Adoption Trend Analysis

Рішення

Challenge: Adoption Trend Analysis

Рішення