
import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import importlib
import io
import sys
import pandas as pd
from contextlib import redirect_stdout

class TestTask(unittest.TestCase):
    def setUp(self):
        self.data = {
            'amount': [5000, 15000, 9000, 20000, 7000],
            'country': ['USA', 'USA', 'Iran', 'France', 'North Korea']
        }
        self.df = pd.DataFrame(self.data)
        self.high_risk = ['Iran', 'North Korea']

    def test_column_added(self):
        import user_code
        importlib.reload(user_code)
        df_copy = self.df.copy()
        result = user_code.flag_compliance(df_copy, self.high_risk)
        _dynamic_test(
            self,
            'compliance_flag' in result.columns,
            "compliance_flag column is added to DataFrame",
            "compliance_flag column was not added to DataFrame"
        )

    def test_amount_flag(self):
        import user_code
        importlib.reload(user_code)
        df_copy = self.df.copy()
        result = user_code.flag_compliance(df_copy, self.high_risk)
        # Index 1 and 3 have amounts > 10000
        _dynamic_test(
            self,
            result.loc[1, 'compliance_flag'] == True and result.loc[3, 'compliance_flag'] == True,
            "Transactions with amount > 10000 are flagged",
            "Transactions with amount > 10000 are not correctly flagged"
        )

    def test_high_risk_country_flag(self):
        import user_code
        importlib.reload(user_code)
        df_copy = self.df.copy()
        result = user_code.flag_compliance(df_copy, self.high_risk)
        # Index 2 (Iran) and 4 (North Korea) should be flagged
        _dynamic_test(
            self,
            result.loc[2, 'compliance_flag'] == True and result.loc[4, 'compliance_flag'] == True,
            "Transactions from high-risk countries are flagged",
            "Transactions from high-risk countries are not correctly flagged"
        )

    def test_non_flagged_transaction(self):
        import user_code
        importlib.reload(user_code)
        df_copy = self.df.copy()
        result = user_code.flag_compliance(df_copy, self.high_risk)
        # Index 0: amount=5000, country=USA, not flagged
        _dynamic_test(
            self,
            result.loc[0, 'compliance_flag'] == False,
            "Non-flagged transactions remain unflagged",
            "Non-flagged transactions are incorrectly flagged"
        )

    def test_summary_output(self):
        import user_code
        importlib.reload(user_code)
        df_copy = self.df.copy()
        f = io.StringIO()
        with redirect_stdout(f):
            user_code.flag_compliance(df_copy, self.high_risk)
        output = normalize_text(f.getvalue())
        expected = normalize_text("Flagged 4 out of 5 transactions for compliance review.")
        _dynamic_test(
            self,
            expected in output,
            "Summary statement prints correct flagged and total count",
            f"Expected output: '{expected}' not found in: {output}"
        )

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()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    changed = False
    # Collect all assignment nodes to modify
    assign_nodes = [
        (i, node)
        for i, node in enumerate(tree.body)
        if isinstance(node, ast.Assign)
        and any(isinstance(target, ast.Name) and target.id == var_name for target in node.targets)
    ]

    # If nothing to change, return unmodified code
    if not assign_nodes:
        return code

    # Perform replacements for all matching assignments (from last to first to not break line offsets)
    for i, node in reversed(assign_nodes):
        start_line = node.lineno - 1
        line = lines[start_line]
        indent = ' ' * (len(line) - len(line.lstrip()))
        lines[start_line] = f"{indent}{var_name} = {value}"
        next_line = len(lines)
        for next_node in tree.body[i+1:]:
            if hasattr(next_node, 'lineno'):
                next_line = next_node.lineno - 1
                break
        if next_line > start_line + 1:
            lines[start_line+1:next_line] = []
        changed = True

    return '\\n'.join(lines) if changed else code

if __name__ == "__main__":
    unittest.main()


test_main.py

A practical Python course tailored for banking professionals, focusing on real-world financial data analysis, risk assessment, and automation tasks. Learn how Python can streamline banking operations, analyze financial data, and solve common problems faced by bankers.

Learn how to use Python to analyze, manipulate, and visualize financial data relevant to banking operations.

Apply Python to assess credit risk, analyze loan portfolios, and perform key calculations relevant to banking risk management.

Explore how Python can automate banking tasks and help detect fraudulent activities using data analysis techniques.

Challenge: Implement Compliance Flags

解答