[ruff] IO operations performed on closed IO objects (RUF050)

crates/ruff_linter/resources/test/fixtures/ruff/RUF050.py

@@ -0,0 +1,106 @@
+### Errors
+
+def exhaustiveness():
+    with open() as f:
+        ...
+
+    f.__iter__()
+    f.__next__()
+    f.detach()
+    f.fileno()
+    f.flush()
+    f.isatty()
+    f.read()
+    f.readline()
+    f.readlines()
+    f.reconfigure()
+    f.seek()
+    f.seekable()
+    f.tell()
+    f.truncate()
+    f.writable()
+    f.write()
+    f.writelines()
+
+def contains():
+    with open() as f:
+        ...
+
+    _ = '' in f
+    _ = '' not in f
+    _ = '' in f is {}
+    _ = '' not in f == {}
+
+
+def for_loop():
+    with open() as f:
+        ...
+
+    for _ in f: ...
+
+
+def mode_is_unimportant():
+    with open("", "r") as f:
+        ...
+
+    f.write()
+
+
+def _():
+    with open() as f:
+        ...
+
+    _ = f.name
+    f.readlines()
+
+
+### No errors
+
+def non_io():
+    from somewhere import Lorem
+
+    with Lorem() as l:
+        l.write("")
+
+    print(l.read())
+
+def non_operations():
+    with open() as f:
+        ...
+
+    _ = f.name
+    _ = f.line_buffering()
+
+
+def compare_but_not_contains():
+    with open() as f:
+        ...
+
+    _ = a != f
+    _ = '' is not f not in {}
+
+
+def for_loop_wrapped():
+    with open() as f:
+        ...
+
+    for _ in foo(f): ...
+
+
+def aliasing():
+    with open() as f:
+        ...
+
+    g = f
+    g.readlines()
+
+
+def multiple():
+    with open() as f:
+        f.read()
+
+    with open() as f:
+        f.write()
+
+    with open() as f:
+        f.seek()

crates/ruff_linter/src/checkers/ast/analyze/bindings.rs

@@ -24,6 +24,7 @@ pub(crate) fn bindings(checker: &mut Checker) {
         Rule::PytestUnittestRaisesAssertion,
         Rule::ForLoopWrites,
         Rule::CustomTypeVarForSelf,
+        Rule::OperationOnClosedIO,
     ]) {
         return;
     }
@@ -123,5 +124,10 @@ pub(crate) fn bindings(checker: &mut Checker) {
                 checker.diagnostics.push(diagnostic);
             }
         }
+        if checker.enabled(Rule::OperationOnClosedIO) {
+            if let Some(mut diagnostics) = ruff::rules::operation_on_closed_io(checker, binding) {
+                checker.diagnostics.append(&mut diagnostics);
+            }
+        }
     }
 }

crates/ruff_linter/src/codes.rs

@@ -1002,6 +1002,7 @@ pub fn code_to_rule(linter: Linter, code: &str) -> Option<(RuleGroup, Rule)> {
         (Ruff, "047") => (RuleGroup::Preview, rules::ruff::rules::NeedlessElse),
         (Ruff, "048") => (RuleGroup::Preview, rules::ruff::rules::MapIntVersionParsing),
         (Ruff, "049") => (RuleGroup::Preview, rules::ruff::rules::DataclassEnum),
+        (Ruff, "050") => (RuleGroup::Preview, rules::ruff::rules::OperationOnClosedIO),
         (Ruff, "051") => (RuleGroup::Preview, rules::ruff::rules::IfKeyInDictDel),
         (Ruff, "052") => (RuleGroup::Preview, rules::ruff::rules::UsedDummyVariable),
         (Ruff, "055") => (RuleGroup::Preview, rules::ruff::rules::UnnecessaryRegularExpression),

crates/ruff_linter/src/rules/ruff/mod.rs

@@ -88,6 +88,7 @@ mod tests {
     #[test_case(Rule::FalsyDictGetFallback, Path::new("RUF056.py"))]
     #[test_case(Rule::RedirectedNOQA, Path::new("RUF101_0.py"))]
     #[test_case(Rule::RedirectedNOQA, Path::new("RUF101_1.py"))]
+    #[test_case(Rule::OperationOnClosedIO, Path::new("RUF050.py"))]
     fn rules(rule_code: Rule, path: &Path) -> Result<()> {
         let snapshot = format!("{}_{}", rule_code.noqa_code(), path.to_string_lossy());
         let diagnostics = test_path(

crates/ruff_linter/src/rules/ruff/rules/mod.rs

@@ -24,6 +24,7 @@ pub(crate) use mutable_fromkeys_value::*;
 pub(crate) use needless_else::*;
 pub(crate) use never_union::*;
 pub(crate) use none_not_at_end_of_union::*;
+pub(crate) use operation_on_closed_io::*;
 pub(crate) use parenthesize_chained_operators::*;
 pub(crate) use post_init_default::*;
 pub(crate) use pytest_raises_ambiguous_pattern::*;
@@ -79,6 +80,7 @@ mod mutable_fromkeys_value;
 mod needless_else;
 mod never_union;
 mod none_not_at_end_of_union;
+mod operation_on_closed_io;
 mod parenthesize_chained_operators;
 mod post_init_default;
 mod pytest_raises_ambiguous_pattern;

crates/ruff_linter/src/rules/ruff/rules/operation_on_closed_io.rs

@@ -0,0 +1,172 @@
+use crate::checkers::ast::Checker;
+use ruff_diagnostics::{Diagnostic, Violation};
+use ruff_macros::{derive_message_formats, ViolationMetadata};
+use ruff_python_ast::AnyNodeRef;
+use ruff_python_semantic::analyze::typing;
+use ruff_python_semantic::{Binding, BindingKind, NodeId, SemanticModel};
+use ruff_text_size::{Ranged, TextRange};
+
+/// ## What it does
+/// Checks for usages of IO operation methods of context variables
+/// outside of the original `with` statement.
+///
+/// ## Why is this bad?
+/// Such operations will raise `ValueError: I/O operation on closed file` at runtime.
+///
+/// ## Example
+///
+/// ```python
+/// with open(".txt") as f:
+///     f.read()
+///
+/// with open(".md", "w"):
+///     f.write("")
+/// ```
+///
+/// Use instead:
+///
+/// ```python
+/// with open(".txt") as f:
+///     f.read()
+///
+/// with open(".md", "w") as f:
+///     f.write("")
+/// ```
+#[derive(ViolationMetadata)]
+pub(crate) struct OperationOnClosedIO;
+
+impl Violation for OperationOnClosedIO {
+    #[derive_message_formats]
+    fn message(&self) -> String {
+        "IO operation performed on closed IO object".to_string()
+    }
+}
+
+/// RUF050
+pub(crate) fn operation_on_closed_io(
+    checker: &Checker,
+    binding: &Binding,
+) -> Option<Vec<Diagnostic>> {
+    if !matches!(&binding.kind, BindingKind::WithItemVar) {
+        return None;
+    }
+
+    let semantic = checker.semantic();
+    let with = binding.statement(semantic)?.as_with_stmt()?;
+
+    if !typing::is_io_base(binding, semantic) {
+        return None;
+    }
+
+    let mut diagnostics = vec![];
+
+    for reference_id in binding.references() {
+        let reference = semantic.reference(reference_id);
+
+        if reference.end() <= with.end() {
+            continue;
+        }
+
+        let Some(expression_id) = reference.expression_id() else {
+            continue;
+        };
+
+        let Some(range) = io_method_range(expression_id, semantic)
+            .or_else(|| in_io_range(expression_id, semantic))
+            .or_else(|| for_loop_io_range(expression_id, semantic))
+        else {
+            continue;
+        };
+
+        let diagnostic = Diagnostic::new(OperationOnClosedIO, range);
+
+        diagnostics.push(diagnostic);
+    }
+
+    Some(diagnostics)
+}
+
+// `f.write(...)`
+fn io_method_range(expression_id: NodeId, semantic: &SemanticModel) -> Option<TextRange> {
+    let mut ancestors = semantic.expressions(expression_id);
+
+    let _io_object_ref = ancestors.next()?;
+    let attribute = ancestors.next()?.as_attribute_expr()?;
+
+    if !is_io_operation_method(&attribute.attr.id) {
+        return None;
+    }
+
+    Some(attribute.range)
+}
+
+fn is_io_operation_method(name: &str) -> bool {
+    matches!(
+        name,
+        "__iter__"
+            | "__next__"
+            | "detach"
+            | "fileno"
+            | "flush"
+            | "isatty"
+            | "read"
+            | "readline"
+            | "readlines"
+            | "reconfigure"
+            | "seek"
+            | "seekable"
+            | "tell"
+            | "truncate"
+            | "writable"
+            | "write"
+            | "writelines"
+    )
+}
+
+// `_ in f`
+fn in_io_range(expression_id: NodeId, semantic: &SemanticModel) -> Option<TextRange> {
+    let mut ancestors = semantic.expressions(expression_id);
+
+    let io_object_ref = AnyNodeRef::from(ancestors.next()?);
+    let compare = ancestors.next()?.as_compare_expr()?;
+
+    compare
+        .comparators
+        .iter()
+        .enumerate()
+        .find_map(|(index, comparator)| {
+            if !io_object_ref.ptr_eq(comparator.into()) {
+                return None;
+            }
+
+            let op = compare.ops[index];
+
+            if !op.is_in() && !op.is_not_in() {
+                return None;
+            }
+
+            let start = if index == 0 {
+                compare.left.start()
+            } else {
+                compare.comparators[index - 1].start()
+            };
+
+            Some(TextRange::new(start, comparator.end()))
+        })
+}
+
+// `for _ in f: ...`
+fn for_loop_io_range(expression_id: NodeId, semantic: &SemanticModel) -> Option<TextRange> {
+    let mut ancestor_statements = semantic.statements(expression_id);
+
+    let io_object_ref = AnyNodeRef::from(semantic.expression(expression_id)?);
+
+    let for_loop = ancestor_statements.next()?.as_for_stmt()?;
+    let iter = for_loop.iter.as_ref();
+
+    if !io_object_ref.ptr_eq(iter.into()) {
+        return None;
+    }
+
+    Some(TextRange::new(for_loop.target.start(), iter.end()))
+}