import unittest
import psycopg2
import json
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 setUp(self):
        # Reset data before each test
        self.cursor.execute("DELETE FROM user_profiles;")
        self.cursor.execute("""
        INSERT INTO user_profiles (user_id, profile_data) VALUES
        (1, '{"age": 25, "interests": ["reading", "hiking", "coding"], "address": {"street": "123 Maple St", "city": "Springfield", "state": "IL", "zip": "62704"}}'),
        (2, '{"age": 32, "interests": ["cooking", "traveling"], "address": {"street": "456 Oak Ave", "city": "Greenville", "state": "TX", "zip": "75401"}}'),
        (3, '{"age": 19, "interests": ["gaming", "music", "basketball"], "address": {"street": "789 Pine Rd", "city": "Fairview", "state": "CA", "zip": "94541"}}'),
        (4, '{"age": 41, "interests": ["photography", "gardening"], "address": {"street": "321 Birch Blvd", "city": "Lakeside", "state": "FL", "zip": "32065"}}'),
        (5, '{"age": 28, "interests": ["yoga", "painting", "cycling"], "address": {"street": "654 Cedar Ln", "city": "Hilltown", "state": "NY", "zip": "10940"}}');
        """)
        self.connection.commit()

    def test_age_incremented_by_1(self):
        """Check that each user's age is increased by exactly 1."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        self.connection.commit()
        self.cursor.execute("SELECT user_id, profile_data::text FROM user_profiles ORDER BY user_id;")
        rows = self.cursor.fetchall()
        # Expected ages after increment
        expected_ages = {1: 26, 2: 33, 3: 20, 4: 42, 5: 29}
        all_ok = True
        failed = []
        for user_id, profile_data in rows:
            data = json.loads(profile_data)
            if int(data["age"]) != expected_ages[user_id]:
                all_ok = False
                failed.append(f"user_id {user_id}: expected {expected_ages[user_id]}, got {data['age']}")
        _dynamic_test(
            self,
            all_ok,
            "The age field for every user is incremented by 1.",
            "Incorrect age values after update: " + ", ".join(failed)
        )

    def test_other_fields_unchanged(self):
        """Check that all other fields in profile_data remain unchanged."""
        # Store original data except for age
        self.cursor.execute("SELECT user_id, profile_data::text FROM user_profiles ORDER BY user_id;")
        before = {user_id: json.loads(profile_data) for user_id, profile_data in self.cursor.fetchall()}
        for v in before.values():
            v.pop('age')
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        self.connection.commit()
        self.cursor.execute("SELECT user_id, profile_data::text FROM user_profiles ORDER BY user_id;")
        after = {user_id: json.loads(profile_data) for user_id, profile_data in self.cursor.fetchall()}
        for v in after.values():
            v.pop('age')
        unchanged = before == after
        _dynamic_test(
            self,
            unchanged,
            "All other fields in profile_data remain unchanged.",
            "Some non-age fields were modified."
        )

    def test_all_rows_updated(self):
        """Check that all rows in the table are affected by the update."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        self.connection.commit()
        self.cursor.execute("SELECT COUNT(*) FROM user_profiles;")
        count = self.cursor.fetchone()[0]
        self.cursor.execute("SELECT COUNT(*) FROM user_profiles WHERE (profile_data->>'age')::int IN (26,33,20,42,29);")
        updated_count = self.cursor.fetchone()[0]
        _dynamic_test(
            self,
            count == updated_count,
            "All rows in the table were updated.",
            "Not all rows were updated."
        )


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):
    import re
    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

This beginner course focuses on the essential SQL functions used for data cleaning and transformation. Learners will practice handling missing values, formatting and standardizing data, and applying string, date, numeric, logical, and null handling functions to prepare datasets for analysis. The course provides hands-on challenges directly related to the most common data preparation tasks in SQL.

Challenge: Update User Age

解答