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_columns_present(self):
        """Check that all original columns plus masked_name are present in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected_columns = set(["id", "name", "masked_name", "email", "city"])
        _dynamic_test(
            self,
            expected_columns.issubset(set(columns)),
            "All required columns are present in the result.",
            f"Missing columns: {expected_columns - set(columns)}"
        )

    def test_all_customers_included(self):
        """Check that all customers are included in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT COUNT(*) FROM customers;")
        expected_count = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == expected_count,
            "All customers are included in the result.",
            f"Expected {expected_count} customers, got {len(rows)}."
        )

    def test_masked_name_format(self):
        """Check that masked_name is correct: first character + asterisks for the rest."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT id, name FROM customers;")
        customer_data = {row[0]: row[1] for row in self.cursor.fetchall()}
        self.cursor.execute(query)
        result = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        idx_id = columns.index("id")
        idx_masked = columns.index("masked_name")
        errors = []
        for row in result:
            cid = row[idx_id]
            name = customer_data[cid]
            masked = row[idx_masked]
            if len(name) < 1:
                expected = ''
            else:
                expected = name[0] + ('*' * (len(name)-1))
            if masked != expected:
                errors.append(f"id={cid}: expected '{expected}', got '{masked}'")
        _dynamic_test(
            self,
            not errors,
            "masked_name column is formatted correctly for all rows.",
            "Incorrect masking in rows:\\n" + "\\n".join(errors)
        )

    def test_only_first_character_visible(self):
        """Check that only the first character from the name is visible in masked_name, rest are asterisks."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT id, name FROM customers;")
        customer_data = {row[0]: row[1] for row in self.cursor.fetchall()}
        self.cursor.execute(query)
        result = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        idx_id = columns.index("id")
        idx_masked = columns.index("masked_name")
        errors = []
        for row in result:
            cid = row[idx_id]
            name = customer_data[cid]
            masked = row[idx_masked]
            if len(name) > 1 and not all(c == '*' for c in masked[1:]):
                errors.append(f"id={cid}: masked_name has non-asterisk characters after first letter: {masked}")
        _dynamic_test(
            self,
            not errors,
            "Only the first character is visible in masked_name for all rows.",
            "Non-asterisk characters found after the first letter in:\\n" + "\\n".join(errors)
        )

    def test_asterisk_count(self):
        """Check that the number of asterisks matches the number of characters in the name minus one."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT id, name FROM customers;")
        customer_data = {row[0]: row[1] for row in self.cursor.fetchall()}
        self.cursor.execute(query)
        result = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        idx_id = columns.index("id")
        idx_masked = columns.index("masked_name")
        errors = []
        for row in result:
            cid = row[idx_id]
            name = customer_data[cid]
            masked = row[idx_masked]
            expected_asterisks = max(0, len(name) - 1)
            actual_asterisks = masked.count('*')
            if actual_asterisks != expected_asterisks:
                errors.append(f"id={cid}: expected {expected_asterisks} asterisks, got {actual_asterisks} in '{masked}'")
        _dynamic_test(
            self,
            not errors,
            "Asterisk count in masked_name is correct for all rows.",
            "Incorrect asterisk count in:\\n" + "\\n".join(errors)
        )

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: Mask Sensitive Information

Solution

Challenge: Mask Sensitive Information

Solution