import unittest
import user_code
import ast
import re   
import importlib
import csv
import polars as pl

class TestTask(unittest.TestCase):
    def setUp(self):
        # Setup mock primary dataframe matching raw specifications
        self.df = pl.DataFrame({
            "Name": ["Portal", "Half-Life", "Left 4 Dead"],
            "Price": [9.99, 9.99, 19.99]
        })
        
        # Setup mock insights dataframe using string keys
        self.insights_df = pl.DataFrame({
            "Name": ["Portal", "Left 4 Dead", "Dota 2"], # Half-Life is missing, Dota 2 is unmapped
            "Marketing_Tier": ["Gold", "Silver", "Bronze"],
            "Ad_Budget": [5000, 2000, 1500]
        })

    def test_join_columns_exist_and_unaltered(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, "merge_marketing_by_name", None)
        _dynamic_test(
            self,
            callable(func),
            "merge_marketing_by_name function exists and is callable.",
            "merge_marketing_by_name function is missing or not callable."
        )
        result = func(self.df, self.insights_df)
        
        _dynamic_test(
            self,
            result is not None and "Name" in result.columns,
            "Returned DataFrame retains original key tracking column 'Name'.",
            "Returned DataFrame is missing 'Name'. Ensure exact casing is used."
        )
        for col in ["Price", "Marketing_Tier", "Ad_Budget"]:
            _dynamic_test(
                self,
                col in result.columns,
                f"Merged column layout includes variable '{col}'.",
                f"Missing expected attribute column '{col}' after running the join."
            )

    def test_left_join_logic_and_null_fill(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, "merge_marketing_by_name", None)
        result = func(self.df, self.insights_df)
        
        if result is not None and "Marketing_Tier" in result.columns:
            _dynamic_test(
                self,
                len(result) == 3,
                "Left join operation successfully maintained primary dataset row lengths.",
                f"Row counts altered. Expected 3 rows, but received {len(result)} rows."
            )
            
            # Aligned key evaluation (Name: "Portal")
            portal_row = result.filter(pl.col("Name") == "Portal")
            _dynamic_test(
                self,
                portal_row["Marketing_Tier"][0] == "Gold",
                "Successfully matched and merged attributes using string keys.",
                f"Incorrect tier mapping for Portal. Expected 'Gold', Got: {portal_row['Marketing_Tier'][0]}"
            )
            
            # Unaligned key verification (Name: "Half-Life" -> missing from right table)
            halflife_row = result.filter(pl.col("Name") == "Half-Life")
            _dynamic_test(
                self,
                halflife_row["Marketing_Tier"][0] is None and halflife_row["Ad_Budget"][0] is None,
                "Left join rules validated: missing right values successfully defaulted to Null.",
                f"Error on missing key values. Expected Null assignments, Got: {halflife_row['Marketing_Tier'][0]}"
            )
        else:
            _dynamic_test(self, False, "", "Join verification halted due to schema structural sorting 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()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    changed = False
    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 not assign_nodes:
        return code

    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

Ein praxisorientierter, mittlerer Kurs für Python-Entwickler zur Beherrschung des Data Wrangling mit Polars anhand von realen Steam-Spieldatensätzen. Vermittelt wird das spaltenorientierte Paradigma von Polars, Selektion, Aggregation, Verknüpfung, Umformung sowie grundlegende Operationen mit Zeichenketten, Datumswerten und fehlenden Daten.

Untersuchung der grundlegenden Unterschiede zwischen Polars und traditionellen zeilenbasierten DataFrames mit Schwerpunkt auf spaltenbasierten Operationen, Selektion und bedingter Logik anhand des Steam-Games-Datensatzes.

Erwerben Sie ein fundiertes Verständnis der parallelen Group-by-, Aggregations-, Join- und Umformungsfunktionen von Polars anhand der Steam-Spiele- und Spy-Insights-Datensätze.

Erkunden Sie die Grundlagen der Datenbereinigung: Textnormalisierung, Datumsanalyse und robuste Behandlung fehlender Werte in Polars.

Challenge: Zusammenführen von Daten

Lösung