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_all_rows_returned(self):
        """Check that the query returns all customers."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == 10,
            "Returned all rows from customers table.",
            f"Expected 10 rows, but got {len(rows)}."
        )

    def test_cleaned_city_no_spaces(self):
        """Check that cleaned_city has no leading or trailing spaces."""
        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]
        cleaned_city_idx = columns.index("cleaned_city")
        spaces_found = [row[cleaned_city_idx] for row in rows if row[cleaned_city_idx] != row[cleaned_city_idx].strip()]
        _dynamic_test(
            self,
            len(spaces_found) == 0,
            "No leading or trailing spaces in cleaned_city.",
            f"Found spaces in cleaned_city values: {spaces_found}"
        )

    def test_cleaned_city_first_letter_upper(self):
        """Check that first letter of cleaned_city is uppercase."""
        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]
        cleaned_city_idx = columns.index("cleaned_city")
        failures = [row[cleaned_city_idx] for row in rows if not row[cleaned_city_idx][0].isupper()]
        _dynamic_test(
            self,
            len(failures) == 0,
            "First letter of cleaned_city is uppercase for all rows.",
            f"First letter is not uppercase in: {failures}"
        )

    def test_cleaned_city_rest_lower(self):
        """Check that all other letters in cleaned_city are lowercase."""
        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]
        cleaned_city_idx = columns.index("cleaned_city")
        failures = []
        for row in rows:
            city = row[cleaned_city_idx]
            if len(city) > 1 and city[1:] != city[1:].lower():
                failures.append(city)
        _dynamic_test(
            self,
            len(failures) == 0,
            "All letters after the first in cleaned_city are lowercase.",
            f"Non-lowercase letters found in: {failures}"
        )

    def test_cleaned_city_matches_expected(self):
        """Check that cleaned_city contains the cleaned version of city for every customer."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        result_rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(result_rows, columns=columns)
        # Build expected cleaned city column
        self.cursor.execute("SELECT id, city FROM customers")
        orig = self.cursor.fetchall()
        expected = {}
        for row in orig:
            city = row[1]
            city_trim = city.strip()
            if city_trim:
                cleaned = city_trim[0].upper() + city_trim[1:].lower()
            else:
                cleaned = ''
            expected[row[0]] = cleaned
        # Compare cleaned_city for each row
        mismatches = []
        for _, row in result_df.iterrows():
            if row['id'] in expected:
                if row['cleaned_city'] != expected[row['id']]:
                    mismatches.append((row['id'], row['cleaned_city'], expected[row['id']]))
        _dynamic_test(
            self,
            len(mismatches) == 0,
            "cleaned_city matches expected transformation for all rows.",
            f"Rows with mismatched cleaned_city: {mismatches}"
        )

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 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: Clean Up City Names

Lösung

Challenge: Clean Up City Names

Lösung