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

Master the use of built-in SQL functions to manipulate strings, dates, numbers, logical values, nulls, and JSON data. This course provides a hands-on approach to understanding and applying SQL functions in real-world scenarios, enhancing your data querying and transformation skills.

Explore SQL's powerful string functions for manipulating and analyzing textual data in databases.

Master SQL's date and time functions to work with temporal data effectively.

Learn to use SQL's numeric functions for calculations and data analysis.

Harness SQL's logical functions to control query flow and make decisions based on data.

Master handling of NULL values in SQL using specialized functions.

Learn to work with JSON data in SQL using specialized functions for querying and manipulation.

Challenge: Update User Age

Lösung

Challenge: Update User Age

Lösung