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

組み込みSQL関数を活用して、文字列、日付、数値、論理値、NULL、JSONデータを操作する方法を習得します。本モジュールでは、実践的なアプローチでSQL関数の理解と応用を深め、現実のシナリオにおけるデータクエリおよび変換スキルを向上させます。

データベース内のテキストデータを操作および分析するための、SQLの強力な文字列関数を学習します。

チャレンジ：ユーザー年齢を更新する

解答