You have explored logical functions in SQL, learning how to use them to make decisions based on data values. In this challenge, you will apply that knowledge to identify potentially suspicious transactions in a payments table. Your goal is to flag transactions as **"Suspicious"** if their `amount` is greater than or equal to `1000`, or if their `status` is `"failed"`. This is a common scenario in fraud detection and risk management, where it is important to highlight transactions that require further review.

import unittest
import psycopg2
import pandas as pd
import re

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_flag_column_present(self):
        """Check that the result includes a 'flag' column."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        condition = 'flag' in [col.lower() for col in columns]
        _dynamic_test(
            self,
            condition,
            "The result includes a 'flag' column.",
            "The result does not include a 'flag' column."
        )

    def test_amount_1000_or_more_flagged_suspicious(self):
        """Transactions with amount >= 1000 are flagged as 'Suspicious'."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Add a transaction >= 1000 for this test
        self.cursor.execute("INSERT INTO transactions (transaction_id, amount, status, payment_method) VALUES (100, 1200.00, 'completed', 'credit_card') ON CONFLICT DO NOTHING;")
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        flagged = df[(df['transaction_id'] == 100)]
        condition = not flagged.empty and normalize_text(flagged.iloc[0]['flag']) == 'suspicious'
        _dynamic_test(
            self,
            condition,
            "Transactions with amount >= 1000 are flagged as 'Suspicious'.",
            "A transaction with amount >= 1000 was not flagged as 'Suspicious'."
        )

    def test_status_failed_flagged_suspicious(self):
        """Transactions with status = 'failed' are flagged as 'Suspicious'."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        failed_rows = df[df['status'].apply(lambda x: normalize_text(str(x)) == 'failed')]
        suspicious_flags = failed_rows['flag'].apply(lambda x: normalize_text(str(x)) == 'suspicious')
        condition = suspicious_flags.all() and not failed_rows.empty
        _dynamic_test(
            self,
            condition,
            "All transactions with status = 'failed' are flagged as 'Suspicious'.",
            "Not all transactions with status = 'failed' are flagged as 'Suspicious'."
        )

    def test_other_transactions_flagged_normal(self):
        """All other transactions are flagged as 'Normal'."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        # Exclude rows where amount >= 1000 or status = 'failed'
        mask = (df['amount'] < 1000) & (df['status'].apply(lambda x: normalize_text(str(x)) != 'failed'))
        normal_rows = df[mask]
        if not normal_rows.empty:
            all_normal = normal_rows['flag'].apply(lambda x: normalize_text(str(x)) == 'normal').all()
        else:
            all_normal = True
        _dynamic_test(
            self,
            all_normal,
            "All other transactions are flagged as 'Normal'.",
            "Not all other transactions are flagged as 'Normal'."
        )

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: Flag Suspicious Transactions

Lösung

Challenge: Flag Suspicious Transactions

Lösung