Imagine you are working on a Python script that writes data to a file. The code manually opens the file, writes to it, and then closes it after the operation. However, if an error occurs during the write operation, the file may not be closed properly, leading to resource leaks or data corruption. This kind of resource management issue can be subtle and may cause problems that are hard to debug, especially as your codebase grows.


import unittest
import user_code
import ast
import re   
import unittest
import importlib
import os
import ast

class TestTask(unittest.TestCase):
    def setUp(self):
        self.test_filename = "test_file.txt"
        try:
            os.remove(self.test_filename)
        except FileNotFoundError:
            pass

    def tearDown(self):
        try:
            os.remove(self.test_filename)
        except FileNotFoundError:
            pass

    def test_file_closed_after_write(self):
        import user_code
        importlib.reload(user_code)
        user_code.write_data(self.test_filename, "abc")
        try:
            os.rename(self.test_filename, self.test_filename + ".bak")
            closed = True
            os.rename(self.test_filename + ".bak", self.test_filename)
        except PermissionError:
            closed = False
        _dynamic_test(
            self,
            closed,
            "File is closed after writing.",
            "File was not closed after writing.",
        )

    def test_file_closed_on_exception(self):
        import user_code
        importlib.reload(user_code)
        class DummyData:
            def __str__(self):
                raise ValueError("Simulated write error")
        try:
            user_code.write_data(self.test_filename, DummyData())
        except Exception:
            pass
        try:
            os.rename(self.test_filename, self.test_filename + ".bak")
            closed = True
            os.rename(self.test_filename + ".bak", self.test_filename)
        except PermissionError:
            closed = False
        _dynamic_test(
            self,
            closed,
            "File is closed even if exception is raised during write.",
            "File was not closed after exception during write.",
        )

    def test_file_content_written(self):
        import user_code
        importlib.reload(user_code)
        test_content = "Hello, context managers!"
        user_code.write_data(self.test_filename, test_content)
        with open(self.test_filename, "r") as f:
            content = f.read()
        _dynamic_test(
            self,
            content == test_content,
            "Function writes provided data to the specified file.",
            f"Expected '{test_content}', got '{content}'",
        )

    def test_uses_with_statement(self):
        # Parse the AST to check for 'with' statement inside write_data
        with open("user_code.py", "r") as f:
            code = f.read()
        tree = ast.parse(code)
        found = False
        for node in ast.walk(tree):
            if isinstance(node, ast.FunctionDef) and node.name == "write_data":
                for child in ast.walk(node):
                    if isinstance(child, ast.With):
                        found = True
        _dynamic_test(
            self,
            found,
            "Function uses a context manager (with statement) for file handling.",
            "Function does not use a with statement for file handling.",
        )

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()
    for i, node in enumerate(tree.body):
        if isinstance(node, ast.Assign):
            for target in node.targets:
                if isinstance(target, ast.Name) and target.id == var_name:
                    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] = []
                    
                    return '\\n'.join(lines)
    return code

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


test_main.py

Master the art of resource management in Python by learning how to use and create context managers. This course covers the underlying protocol, practical applications, and advanced techniques for robust and efficient code.

Explore the fundamentals of context managers, their protocol, and how they simplify resource management in Python.

Delve into creating your own context managers using both classes and generator functions.

Solidify your understanding by tackling practical tasks that require context managers in various scenarios.

Challenge: Identify Resource Management Issues

Solution

Awesome!

Challenge: Identify Resource Management Issues

Solution