1use brokk_bifrost_core::analyzer::Range;
14use brokk_bifrost_core::analyzer::semantic_diagnostics::contains_node as covers;
15use brokk_bifrost_core::analyzer::structural::spec::StructuralSpec;
16use std::collections::{HashMap, HashSet};
17use tree_sitter::Node;
18
19use crate::bindings::{
20 PythonDirectScopeBindingKind, PythonLexicalNameResolution, PythonLexicalScopeInventory,
21 python_direct_scope_bindings_bounded, python_executable_scope_children,
22 python_name_resolution_at,
23};
24use crate::imports::python_import_bindings_from_tree;
25use crate::syntax::{python_parameter_name, python_plain_string_literal};
26
27const PROCESS_INPUT_CONTAINERS: [(&str, &str); 2] = [("os", "environ"), ("sys", "argv")];
35
36#[derive(Clone, Debug, Eq, PartialEq)]
41pub enum PythonRuntimeAccessKey {
42 Property(String),
44 Index(u128),
46 Dynamic,
48 Unsupported,
51}
52
53#[derive(Clone, Copy, Debug, Eq, PartialEq)]
55pub enum PythonRuntimeRootResolution {
56 ImportedModule,
59 LexicallyBound,
62 Unknown,
64}
65
66#[derive(Clone, Copy, Debug, Eq, PartialEq)]
68pub enum PythonRuntimeMutationEvidence {
69 NoKnownWrite,
70 KnownWrite,
71 KnownWriteAndUnknownEffects,
72 UnknownEffects,
73}
74
75#[derive(Clone, Copy, Debug, Eq, PartialEq)]
78pub enum PythonRuntimeAccessorCoverage {
79 NoKnownAccessorEffects,
80 UnknownAccessorEffects,
81}
82
83#[derive(Clone, Copy, Debug, Eq, PartialEq)]
85pub enum PythonRuntimeExecutionContext {
86 Module,
87 DeferredFunction(Range),
88}
89
90#[derive(Clone, Copy, Debug, Eq, PartialEq)]
92pub enum PythonRuntimeWriteTarget {
93 Container,
95 Key,
97 ContainerEscape,
100 ModuleEscape,
103}
104
105#[derive(Clone, Debug, Eq, PartialEq)]
107pub struct PythonRuntimeWrite {
108 pub root_name: String,
110 pub container: Option<String>,
113 pub range: Range,
114 pub target: PythonRuntimeWriteTarget,
115}
116
117#[derive(Clone, Debug, Eq, PartialEq)]
119pub struct PythonRuntimeRead {
120 pub binding_name: String,
122 pub root_name: String,
124 pub container: String,
125 pub access: PythonRuntimeAccessKey,
126 pub range: Range,
128 pub root_range: Range,
129 pub container_range: Range,
133 pub candidate_anchor: Range,
134 pub key_range: Range,
135 pub lexical_resolution: PythonRuntimeRootResolution,
136 pub mutation: PythonRuntimeMutationEvidence,
137 pub execution: PythonRuntimeExecutionContext,
138 pub accessor: PythonRuntimeAccessorCoverage,
142}
143
144#[derive(Clone, Debug, Eq, PartialEq)]
146pub struct PythonRuntimeReadFacts {
147 pub reads: Vec<PythonRuntimeRead>,
148 pub writes: Vec<PythonRuntimeWrite>,
149 pub visited_nodes: usize,
151 pub complete: bool,
154}
155
156#[derive(Debug)]
158struct PythonRuntimeBudget {
159 remaining: usize,
160 exhausted: bool,
161 visited_nodes: usize,
162}
163
164impl PythonRuntimeBudget {
165 fn for_facts(max_facts: usize) -> Self {
168 Self {
169 remaining: max_facts.saturating_mul(64).max(4096),
170 exhausted: false,
171 visited_nodes: 0,
172 }
173 }
174
175 fn visit(&mut self) -> bool {
176 if self.remaining == 0 {
177 self.exhausted = true;
178 return false;
179 }
180 self.remaining -= 1;
181 self.visited_nodes += 1;
182 true
183 }
184}
185
186struct ModuleBinding {
188 local_name: String,
189 root_module: String,
191 scope_start_byte: usize,
192 scope_end_byte: usize,
193 function_scoped: bool,
194}
195
196pub fn extract_python_runtime_reads(
202 root: Node<'_>,
203 source: &str,
204 max_facts: usize,
205) -> PythonRuntimeReadFacts {
206 let mut budget = PythonRuntimeBudget::for_facts(max_facts);
207 let bindings = module_bindings(root, source);
208 let mut reads = Vec::new();
209 let mut writes = Vec::new();
210 let mut stack = vec![root];
211 while let Some(node) = stack.pop() {
212 if !budget.visit() {
213 break;
214 }
215 if node.kind() == "attribute"
216 && reads.len() + writes.len() < max_facts
217 && let Some(path) = reviewed_attribute_path(node, source)
218 {
219 let resolution = root_resolution(
220 path.root,
221 &path.root_module,
222 &bindings,
223 root,
224 source,
225 &mut budget,
226 );
227 if resolution == PythonRuntimeRootResolution::ImportedModule {
228 classify_container_occurrence(
229 node,
230 &path,
231 source,
232 root,
233 &mut budget,
234 &mut reads,
235 &mut writes,
236 );
237 }
238 }
239 if node.kind() == "identifier"
240 && writes.len() + reads.len() < max_facts
241 && let Some(module) =
242 escaping_module_identifier(node, &bindings, root, source, &mut budget)
243 {
244 writes.push(PythonRuntimeWrite {
245 root_name: module,
246 container: None,
247 range: node_source_range(node),
248 target: PythonRuntimeWriteTarget::ModuleEscape,
249 });
250 }
251 let mut cursor = node.walk();
252 stack.extend(node.named_children(&mut cursor));
253 }
254 for read in &mut reads {
257 let known_write = writes.iter().any(|write| {
258 write.root_name == read.root_name
259 && write
260 .container
261 .as_ref()
262 .is_none_or(|container| *container == read.container)
263 });
264 read.mutation = match (known_write, read.accessor) {
265 (false, PythonRuntimeAccessorCoverage::NoKnownAccessorEffects) => {
266 PythonRuntimeMutationEvidence::NoKnownWrite
267 }
268 (false, PythonRuntimeAccessorCoverage::UnknownAccessorEffects) => {
269 PythonRuntimeMutationEvidence::UnknownEffects
270 }
271 (true, PythonRuntimeAccessorCoverage::NoKnownAccessorEffects) => {
272 PythonRuntimeMutationEvidence::KnownWrite
273 }
274 (true, PythonRuntimeAccessorCoverage::UnknownAccessorEffects) => {
275 PythonRuntimeMutationEvidence::KnownWriteAndUnknownEffects
276 }
277 };
278 }
279 reads.sort_by_key(|read| read.range.start_byte);
280 writes.sort_by_key(|write| write.range.start_byte);
281 PythonRuntimeReadFacts {
282 complete: !budget.exhausted && reads.len() + writes.len() < max_facts,
283 visited_nodes: budget.visited_nodes,
284 reads,
285 writes,
286 }
287}
288
289pub fn python_runtime_keyed_access_seed<'tree>(node: Node<'tree>) -> Option<Node<'tree>> {
297 if node.kind() != "subscript" {
298 return None;
299 }
300 let base = node.child_by_field_name("value")?;
301 if base.kind() != "attribute" {
302 return None;
303 }
304 let attribute = base.child_by_field_name("attribute")?;
305 let object = base.child_by_field_name("object")?;
306 if object.kind() != "identifier" || attribute.kind() != "identifier" {
307 return None;
308 }
309 Some(base)
310}
311
312struct ReviewedPath<'tree> {
314 root: Node<'tree>,
315 root_module: String,
316 container: String,
317}
318
319fn reviewed_attribute_path<'tree>(node: Node<'tree>, source: &str) -> Option<ReviewedPath<'tree>> {
322 let object = node.child_by_field_name("object")?;
323 let attribute = node.child_by_field_name("attribute")?;
324 if object.kind() != "identifier" || attribute.kind() != "identifier" {
325 return None;
326 }
327 let container = node_text(attribute, source);
328 let (root_module, _) = PROCESS_INPUT_CONTAINERS
329 .iter()
330 .find(|(_, reviewed)| *reviewed == container)?;
331 Some(ReviewedPath {
332 root: object,
333 root_module: (*root_module).to_owned(),
334 container: container.to_owned(),
335 })
336}
337
338fn classify_container_occurrence<'tree>(
341 attribute: Node<'tree>,
342 path: &ReviewedPath<'tree>,
343 source: &str,
344 module_root: Node<'tree>,
345 budget: &mut PythonRuntimeBudget,
346 reads: &mut Vec<PythonRuntimeRead>,
347 writes: &mut Vec<PythonRuntimeWrite>,
348) {
349 let write = |target| PythonRuntimeWrite {
350 root_name: path.root_module.clone(),
351 container: Some(path.container.clone()),
352 range: node_source_range(attribute),
353 target,
354 };
355 let Some(parent) = attribute.parent() else {
356 writes.push(write(PythonRuntimeWriteTarget::ContainerEscape));
357 return;
358 };
359 if parent.kind() != "subscript"
360 || parent
361 .child_by_field_name("value")
362 .is_none_or(|value| value.id() != attribute.id())
363 {
364 writes.push(write(if is_binding_target(attribute) {
369 PythonRuntimeWriteTarget::Container
370 } else {
371 PythonRuntimeWriteTarget::ContainerEscape
372 }));
373 return;
374 }
375 if is_binding_target(parent) {
376 writes.push(write(PythonRuntimeWriteTarget::Key));
377 return;
378 }
379 let Some(subscript) = parent.child_by_field_name("subscript") else {
380 writes.push(write(PythonRuntimeWriteTarget::ContainerEscape));
381 return;
382 };
383 let mut cursor = parent.walk();
386 let access = if parent
387 .children(&mut cursor)
388 .any(|child| child.kind() == ",")
389 {
390 PythonRuntimeAccessKey::Unsupported
391 } else {
392 access_key(subscript, source)
393 };
394 let read_range = node_source_range(parent);
395 let execution = match enclosing_execution_context(attribute) {
396 Some(function) => {
397 PythonRuntimeExecutionContext::DeferredFunction(node_source_range(function))
398 }
399 None => PythonRuntimeExecutionContext::Module,
400 };
401 let accessor = execution_context_accessor_coverage(attribute, module_root, read_range, budget);
402 reads.push(PythonRuntimeRead {
403 binding_name: node_text(path.root, source).to_owned(),
404 root_name: path.root_module.clone(),
405 container: path.container.clone(),
406 access,
407 range: read_range,
408 root_range: node_source_range(path.root),
409 container_range: node_source_range(attribute),
410 candidate_anchor: node_source_range(attribute),
411 key_range: node_source_range(subscript),
412 lexical_resolution: PythonRuntimeRootResolution::ImportedModule,
413 mutation: PythonRuntimeMutationEvidence::UnknownEffects,
415 execution,
416 accessor,
417 });
418}
419
420fn access_key(subscript: Node<'_>, source: &str) -> PythonRuntimeAccessKey {
422 if let Some(literal) = python_plain_string_literal(subscript, source) {
423 return PythonRuntimeAccessKey::Property(literal.to_owned());
424 }
425 match subscript.kind() {
426 "integer" => crate::structural::PYTHON_STRUCTURAL_SPEC
427 .integer_literal_value(subscript, source)
428 .and_then(|value| value.decimal)
429 .and_then(|decimal| decimal.parse::<u128>().ok())
430 .map_or(PythonRuntimeAccessKey::Unsupported, |index| {
431 PythonRuntimeAccessKey::Index(index)
432 }),
433 "slice" | "tuple" | "list_splat" => PythonRuntimeAccessKey::Unsupported,
434 "string" | "concatenated_string" => PythonRuntimeAccessKey::Unsupported,
435 _ => PythonRuntimeAccessKey::Dynamic,
436 }
437}
438
439fn is_binding_target(node: Node<'_>) -> bool {
442 let mut current = node;
443 while let Some(parent) = current.parent() {
444 match parent.kind() {
445 "assignment" | "augmented_assignment" | "for_statement" | "for_in_clause" => {
446 return parent
447 .child_by_field_name("left")
448 .is_some_and(|left| left.id() == current.id());
449 }
450 "delete_statement" => return true,
451 "pattern_list" | "tuple_pattern" | "list_pattern" | "list_splat_pattern" => {
452 current = parent;
453 }
454 _ => return false,
455 }
456 }
457 false
458}
459
460pub(crate) fn enclosing_execution_context<'tree>(node: Node<'tree>) -> Option<Node<'tree>> {
463 let mut current = node;
464 while let Some(parent) = current.parent() {
465 if matches!(parent.kind(), "function_definition" | "lambda")
466 && parent
467 .child_by_field_name("body")
468 .is_some_and(|body| covers(body, current))
469 {
470 return Some(parent);
471 }
472 current = parent;
473 }
474 None
475}
476
477fn execution_context_accessor_coverage(
484 attribute: Node<'_>,
485 module_root: Node<'_>,
486 read_range: Range,
487 budget: &mut PythonRuntimeBudget,
488) -> PythonRuntimeAccessorCoverage {
489 let scope = enclosing_execution_context(attribute)
490 .and_then(|function| function.child_by_field_name("body"))
491 .unwrap_or(module_root);
492 let mut stack = vec![scope];
493 while let Some(node) = stack.pop() {
494 if !budget.visit() {
495 return PythonRuntimeAccessorCoverage::UnknownAccessorEffects;
496 }
497 if matches!(node.kind(), "function_definition" | "lambda") {
498 let mut cursor = node.walk();
501 stack.extend(node.named_children(&mut cursor).filter(|child| {
502 Some(child.id()) != node.child_by_field_name("body").map(|body| body.id())
503 }));
504 continue;
505 }
506 if node.kind() == "call"
507 && python_hazard_can_precede_bounded(node_source_range(node), read_range, node, || {
508 budget.visit()
509 }) != Some(false)
510 {
511 return PythonRuntimeAccessorCoverage::UnknownAccessorEffects;
512 }
513 let mut cursor = node.walk();
514 stack.extend(node.named_children(&mut cursor));
515 }
516 PythonRuntimeAccessorCoverage::NoKnownAccessorEffects
517}
518
519pub(crate) fn python_hazard_can_precede_bounded(
529 hazard: Range,
530 read_range: Range,
531 hazard_node: Node<'_>,
532 mut scope_step: impl FnMut() -> bool,
533) -> Option<bool> {
534 if !scope_step() {
535 return None;
536 }
537 if hazard.end_byte <= read_range.start_byte {
538 return Some(true);
539 }
540 let mut current = hazard_node;
541 while let Some(parent) = current.parent() {
542 if !scope_step() {
543 return None;
544 }
545 if matches!(parent.kind(), "for_statement" | "while_statement")
546 && parent.start_byte() <= read_range.start_byte
547 && read_range.end_byte <= parent.end_byte()
548 {
549 return Some(true);
550 }
551 current = parent;
552 }
553 Some(false)
554}
555
556fn escaping_module_identifier(
562 node: Node<'_>,
563 bindings: &[ModuleBinding],
564 module_root: Node<'_>,
565 source: &str,
566 budget: &mut PythonRuntimeBudget,
567) -> Option<String> {
568 let name = node_text(node, source);
569 let module = binding_root_module(node, name, bindings)?;
570 if root_resolution(node, &module, bindings, module_root, source, budget)
574 != PythonRuntimeRootResolution::ImportedModule
575 {
576 return None;
577 }
578 let parent = node.parent()?;
579 if parent.kind() == "attribute"
580 && parent
581 .child_by_field_name("object")
582 .is_some_and(|object| object.id() == node.id())
583 {
584 return None;
585 }
586 if matches!(
587 parent.kind(),
588 "import_statement" | "import_from_statement" | "aliased_import" | "dotted_name"
589 ) {
590 return None;
591 }
592 Some(module)
593}
594
595fn binding_root_module(
598 reference: Node<'_>,
599 name: &str,
600 bindings: &[ModuleBinding],
601) -> Option<String> {
602 let offset = reference.start_byte();
603 let mut module = None;
604 for binding in bindings {
605 if binding.local_name != name
606 || offset < binding.scope_start_byte
607 || offset > binding.scope_end_byte
608 {
609 continue;
610 }
611 match &module {
612 None => module = Some(binding.root_module.clone()),
613 Some(existing) if *existing == binding.root_module => {}
614 Some(_) => return None,
615 }
616 }
617 module.filter(|module| {
618 PROCESS_INPUT_CONTAINERS
619 .iter()
620 .any(|(reviewed, _)| reviewed == module)
621 })
622}
623
624fn root_resolution(
631 root: Node<'_>,
632 expected_module: &str,
633 bindings: &[ModuleBinding],
634 module_root: Node<'_>,
635 source: &str,
636 budget: &mut PythonRuntimeBudget,
637) -> PythonRuntimeRootResolution {
638 let name = node_text(root, source);
639 match python_name_resolution_at(name, root, source) {
640 PythonLexicalNameResolution::Local | PythonLexicalNameResolution::Nonlocal => {
641 return if bindings
645 .iter()
646 .any(|binding| binding.local_name == name && binding.function_scoped)
647 {
648 PythonRuntimeRootResolution::Unknown
649 } else {
650 PythonRuntimeRootResolution::LexicallyBound
651 };
652 }
653 PythonLexicalNameResolution::Global | PythonLexicalNameResolution::Unbound => {}
654 }
655 match module_scope_binds_name_through_imports(module_root, name, source, budget) {
656 Some(true) => {}
657 Some(false) => return PythonRuntimeRootResolution::LexicallyBound,
658 None => return PythonRuntimeRootResolution::Unknown,
659 }
660 match binding_root_module(root, name, bindings) {
661 Some(module) if module == expected_module => PythonRuntimeRootResolution::ImportedModule,
662 Some(_) => PythonRuntimeRootResolution::LexicallyBound,
663 None => PythonRuntimeRootResolution::Unknown,
664 }
665}
666
667fn module_scope_binds_name_through_imports(
673 module_root: Node<'_>,
674 name: &str,
675 source: &str,
676 budget: &mut PythonRuntimeBudget,
677) -> Option<bool> {
678 let mut imported = false;
679 let mut stack = vec![module_root];
680 while let Some(node) = stack.pop() {
681 if !budget.visit() {
682 return None;
683 }
684 if node.kind() == "wildcard_import" {
685 return None;
686 }
687 for binding in python_direct_scope_bindings_bounded(node, source, || budget.visit())? {
688 if node_text(binding.declaration, source) != name {
689 continue;
690 }
691 match binding.kind {
692 PythonDirectScopeBindingKind::Import => imported = true,
693 _ => return Some(false),
694 }
695 }
696 stack.extend(python_executable_scope_children(node));
699 }
700 Some(imported)
701}
702
703fn module_bindings(root: Node<'_>, source: &str) -> Vec<ModuleBinding> {
710 let mut bindings = Vec::new();
711 for binding in python_import_bindings_from_tree(root, source) {
712 let Some(root_module) = binding.qualified_name.split('.').next() else {
713 continue;
714 };
715 if !PROCESS_INPUT_CONTAINERS
716 .iter()
717 .any(|(reviewed, _)| *reviewed == root_module)
718 {
719 continue;
720 }
721 if binding.qualified_name != root_module && binding.consumed_attributes == 0 {
723 continue;
724 }
725 bindings.push(ModuleBinding {
726 local_name: binding.local_name.clone(),
727 root_module: root_module.to_owned(),
728 scope_start_byte: binding.scope_start_byte,
729 scope_end_byte: binding.scope_end_byte,
730 function_scoped: binding.is_function_scoped(),
731 });
732 }
733 bindings
734}
735
736use brokk_bifrost_core::analyzer::usages::reference_site::node_range as node_source_range;
737
738fn node_text<'a>(node: Node<'_>, source: &'a str) -> &'a str {
739 brokk_bifrost_core::analyzer::common::try_node_source_text(node, source)
740 .expect("a tree-sitter node range is valid UTF-8 source")
741}
742
743#[derive(Clone, Debug, Eq, PartialEq)]
745pub enum PythonModuleMemberMutation {
746 Write(Range),
748 Escape(Range),
751 Budget,
753}
754
755pub fn python_module_member_mutation(
782 root: Node<'_>,
783 source: &str,
784 source_module: &str,
785 module: &str,
786 member: &str,
787 max_nodes: usize,
788) -> Option<PythonModuleMemberMutation> {
789 let mut budget = PythonRuntimeBudget {
790 remaining: max_nodes,
791 exhausted: false,
792 visited_nodes: 0,
793 };
794 let bindings = python_import_bindings_from_tree(root, source);
795 let Some(facts) =
796 python_sys_modules_aliases(root, source, source_module, module, &bindings, &mut budget)
797 else {
798 return Some(PythonModuleMemberMutation::Budget);
799 };
800 let table_aliases = &facts.aliases;
801 let locals: Vec<String> = bindings
802 .iter()
803 .filter(|binding| {
804 binding.qualified_name == module
805 || binding
806 .qualified_name
807 .split('.')
808 .next()
809 .is_some_and(|root_module| {
810 root_module == module && binding.consumed_attributes > 0
811 })
812 })
813 .map(|binding| binding.local_name.clone())
814 .collect();
815 let mut stack = vec![root];
816 while let Some(node) = stack.pop() {
817 if !budget.visit() {
818 return Some(PythonModuleMemberMutation::Budget);
819 }
820 if node.kind() == "call"
821 && python_mock_patch_may_replace_module_member(node, source, &bindings, module, member)
822 {
823 return Some(PythonModuleMemberMutation::Write(node_source_range(node)));
824 }
825 if node.kind() == "identifier" {
826 let name = node_text(node, source);
827 if let Some(table) = python_sys_modules_table(node, source) {
828 if python_imported_path(table, source, &bindings)
829 .is_some_and(|path| path == ["sys", "modules"])
830 {
831 match python_sys_modules_lookup(
832 table,
833 source,
834 source_module,
835 module,
836 &bindings,
837 &facts,
838 &mut budget,
839 ) {
840 PythonSysModulesLookup::Unrelated => {}
841 PythonSysModulesLookup::Escape(range) => {
842 return Some(PythonModuleMemberMutation::Escape(range));
843 }
844 PythonSysModulesLookup::Target(read) => {
845 if let Some(mutation) =
846 python_sys_modules_read_use(read, source, member, table_aliases)
847 {
848 return Some(mutation);
849 }
850 }
851 }
852 } else {
853 return Some(PythonModuleMemberMutation::Escape(node_source_range(node)));
854 }
855 }
856 let table_alias = table_aliases.contains(name);
857 if table_alias && python_module_table_alias_binding_use(node, source, table_aliases) {
858 } else if (table_alias
860 || (locals.iter().any(|local| local == name)
861 && !matches!(
862 python_name_resolution_at(name, node, source),
863 PythonLexicalNameResolution::Local | PythonLexicalNameResolution::Nonlocal
864 )))
865 && let Some(mutation) =
866 module_binding_use(node, source, &bindings, table_aliases, module, member)
867 {
868 return Some(mutation);
869 }
870 }
871 let mut cursor = node.walk();
872 stack.extend(node.named_children(&mut cursor));
873 }
874 None
875}
876
877enum PythonSysModulesLookup<'tree> {
878 Unrelated,
879 Target(Node<'tree>),
880 Escape(Range),
881}
882
883fn python_sys_modules_table<'tree>(node: Node<'tree>, source: &str) -> Option<Node<'tree>> {
885 let table = node.parent()?;
886 (table.kind() == "attribute"
887 && table
888 .child_by_field_name("object")
889 .is_some_and(|object| object.id() == node.id())
890 && table
891 .child_by_field_name("attribute")
892 .is_some_and(|attribute| node_text(attribute, source) == "modules"))
893 .then_some(table)
894}
895
896fn python_sys_modules_read_table<'tree>(read: Node<'tree>, source: &str) -> Option<Node<'tree>> {
897 if read.kind() == "subscript" {
898 let table = read.child_by_field_name("value")?;
899 return (table.kind() == "attribute"
900 && table
901 .child_by_field_name("attribute")
902 .is_some_and(|attribute| node_text(attribute, source) == "modules"))
903 .then_some(table);
904 }
905 if read.kind() == "call" {
906 let function = read.child_by_field_name("function")?;
907 if function.kind() != "attribute"
908 || function
909 .child_by_field_name("attribute")
910 .is_none_or(|attribute| {
911 !matches!(node_text(attribute, source), "get" | "pop" | "setdefault")
912 })
913 {
914 return None;
915 }
916 let table = function.child_by_field_name("object")?;
917 return (table.kind() == "attribute"
918 && table
919 .child_by_field_name("attribute")
920 .is_some_and(|attribute| node_text(attribute, source) == "modules"))
921 .then_some(table);
922 }
923 None
924}
925
926fn python_sys_modules_lookup<'tree>(
927 table: Node<'tree>,
928 source: &str,
929 source_module: &str,
930 module: &str,
931 bindings: &[crate::imports::PythonImportBinding],
932 facts: &PythonSysModulesFacts<'tree>,
933 budget: &mut PythonRuntimeBudget,
934) -> PythonSysModulesLookup<'tree> {
935 let escaped = || PythonSysModulesLookup::Escape(node_source_range(table));
939 if python_imported_path(table, source, bindings).is_none_or(|path| path != ["sys", "modules"]) {
940 return escaped();
941 }
942 let Some(parent) = table.parent() else {
943 return escaped();
944 };
945 let proof = PythonSysModulesProofContext {
946 source,
947 source_module,
948 module,
949 bindings,
950 facts,
951 };
952 if parent.kind() == "subscript"
953 && parent
954 .child_by_field_name("value")
955 .is_some_and(|value| value.id() == table.id())
956 {
957 if is_binding_target(parent) {
958 if parent
959 .parent()
960 .is_some_and(|parent| parent.kind() == "delete_statement")
961 {
962 return PythonSysModulesLookup::Unrelated;
963 }
964 let Some(value) = parent
965 .parent()
966 .filter(|assignment| assignment.kind() == "assignment")
967 .and_then(|assignment| assignment.child_by_field_name("right"))
968 else {
969 return escaped();
970 };
971 return match python_sys_modules_stored_value(
972 value,
973 &proof,
974 budget,
975 &mut HashSet::default(),
976 ) {
977 PythonSysModulesStoredValue::Unrelated => PythonSysModulesLookup::Unrelated,
978 PythonSysModulesStoredValue::Target | PythonSysModulesStoredValue::Unknown => {
979 escaped()
980 }
981 };
982 }
983 return python_sys_modules_read_identity(
984 parent.child_by_field_name("subscript"),
985 parent,
986 &proof,
987 budget,
988 );
989 }
990 if parent.kind() == "attribute"
991 && parent
992 .child_by_field_name("object")
993 .is_some_and(|object| object.id() == table.id())
994 {
995 let Some(call) = parent.parent().filter(|call| {
996 call.kind() == "call"
997 && call
998 .child_by_field_name("function")
999 .is_some_and(|function| function.id() == parent.id())
1000 }) else {
1001 return escaped();
1002 };
1003 let method = parent
1004 .child_by_field_name("attribute")
1005 .map(|attribute| node_text(attribute, source));
1006 match method {
1007 Some("get" | "pop") => {
1008 let args = python_sys_modules_arguments(call);
1009 return python_sys_modules_read_identity(
1010 args.first().copied(),
1011 call,
1012 &proof,
1013 budget,
1014 );
1015 }
1016 Some("setdefault") => {
1017 let args = python_sys_modules_arguments(call);
1018 let default = args
1019 .get(1)
1020 .copied()
1021 .or_else(|| python_sys_modules_keyword_argument(call, source, "default"));
1022 if let Some(default) = default
1023 && python_sys_modules_stored_value(
1024 default,
1025 &proof,
1026 budget,
1027 &mut HashSet::default(),
1028 ) != PythonSysModulesStoredValue::Unrelated
1029 {
1030 return escaped();
1031 }
1032 return python_sys_modules_read_identity(
1033 args.first().copied(),
1034 call,
1035 &proof,
1036 budget,
1037 );
1038 }
1039 Some("update") => {
1040 let Some(arguments) = call.child_by_field_name("arguments") else {
1041 return escaped();
1042 };
1043 let mut cursor = arguments.walk();
1044 for argument in arguments.named_children(&mut cursor) {
1045 if python_sys_modules_update_argument(argument, &proof, budget)
1046 != PythonSysModulesStoredValue::Unrelated
1047 {
1048 return escaped();
1049 }
1050 }
1051 return PythonSysModulesLookup::Unrelated;
1052 }
1053 Some("clear" | "keys") => return PythonSysModulesLookup::Unrelated,
1054 _ => return escaped(),
1055 }
1056 }
1057 if parent.kind() == "argument_list"
1058 && let Some(call) = parent.parent().filter(|call| call.kind() == "call")
1059 && python_sys_modules_arguments(call)
1060 .first()
1061 .is_some_and(|first| first.id() == table.id())
1062 {
1063 match python_test_helper_call(call, source, bindings) {
1066 Some(PythonTestHelperCall::MonkeyPatchSetItem(receiver))
1067 if python_monkeypatch_receiver(receiver, &proof) =>
1068 {
1069 let value = python_sys_modules_arguments(call)
1070 .get(2)
1071 .copied()
1072 .or_else(|| python_sys_modules_keyword_argument(call, source, "value"));
1073 return match value.map(|value| {
1074 python_sys_modules_stored_value(value, &proof, budget, &mut HashSet::default())
1075 }) {
1076 Some(PythonSysModulesStoredValue::Unrelated) => {
1077 PythonSysModulesLookup::Unrelated
1078 }
1079 _ => escaped(),
1080 };
1081 }
1082 Some(PythonTestHelperCall::MonkeyPatchDelItem(receiver))
1083 if python_monkeypatch_receiver(receiver, &proof) =>
1084 {
1085 return PythonSysModulesLookup::Unrelated;
1086 }
1087 Some(PythonTestHelperCall::MockPatchDict) => {
1088 let Some(arguments) = call.child_by_field_name("arguments") else {
1089 return escaped();
1090 };
1091 let mut cursor = arguments.walk();
1092 for argument in arguments.named_children(&mut cursor) {
1093 let keyword = (argument.kind() == "keyword_argument")
1094 .then(|| argument.child_by_field_name("name"))
1095 .flatten()
1096 .map(|name| node_text(name, source));
1097 let identity = if argument.id() == table.id() || keyword == Some("clear") {
1098 PythonSysModulesStoredValue::Unrelated
1099 } else if keyword == Some("values") {
1100 argument
1101 .child_by_field_name("value")
1102 .map(|value| {
1103 python_sys_modules_update_values(
1104 value,
1105 &proof,
1106 budget,
1107 &mut HashSet::default(),
1108 )
1109 })
1110 .unwrap_or(PythonSysModulesStoredValue::Unknown)
1111 } else {
1112 python_sys_modules_update_argument(argument, &proof, budget)
1113 };
1114 if identity != PythonSysModulesStoredValue::Unrelated {
1115 return escaped();
1116 }
1117 }
1118 return PythonSysModulesLookup::Unrelated;
1119 }
1120 _ => {}
1121 }
1122 }
1123 escaped()
1124}
1125
1126fn python_sys_modules_update_argument<'tree>(
1129 argument: Node<'tree>,
1130 proof: &PythonSysModulesProofContext<'_, 'tree>,
1131 budget: &mut PythonRuntimeBudget,
1132) -> PythonSysModulesStoredValue {
1133 if argument.kind() == "keyword_argument" {
1134 argument
1135 .child_by_field_name("value")
1136 .map(|value| {
1137 python_sys_modules_stored_value(value, proof, budget, &mut HashSet::default())
1138 })
1139 .unwrap_or(PythonSysModulesStoredValue::Unknown)
1140 } else if matches!(argument.kind(), "dictionary_splat" | "list_splat") {
1141 PythonSysModulesStoredValue::Unknown
1142 } else {
1143 python_sys_modules_update_values(argument, proof, budget, &mut HashSet::default())
1144 }
1145}
1146
1147fn python_monkeypatch_receiver(
1152 receiver: Node<'_>,
1153 proof: &PythonSysModulesProofContext<'_, '_>,
1154) -> bool {
1155 let source = proof.source;
1156 let is_monkeypatch_class = |expression: Option<Node<'_>>| {
1157 expression
1158 .and_then(|expression| python_imported_path(expression, source, proof.bindings))
1159 .is_some_and(|path| path.last().is_some_and(|name| name == "MonkeyPatch"))
1160 };
1161 if receiver.kind() != "identifier" {
1162 return false;
1163 }
1164 let name = node_text(receiver, source);
1165 let scope = python_alias_scope(receiver);
1166 if scope.kind() != "function_definition"
1167 || !python_sys_modules_has_one_binding(proof, scope, name)
1168 {
1169 return false;
1170 }
1171 if let Some(parameters) = scope.child_by_field_name("parameters") {
1172 let mut cursor = parameters.walk();
1173 if let Some(parameter) = parameters
1174 .named_children(&mut cursor)
1175 .find(|parameter| python_parameter_name(*parameter, source).as_deref() == Some(name))
1176 {
1177 return name == "monkeypatch"
1178 || is_monkeypatch_class(
1179 parameter
1180 .child_by_field_name("type")
1181 .and_then(|annotation| annotation.named_child(0)),
1182 );
1183 }
1184 }
1185 let key = (scope.start_byte(), scope.end_byte(), name.to_owned());
1186 proof
1187 .facts
1188 .assignments
1189 .get(&key)
1190 .is_some_and(|values| match values.as_slice() {
1191 [value] => {
1192 value.kind() == "call"
1193 && is_monkeypatch_class(value.child_by_field_name("function"))
1194 }
1195 _ => false,
1196 })
1197}
1198
1199fn python_sys_modules_arguments(call: Node<'_>) -> Vec<Node<'_>> {
1200 let Some(arguments) = call.child_by_field_name("arguments") else {
1201 return Vec::new();
1202 };
1203 let mut cursor = arguments.walk();
1204 arguments
1205 .named_children(&mut cursor)
1206 .filter(|argument| argument.kind() != "keyword_argument")
1207 .collect()
1208}
1209
1210fn python_sys_modules_keyword_argument<'tree>(
1211 call: Node<'tree>,
1212 source: &str,
1213 name: &str,
1214) -> Option<Node<'tree>> {
1215 let arguments = call.child_by_field_name("arguments")?;
1216 let mut cursor = arguments.walk();
1217 arguments
1218 .named_children(&mut cursor)
1219 .find(|argument| {
1220 argument.kind() == "keyword_argument"
1221 && argument
1222 .child_by_field_name("name")
1223 .is_some_and(|key| node_text(key, source) == name)
1224 })
1225 .and_then(|argument| argument.child_by_field_name("value"))
1226}
1227
1228fn python_sys_modules_read_identity<'tree>(
1229 key: Option<Node<'tree>>,
1230 read: Node<'tree>,
1231 proof: &PythonSysModulesProofContext<'_, 'tree>,
1232 budget: &mut PythonRuntimeBudget,
1233) -> PythonSysModulesLookup<'tree> {
1234 let Some(key) = key else {
1235 return PythonSysModulesLookup::Escape(node_source_range(read));
1236 };
1237 if key.kind() == "identifier" && node_text(key, proof.source) == "__name__" {
1238 return if proof.source_module == proof.module {
1239 PythonSysModulesLookup::Target(read)
1240 } else {
1241 PythonSysModulesLookup::Unrelated
1242 };
1243 }
1244 let Some(key_value) = crate::decode_python_string_value(key, proof.source).ok() else {
1245 if python_sys_modules_guard_excludes_module(key, read, proof, budget) {
1246 return PythonSysModulesLookup::Unrelated;
1247 }
1248 return PythonSysModulesLookup::Target(read);
1249 };
1250 if key_value == proof.module {
1251 PythonSysModulesLookup::Target(read)
1252 } else {
1253 PythonSysModulesLookup::Unrelated
1254 }
1255}
1256
1257#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1258enum PythonGuardTruth {
1259 True,
1260 False,
1261 Unknown,
1262}
1263
1264fn python_sys_modules_guard_excludes_module(
1267 key: Node<'_>,
1268 read: Node<'_>,
1269 proof: &PythonSysModulesProofContext<'_, '_>,
1270 budget: &mut PythonRuntimeBudget,
1271) -> bool {
1272 if key.kind() != "identifier" {
1273 return false;
1274 }
1275 let key_name = node_text(key, proof.source);
1276 let mut child = read;
1277 let mut ancestor = read.parent();
1278 while let Some(node) = ancestor {
1279 if !budget.visit() {
1280 return false;
1281 }
1282 if matches!(node.kind(), "if_statement" | "elif_clause")
1283 && let (Some(condition), Some(consequence)) = (
1284 node.child_by_field_name("condition"),
1285 node.child_by_field_name("consequence"),
1286 )
1287 && consequence.start_byte() <= read.start_byte()
1288 && read.end_byte() <= consequence.end_byte()
1289 && python_sys_modules_key_binding_unchanged(key, condition, read, proof, budget)
1290 && python_sys_modules_guard_truth(
1291 condition,
1292 key_name,
1293 proof.module,
1294 proof.source,
1295 budget,
1296 ) == PythonGuardTruth::False
1297 {
1298 return true;
1299 }
1300 if node.kind() == "block"
1301 && let Some(statement) = python_sys_modules_direct_child_containing(node, child)
1302 && python_sys_modules_prior_continue_guard_excludes(
1303 node, statement, key, read, key_name, proof, budget,
1304 )
1305 {
1306 return true;
1307 }
1308 child = node;
1309 ancestor = node.parent();
1310 }
1311 false
1312}
1313
1314fn python_sys_modules_key_binding_unchanged(
1319 key: Node<'_>,
1320 guard: Node<'_>,
1321 read: Node<'_>,
1322 proof: &PythonSysModulesProofContext<'_, '_>,
1323 budget: &mut PythonRuntimeBudget,
1324) -> bool {
1325 let name = node_text(key, proof.source);
1326 let guard_reference = python_sys_modules_guard_key_reference(guard, name, proof.source, budget);
1327 let Some(guard_reference) = guard_reference else {
1328 return false;
1329 };
1330 let scope = python_alias_scope(key);
1331 let key_resolution = python_name_resolution_at(name, key, proof.source);
1332 let guard_resolution = python_name_resolution_at(name, guard_reference, proof.source);
1333 let same_lexical_binding = key_resolution == guard_resolution
1334 && (key_resolution == PythonLexicalNameResolution::Local
1335 || (scope.kind() == "module"
1336 && key_resolution == PythonLexicalNameResolution::Unbound));
1337 if python_alias_scope(guard_reference).id() != scope.id()
1338 || python_alias_scope(read).id() != scope.id()
1339 || !same_lexical_binding
1340 {
1341 return false;
1342 }
1343 if !python_sys_modules_has_one_binding(proof, scope, name) {
1344 return false;
1345 }
1346 let start = guard.start_byte();
1347 let guard_end = guard.end_byte();
1348 let end = read.start_byte();
1349 let mut stack = vec![scope.child_by_field_name("body").unwrap_or(scope)];
1350 while let Some(node) = stack.pop() {
1351 if !budget.visit() {
1352 return false;
1353 }
1354 if node.start_byte() >= end || node.end_byte() <= start {
1355 continue;
1356 }
1357 if python_direct_scope_bindings_bounded(node, proof.source, || budget.visit()).is_none_or(
1358 |bindings| {
1359 bindings.into_iter().any(|binding| {
1360 node_text(binding.declaration, proof.source) == name
1361 && start <= binding.declaration.start_byte()
1362 && binding.declaration.start_byte() < end
1363 })
1364 },
1365 ) {
1366 return false;
1367 }
1368 let mut cursor = node.walk();
1369 for child in node.named_children(&mut cursor) {
1370 if matches!(child.kind(), "call")
1371 && guard_end <= child.start_byte()
1372 && child.end_byte() < end
1373 {
1374 return false;
1375 }
1376 }
1377 stack.extend(python_executable_scope_children(node));
1378 }
1379 true
1380}
1381
1382fn python_sys_modules_guard_key_reference<'tree>(
1383 condition: Node<'tree>,
1384 name: &str,
1385 source: &str,
1386 budget: &mut PythonRuntimeBudget,
1387) -> Option<Node<'tree>> {
1388 let mut stack = vec![condition];
1389 while let Some(node) = stack.pop() {
1390 if !budget.visit() {
1391 return None;
1392 }
1393 if node.kind() == "identifier" && node_text(node, source) == name {
1394 return Some(node);
1395 }
1396 let mut cursor = node.walk();
1397 stack.extend(node.named_children(&mut cursor));
1398 }
1399 None
1400}
1401
1402fn python_sys_modules_direct_child_containing<'tree>(
1403 block: Node<'tree>,
1404 child: Node<'tree>,
1405) -> Option<Node<'tree>> {
1406 let mut direct = child;
1407 while direct
1408 .parent()
1409 .is_some_and(|parent| parent.id() != block.id())
1410 {
1411 direct = direct.parent()?;
1412 }
1413 direct
1414 .parent()
1415 .is_some_and(|parent| parent.id() == block.id())
1416 .then_some(direct)
1417}
1418
1419fn python_sys_modules_prior_continue_guard_excludes(
1420 block: Node<'_>,
1421 read_statement: Node<'_>,
1422 key: Node<'_>,
1423 read: Node<'_>,
1424 key_name: &str,
1425 proof: &PythonSysModulesProofContext<'_, '_>,
1426 budget: &mut PythonRuntimeBudget,
1427) -> bool {
1428 let mut cursor = block.walk();
1429 for statement in block.named_children(&mut cursor) {
1430 if statement.id() == read_statement.id() {
1431 break;
1432 }
1433 if statement.kind() != "if_statement" {
1434 continue;
1435 }
1436 let Some(condition) = statement.child_by_field_name("condition") else {
1437 continue;
1438 };
1439 if condition.kind() != "not_operator"
1440 || !python_sys_modules_guard_has_direct_abrupt_consequence(statement)
1441 || !python_sys_modules_key_binding_unchanged(key, condition, read, proof, budget)
1442 {
1443 continue;
1444 }
1445 let Some(guard_condition) = condition.named_child(0) else {
1446 continue;
1447 };
1448 if python_sys_modules_guard_truth(
1449 guard_condition,
1450 key_name,
1451 proof.module,
1452 proof.source,
1453 budget,
1454 ) == PythonGuardTruth::False
1455 {
1456 return true;
1457 }
1458 }
1459 false
1460}
1461
1462fn python_sys_modules_guard_has_direct_abrupt_consequence(statement: Node<'_>) -> bool {
1463 let Some(consequence) = statement.child_by_field_name("consequence") else {
1464 return false;
1465 };
1466 let mut cursor = consequence.walk();
1467 let mut statements = consequence.named_children(&mut cursor);
1468 let Some(only) = statements.next() else {
1469 return false;
1470 };
1471 statements.next().is_none()
1472 && matches!(
1473 only.kind(),
1474 "break_statement" | "continue_statement" | "return_statement" | "raise_statement"
1475 )
1476}
1477
1478fn python_sys_modules_guard_truth(
1479 condition: Node<'_>,
1480 key_name: &str,
1481 module: &str,
1482 source: &str,
1483 budget: &mut PythonRuntimeBudget,
1484) -> PythonGuardTruth {
1485 let mut pending = vec![(condition, false)];
1486 let mut results = HashMap::<usize, PythonGuardTruth>::default();
1487 while let Some((node, visited)) = pending.pop() {
1488 if !budget.visit() {
1489 return PythonGuardTruth::Unknown;
1490 }
1491 if !visited
1492 && matches!(
1493 node.kind(),
1494 "boolean_operator" | "not_operator" | "parenthesized_expression"
1495 )
1496 {
1497 pending.push((node, true));
1498 let mut cursor = node.walk();
1499 for child in node.named_children(&mut cursor) {
1500 pending.push((child, false));
1501 }
1502 continue;
1503 }
1504 let value = match node.kind() {
1505 "boolean_operator" => {
1506 let Some(left) = node.child_by_field_name("left") else {
1507 return PythonGuardTruth::Unknown;
1508 };
1509 let Some(right) = node.child_by_field_name("right") else {
1510 return PythonGuardTruth::Unknown;
1511 };
1512 let Some(operator) = node.child_by_field_name("operator") else {
1513 return PythonGuardTruth::Unknown;
1514 };
1515 let (Some(left), Some(right)) = (results.get(&left.id()), results.get(&right.id()))
1516 else {
1517 return PythonGuardTruth::Unknown;
1518 };
1519 match node_text(operator, source) {
1520 "and" => python_guard_and(*left, *right),
1521 "or" => python_guard_or(*left, *right),
1522 _ => PythonGuardTruth::Unknown,
1523 }
1524 }
1525 "not_operator" => node
1526 .named_child(0)
1527 .and_then(|child| results.get(&child.id()).copied())
1528 .map(python_guard_not)
1529 .unwrap_or(PythonGuardTruth::Unknown),
1530 "parenthesized_expression" => node
1531 .named_child(0)
1532 .and_then(|child| results.get(&child.id()).copied())
1533 .unwrap_or(PythonGuardTruth::Unknown),
1534 _ => python_sys_modules_guard_atom(node, key_name, module, source),
1535 };
1536 results.insert(node.id(), value);
1537 }
1538 results
1539 .get(&condition.id())
1540 .copied()
1541 .unwrap_or(PythonGuardTruth::Unknown)
1542}
1543
1544fn python_guard_and(left: PythonGuardTruth, right: PythonGuardTruth) -> PythonGuardTruth {
1545 match (left, right) {
1546 (PythonGuardTruth::False, _) | (_, PythonGuardTruth::False) => PythonGuardTruth::False,
1547 (PythonGuardTruth::True, PythonGuardTruth::True) => PythonGuardTruth::True,
1548 _ => PythonGuardTruth::Unknown,
1549 }
1550}
1551
1552fn python_guard_or(left: PythonGuardTruth, right: PythonGuardTruth) -> PythonGuardTruth {
1553 match (left, right) {
1554 (PythonGuardTruth::True, _) | (_, PythonGuardTruth::True) => PythonGuardTruth::True,
1555 (PythonGuardTruth::False, PythonGuardTruth::False) => PythonGuardTruth::False,
1556 _ => PythonGuardTruth::Unknown,
1557 }
1558}
1559
1560fn python_guard_not(value: PythonGuardTruth) -> PythonGuardTruth {
1561 match value {
1562 PythonGuardTruth::True => PythonGuardTruth::False,
1563 PythonGuardTruth::False => PythonGuardTruth::True,
1564 PythonGuardTruth::Unknown => PythonGuardTruth::Unknown,
1565 }
1566}
1567
1568fn python_sys_modules_guard_atom(
1569 node: Node<'_>,
1570 key_name: &str,
1571 module: &str,
1572 source: &str,
1573) -> PythonGuardTruth {
1574 if node.kind() == "comparison_operator" {
1575 return python_sys_modules_guard_comparison(node, key_name, module, source);
1576 }
1577 if node.kind() == "call" {
1578 return python_sys_modules_guard_string_method(node, key_name, module, source);
1579 }
1580 PythonGuardTruth::Unknown
1581}
1582
1583fn python_sys_modules_guard_comparison(
1584 node: Node<'_>,
1585 key_name: &str,
1586 module: &str,
1587 source: &str,
1588) -> PythonGuardTruth {
1589 let mut cursor = node.walk();
1590 let operands = node.named_children(&mut cursor).collect::<Vec<_>>();
1591 let mut cursor = node.walk();
1592 let operators = node
1593 .children_by_field_name("operators", &mut cursor)
1594 .collect::<Vec<_>>();
1595 let ([left, right], [operator]) = (operands.as_slice(), operators.as_slice()) else {
1596 return PythonGuardTruth::Unknown;
1597 };
1598 let operator = node_text(*operator, source);
1599 let key_left = left.kind() == "identifier" && node_text(*left, source) == key_name;
1600 let key_right = right.kind() == "identifier" && node_text(*right, source) == key_name;
1601 if key_left || key_right {
1602 let literal = if key_left {
1603 crate::decode_python_string_value(*right, source)
1604 } else {
1605 crate::decode_python_string_value(*left, source)
1606 };
1607 if matches!(operator, "==" | "!=")
1608 && let Ok(literal) = literal
1609 {
1610 let matches = literal == module;
1611 return python_guard_bool(if operator == "==" { matches } else { !matches });
1612 }
1613 if operator == "in" && key_left {
1614 return python_sys_modules_literal_membership(*right, module, source);
1615 }
1616 if operator == "in"
1617 && key_right
1618 && let Ok(literal) = literal
1619 {
1620 return python_guard_bool(module.contains(&literal));
1621 }
1622 }
1623 PythonGuardTruth::Unknown
1624}
1625
1626fn python_sys_modules_literal_membership(
1627 collection: Node<'_>,
1628 module: &str,
1629 source: &str,
1630) -> PythonGuardTruth {
1631 if !matches!(collection.kind(), "tuple" | "list" | "set") {
1632 return PythonGuardTruth::Unknown;
1633 }
1634 let mut cursor = collection.walk();
1635 let mut contains = false;
1636 for value in collection.named_children(&mut cursor) {
1637 let Ok(literal) = crate::decode_python_string_value(value, source) else {
1638 return PythonGuardTruth::Unknown;
1639 };
1640 contains |= literal == module;
1641 }
1642 python_guard_bool(contains)
1643}
1644
1645fn python_sys_modules_guard_string_method(
1646 call: Node<'_>,
1647 key_name: &str,
1648 module: &str,
1649 source: &str,
1650) -> PythonGuardTruth {
1651 let Some(function) = call.child_by_field_name("function") else {
1652 return PythonGuardTruth::Unknown;
1653 };
1654 if function.kind() != "attribute"
1655 || !function
1656 .child_by_field_name("object")
1657 .is_some_and(|object| {
1658 object.kind() == "identifier" && node_text(object, source) == key_name
1659 })
1660 {
1661 return PythonGuardTruth::Unknown;
1662 }
1663 let Some(method) = function
1664 .child_by_field_name("attribute")
1665 .map(|name| node_text(name, source))
1666 else {
1667 return PythonGuardTruth::Unknown;
1668 };
1669 let Some(arguments) = call.child_by_field_name("arguments") else {
1670 return PythonGuardTruth::Unknown;
1671 };
1672 let mut cursor = arguments.walk();
1673 let mut values = arguments.named_children(&mut cursor);
1674 let Some(first) = values.next() else {
1675 return PythonGuardTruth::Unknown;
1676 };
1677 if values.next().is_some() || first.kind() == "keyword_argument" {
1678 return PythonGuardTruth::Unknown;
1679 }
1680 let matches = if let Ok(literal) = crate::decode_python_string_value(first, source) {
1681 match method {
1682 "startswith" => module.starts_with(&literal),
1683 "endswith" => module.ends_with(&literal),
1684 _ => return PythonGuardTruth::Unknown,
1685 }
1686 } else if matches!(first.kind(), "tuple" | "list" | "set") {
1687 let mut cursor = first.walk();
1688 let mut matched = false;
1689 for value in first.named_children(&mut cursor) {
1690 let Ok(literal) = crate::decode_python_string_value(value, source) else {
1691 return PythonGuardTruth::Unknown;
1692 };
1693 matched |= match method {
1694 "startswith" => module.starts_with(&literal),
1695 "endswith" => module.ends_with(&literal),
1696 _ => return PythonGuardTruth::Unknown,
1697 };
1698 }
1699 matched
1700 } else {
1701 return PythonGuardTruth::Unknown;
1702 };
1703 python_guard_bool(matches)
1704}
1705
1706fn python_guard_bool(value: bool) -> PythonGuardTruth {
1707 if value {
1708 PythonGuardTruth::True
1709 } else {
1710 PythonGuardTruth::False
1711 }
1712}
1713
1714#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1715enum PythonSysModulesStoredValue {
1716 Target,
1717 Unrelated,
1718 Unknown,
1719}
1720
1721struct PythonSysModulesFacts<'tree> {
1722 aliases: HashSet<String>,
1723 assignments: HashMap<(usize, usize, String), Vec<Node<'tree>>>,
1724 binding_counts: HashMap<(usize, usize, String), usize>,
1725 classes: Vec<PythonExternalClassAlias>,
1726}
1727
1728struct PythonSysModulesProofContext<'a, 'tree> {
1729 source: &'a str,
1730 source_module: &'a str,
1731 module: &'a str,
1732 bindings: &'a [crate::imports::PythonImportBinding],
1733 facts: &'a PythonSysModulesFacts<'tree>,
1734}
1735
1736fn python_sys_modules_has_one_binding(
1737 proof: &PythonSysModulesProofContext<'_, '_>,
1738 scope: Node<'_>,
1739 name: &str,
1740) -> bool {
1741 let count = proof
1742 .facts
1743 .binding_counts
1744 .get(&(scope.start_byte(), scope.end_byte(), name.to_owned()))
1745 .copied()
1746 .unwrap_or_default()
1747 + proof
1748 .bindings
1749 .iter()
1750 .filter(|binding| {
1751 binding.local_name == name
1752 && binding.scope_start_byte == scope.start_byte()
1753 && binding.scope_end_byte == scope.end_byte()
1754 })
1755 .count();
1756 count == 1
1757}
1758
1759fn python_sys_modules_value_scope<'tree>(node: Node<'tree>, source: &str) -> Node<'tree> {
1760 let nearest = python_alias_scope(node);
1761 let name = python_static_path(node, source)
1762 .and_then(|(root, _)| (root.kind() == "identifier").then_some(node_text(root, source)))
1763 .unwrap_or_default();
1764 match python_name_resolution_at(name, node, source) {
1765 PythonLexicalNameResolution::Local | PythonLexicalNameResolution::Nonlocal => nearest,
1766 PythonLexicalNameResolution::Global => {
1767 let mut scope = nearest;
1768 while scope.kind() != "module" {
1769 scope = python_alias_scope(scope.parent().expect("lexical scope has a parent"));
1770 }
1771 scope
1772 }
1773 PythonLexicalNameResolution::Unbound if nearest.kind() == "class_definition" => nearest,
1774 PythonLexicalNameResolution::Unbound => {
1775 let mut scope = nearest;
1776 while scope.kind() != "module" {
1777 scope = python_alias_scope(scope.parent().expect("lexical scope has a parent"));
1778 }
1779 scope
1780 }
1781 }
1782}
1783
1784fn python_sys_modules_stored_value<'tree>(
1785 value: Node<'tree>,
1786 context: &PythonSysModulesProofContext<'_, 'tree>,
1787 budget: &mut PythonRuntimeBudget,
1788 seen: &mut HashSet<(usize, usize, String)>,
1789) -> PythonSysModulesStoredValue {
1790 let source = context.source;
1791 let module = context.module;
1792 let bindings = context.bindings;
1793 let facts = context.facts;
1794 if !budget.visit() {
1795 return PythonSysModulesStoredValue::Unknown;
1796 }
1797 if python_sys_modules_table(value, source).is_some() {
1798 return PythonSysModulesStoredValue::Unknown;
1799 }
1800 if let Some(table) = python_sys_modules_read_table(value, source) {
1801 if python_imported_path(table, source, bindings)
1802 .is_none_or(|path| path != ["sys", "modules"])
1803 {
1804 return PythonSysModulesStoredValue::Unknown;
1805 }
1806 return match python_sys_modules_read_identity(
1807 if value.kind() == "subscript" {
1808 value.child_by_field_name("subscript")
1809 } else {
1810 python_sys_modules_arguments(value).first().copied()
1811 },
1812 value,
1813 context,
1814 budget,
1815 ) {
1816 PythonSysModulesLookup::Target(_) => PythonSysModulesStoredValue::Target,
1817 PythonSysModulesLookup::Unrelated => PythonSysModulesStoredValue::Unrelated,
1818 PythonSysModulesLookup::Escape(_) => PythonSysModulesStoredValue::Unknown,
1819 };
1820 }
1821 if matches!(
1822 value.kind(),
1823 "string"
1824 | "concatenated_string"
1825 | "integer"
1826 | "float"
1827 | "true"
1828 | "false"
1829 | "none"
1830 | "list"
1831 | "tuple"
1832 | "set"
1833 | "dictionary"
1834 | "list_comprehension"
1835 | "set_comprehension"
1836 | "dictionary_comprehension"
1837 | "generator_expression"
1838 ) {
1839 return PythonSysModulesStoredValue::Unrelated;
1840 }
1841 if let Some(path) = python_imported_path(value, source, bindings) {
1842 let root_name = python_static_path(value, source)
1843 .map(|(root, _)| node_text(root, source))
1844 .unwrap_or_default();
1845 let scope = python_sys_modules_value_scope(value, source);
1846 if facts.binding_counts.contains_key(&(
1847 scope.start_byte(),
1848 scope.end_byte(),
1849 root_name.to_owned(),
1850 )) {
1851 return PythonSysModulesStoredValue::Unknown;
1852 }
1853 return if path.join(".") == module {
1854 PythonSysModulesStoredValue::Target
1855 } else {
1856 PythonSysModulesStoredValue::Unrelated
1857 };
1858 }
1859 if value.kind() == "call" {
1860 let Some(function) = value.child_by_field_name("function") else {
1861 return PythonSysModulesStoredValue::Unknown;
1862 };
1863 if python_imported_path(function, source, bindings)
1864 .is_some_and(|path| path == ["types", "ModuleType"])
1865 {
1866 return PythonSysModulesStoredValue::Unrelated;
1867 }
1868 if let Some((root, path)) = python_static_path(function, source)
1869 && path.len() == 1
1870 && python_class_alias_visible_at(node_text(root, source), root, source, &facts.classes)
1871 {
1872 return PythonSysModulesStoredValue::Unrelated;
1873 }
1874 return PythonSysModulesStoredValue::Unknown;
1875 }
1876 let Some((root, path)) = python_static_path(value, source) else {
1877 return PythonSysModulesStoredValue::Unknown;
1878 };
1879 if path.len() == 1 && facts.aliases.contains(node_text(root, source)) {
1880 return PythonSysModulesStoredValue::Target;
1881 }
1882 if path.len() == 1
1883 && python_class_alias_visible_at(node_text(root, source), root, source, &facts.classes)
1884 {
1885 return PythonSysModulesStoredValue::Unrelated;
1886 }
1887 let scope = python_sys_modules_value_scope(value, source);
1888 if scope.kind() == "function_definition"
1889 && scope
1890 .child_by_field_name("parameters")
1891 .is_some_and(|parameters| {
1892 let mut cursor = parameters.walk();
1893 parameters
1894 .named_children(&mut cursor)
1895 .filter_map(|parameter| python_parameter_name(parameter, source))
1896 .any(|parameter| parameter == node_text(root, source))
1897 })
1898 {
1899 return PythonSysModulesStoredValue::Unknown;
1900 }
1901 if !python_sys_modules_has_one_binding(context, scope, node_text(root, source)) {
1902 return PythonSysModulesStoredValue::Unknown;
1903 }
1904 let key = (scope.start_byte(), scope.end_byte(), path.join("."));
1905 if !seen.insert(key.clone()) {
1906 return PythonSysModulesStoredValue::Unknown;
1907 }
1908 let Some([assignment]) = facts.assignments.get(&key).map(Vec::as_slice) else {
1909 return PythonSysModulesStoredValue::Unknown;
1910 };
1911 python_sys_modules_stored_value(*assignment, context, budget, seen)
1912}
1913
1914fn python_sys_modules_update_values<'tree>(
1915 mapping: Node<'tree>,
1916 context: &PythonSysModulesProofContext<'_, 'tree>,
1917 budget: &mut PythonRuntimeBudget,
1918 seen: &mut HashSet<(usize, usize, String)>,
1919) -> PythonSysModulesStoredValue {
1920 let source = context.source;
1921 let facts = context.facts;
1922 if mapping.kind() == "dictionary" {
1923 let mut cursor = mapping.walk();
1924 for pair in mapping.named_children(&mut cursor) {
1925 let Some(value) = pair.child_by_field_name("value") else {
1926 return PythonSysModulesStoredValue::Unknown;
1927 };
1928 let identity = python_sys_modules_stored_value(value, context, budget, seen);
1929 if identity != PythonSysModulesStoredValue::Unrelated {
1930 return identity;
1931 }
1932 }
1933 return PythonSysModulesStoredValue::Unrelated;
1934 }
1935 let Some((root, path)) = python_static_path(mapping, source) else {
1936 return PythonSysModulesStoredValue::Unknown;
1937 };
1938 let scope = python_sys_modules_value_scope(mapping, source);
1939 let key = (scope.start_byte(), scope.end_byte(), path.join("."));
1940 if !seen.insert(key.clone()) {
1941 return PythonSysModulesStoredValue::Unknown;
1942 }
1943 if !python_sys_modules_has_one_binding(context, scope, node_text(root, source)) {
1944 return PythonSysModulesStoredValue::Unknown;
1945 }
1946 let Some([assigned_value]) = facts.assignments.get(&key).map(Vec::as_slice) else {
1947 return PythonSysModulesStoredValue::Unknown;
1948 };
1949 python_sys_modules_update_values(*assigned_value, context, budget, seen)
1950}
1951
1952fn python_sys_modules_aliases<'tree>(
1956 root: Node<'tree>,
1957 source: &str,
1958 source_module: &str,
1959 module: &str,
1960 bindings: &[crate::imports::PythonImportBinding],
1961 budget: &mut PythonRuntimeBudget,
1962) -> Option<PythonSysModulesFacts<'tree>> {
1963 let mut aliases_from = HashMap::<String, Vec<String>>::default();
1964 let mut assignments = HashMap::<(usize, usize, String), Vec<Node<'_>>>::default();
1965 let mut binding_counts = HashMap::<(usize, usize, String), usize>::default();
1966 let mut classes = Vec::new();
1967 let mut pending_table_aliases = Vec::<(Vec<String>, Option<Node<'_>>, Node<'_>)>::new();
1968 let mut stack = vec![root];
1969 while let Some(node) = stack.pop() {
1970 if !budget.visit() {
1971 return None;
1972 }
1973 if node.kind() == "class_definition"
1974 && let Some(name) = node.child_by_field_name("name")
1975 {
1976 let scope = python_alias_scope(node);
1977 classes.push(PythonExternalClassAlias {
1978 name: node_text(name, source).to_owned(),
1979 scope_start_byte: scope.start_byte(),
1980 scope_end_byte: scope.end_byte(),
1981 });
1982 }
1983 for binding in python_direct_scope_bindings_bounded(node, source, || budget.visit())? {
1984 if binding.kind != PythonDirectScopeBindingKind::Import {
1985 let scope = python_alias_scope(binding.declaration);
1986 let key = (
1987 scope.start_byte(),
1988 scope.end_byte(),
1989 node_text(binding.declaration, source).to_owned(),
1990 );
1991 *binding_counts.entry(key).or_default() += 1;
1992 }
1993 }
1994 if matches!(node.kind(), "function_definition" | "lambda")
1995 && let Some(parameters) = node.child_by_field_name("parameters")
1996 {
1997 let scope = node;
1998 let mut cursor = parameters.walk();
1999 for parameter in parameters.named_children(&mut cursor) {
2000 if let Some(name) = python_parameter_name(parameter, source) {
2001 *binding_counts
2002 .entry((scope.start_byte(), scope.end_byte(), name))
2003 .or_default() += 1;
2004 }
2005 }
2006 }
2007 if matches!(node.kind(), "assignment" | "annotated_assignment") {
2008 let (targets, value) = if node.kind() == "assignment" {
2009 crate::imports::python_assignment_chain(node)
2010 } else {
2011 (
2012 node.child_by_field_name("left").into_iter().collect(),
2013 node.child_by_field_name("right"),
2014 )
2015 };
2016 if let Some(value) = value {
2017 for target in targets {
2018 if let Some((_, path)) = python_static_path(target, source) {
2019 let scope = python_alias_scope(target);
2020 let key = (scope.start_byte(), scope.end_byte(), path.join("."));
2021 assignments.entry(key).or_default().push(value);
2022 }
2023 }
2024 }
2025 }
2026 if node.kind() == "assignment" {
2027 let (targets, value) = crate::imports::python_assignment_chain(node);
2028 if let Some(value) = value
2029 && !targets.is_empty()
2030 && targets.iter().all(|target| target.kind() == "identifier")
2031 {
2032 let target_names = targets
2033 .iter()
2034 .map(|target| node_text(*target, source).to_owned())
2035 .collect::<Vec<_>>();
2036 if python_sys_modules_read_table(value, source).is_some_and(|table| {
2037 python_imported_path(table, source, bindings)
2038 .is_some_and(|path| path == ["sys", "modules"])
2039 }) {
2040 pending_table_aliases.push((
2041 target_names,
2042 if value.kind() == "subscript" {
2043 value.child_by_field_name("subscript")
2044 } else {
2045 python_sys_modules_arguments(value).first().copied()
2046 },
2047 value,
2048 ));
2049 } else if value.kind() == "identifier" {
2050 aliases_from
2051 .entry(node_text(value, source).to_owned())
2052 .or_default()
2053 .extend(target_names);
2054 }
2055 }
2056 }
2057 let mut cursor = node.walk();
2058 stack.extend(node.named_children(&mut cursor));
2059 }
2060
2061 let mut facts = PythonSysModulesFacts {
2062 aliases: HashSet::default(),
2063 assignments,
2064 binding_counts,
2065 classes,
2066 };
2067 let mut target_aliases = Vec::new();
2068 {
2069 let proof = PythonSysModulesProofContext {
2070 source,
2071 source_module,
2072 module,
2073 bindings,
2074 facts: &facts,
2075 };
2076 for (targets, key, read) in pending_table_aliases {
2077 if matches!(
2078 python_sys_modules_read_identity(key, read, &proof, budget),
2079 PythonSysModulesLookup::Target(_)
2080 ) {
2081 target_aliases.extend(targets);
2082 }
2083 }
2084 }
2085 facts.aliases.extend(target_aliases);
2086
2087 let mut pending = facts.aliases.iter().cloned().collect::<Vec<_>>();
2088 while let Some(source_name) = pending.pop() {
2089 if let Some(targets) = aliases_from.get(&source_name) {
2090 for target in targets {
2091 if facts.aliases.insert(target.clone()) {
2092 pending.push(target.clone());
2093 }
2094 }
2095 }
2096 }
2097 Some(facts)
2098}
2099
2100fn python_sys_modules_read_use(
2101 read: Node<'_>,
2102 source: &str,
2103 member: &str,
2104 aliases: &HashSet<String>,
2105) -> Option<PythonModuleMemberMutation> {
2106 let mut expression = read;
2107 let mut direct_attribute = None;
2108 while let Some(parent) = expression.parent() {
2109 if parent.kind() != "attribute"
2110 || parent
2111 .child_by_field_name("object")
2112 .is_none_or(|object| object.id() != expression.id())
2113 {
2114 break;
2115 }
2116 if direct_attribute.is_none() {
2117 direct_attribute = Some(parent);
2118 }
2119 expression = parent;
2120 }
2121 if let Some(attribute) = direct_attribute {
2122 let name = attribute
2123 .child_by_field_name("attribute")
2124 .map(|name| node_text(name, source));
2125 if name == Some("__dict__") {
2126 return Some(PythonModuleMemberMutation::Escape(node_source_range(
2127 attribute,
2128 )));
2129 }
2130 if name == Some(member) && is_binding_target(attribute) {
2131 return Some(PythonModuleMemberMutation::Write(node_source_range(
2132 attribute,
2133 )));
2134 }
2135 return None;
2136 }
2137 if python_builtin_module_read(expression, source) {
2138 return None;
2139 }
2140 if read.parent().is_some_and(|parent| {
2141 parent.kind() == "assignment"
2142 && parent
2143 .child_by_field_name("right")
2144 .is_some_and(|right| right.id() == read.id())
2145 && crate::imports::python_assignment_chain(parent)
2146 .0
2147 .iter()
2148 .all(|target| {
2149 target.kind() == "identifier" && aliases.contains(node_text(*target, source))
2150 })
2151 }) || read.parent().is_some_and(|parent| {
2152 parent.kind() == "expression_statement"
2153 && parent
2154 .named_child(0)
2155 .is_some_and(|expression| expression.id() == read.id())
2156 }) {
2157 return None;
2158 }
2159 Some(PythonModuleMemberMutation::Escape(node_source_range(read)))
2160}
2161
2162fn python_builtin_module_read(expression: Node<'_>, source: &str) -> bool {
2163 let Some(arguments) = expression
2164 .parent()
2165 .filter(|node| node.kind() == "argument_list")
2166 else {
2167 return false;
2168 };
2169 let Some(call) = arguments.parent().filter(|node| node.kind() == "call") else {
2170 return false;
2171 };
2172 let Some(function) = call.child_by_field_name("function") else {
2173 return false;
2174 };
2175 if function.kind() != "identifier"
2176 || !matches!(node_text(function, source), "getattr" | "hasattr")
2177 || matches!(
2178 python_name_resolution_at(node_text(function, source), function, source),
2179 PythonLexicalNameResolution::Local | PythonLexicalNameResolution::Nonlocal
2180 )
2181 {
2182 return false;
2183 }
2184 let mut cursor = arguments.walk();
2185 arguments
2186 .named_children(&mut cursor)
2187 .find(|argument| argument.kind() != "keyword_argument")
2188 .is_some_and(|first| first.id() == expression.id())
2189}
2190
2191fn python_module_table_alias_binding_use(
2192 node: Node<'_>,
2193 source: &str,
2194 aliases: &HashSet<String>,
2195) -> bool {
2196 if node.kind() != "identifier" {
2197 return false;
2198 }
2199 let name = node_text(node, source);
2200 if !aliases.contains(name) {
2201 return false;
2202 }
2203 if is_binding_target(node) {
2204 return true;
2205 }
2206 node.parent().is_some_and(|assignment| {
2207 assignment.kind() == "assignment"
2208 && assignment
2209 .child_by_field_name("right")
2210 .is_some_and(|right| right.id() == node.id())
2211 && crate::imports::python_assignment_chain(assignment)
2212 .0
2213 .iter()
2214 .all(|target| {
2215 target.kind() == "identifier" && aliases.contains(node_text(*target, source))
2216 })
2217 })
2218}
2219
2220pub fn python_external_class_member_mutation(
2227 root: Node<'_>,
2228 source: &str,
2229 class_identities: &[&str],
2230 member: &str,
2231 max_nodes: usize,
2232) -> Option<PythonModuleMemberMutation> {
2233 let bindings = python_import_bindings_from_tree(root, source);
2234 let Some(aliases) =
2235 python_external_class_aliases(root, source, &bindings, class_identities, max_nodes)
2236 else {
2237 return Some(PythonModuleMemberMutation::Budget);
2238 };
2239 let mut budget = PythonRuntimeBudget {
2240 remaining: max_nodes,
2241 exhausted: false,
2242 visited_nodes: 0,
2243 };
2244 let mut stack = vec![root];
2245 while let Some(node) = stack.pop() {
2246 if !budget.visit() {
2247 return Some(PythonModuleMemberMutation::Budget);
2248 }
2249 if node.kind() == "attribute"
2250 && node
2251 .child_by_field_name("attribute")
2252 .is_some_and(|attribute| node_text(attribute, source) == member)
2253 && is_binding_target(node)
2254 && let Some(object) = node.child_by_field_name("object")
2255 && python_external_class_expression(
2256 object,
2257 source,
2258 &bindings,
2259 class_identities,
2260 &aliases,
2261 )
2262 {
2263 return Some(PythonModuleMemberMutation::Write(node_source_range(node)));
2264 }
2265 if node.kind() == "call"
2266 && python_setattr_may_replace_class_member(
2267 node,
2268 source,
2269 &bindings,
2270 class_identities,
2271 member,
2272 &aliases,
2273 )
2274 {
2275 return Some(PythonModuleMemberMutation::Write(node_source_range(node)));
2276 }
2277 if node.kind() == "call"
2278 && python_mock_patch_may_replace_class_member(
2279 node,
2280 source,
2281 &bindings,
2282 class_identities,
2283 member,
2284 &aliases,
2285 )
2286 {
2287 return Some(PythonModuleMemberMutation::Write(node_source_range(node)));
2288 }
2289 if node.kind() == "identifier"
2290 && python_class_alias_visible_at(node_text(node, source), node, source, &aliases)
2291 && python_class_alias_escapes_at(node, source, &aliases)
2292 {
2293 return Some(PythonModuleMemberMutation::Escape(node_source_range(node)));
2294 }
2295 let mut cursor = node.walk();
2296 stack.extend(node.named_children(&mut cursor));
2297 }
2298 None
2299}
2300
2301#[derive(Clone)]
2302struct PythonExternalClassAlias {
2303 name: String,
2304 scope_start_byte: usize,
2305 scope_end_byte: usize,
2306}
2307
2308fn python_external_class_aliases(
2309 root: Node<'_>,
2310 source: &str,
2311 bindings: &[crate::imports::PythonImportBinding],
2312 class_identities: &[&str],
2313 max_nodes: usize,
2314) -> Option<Vec<PythonExternalClassAlias>> {
2315 let mut budget = PythonRuntimeBudget {
2316 remaining: max_nodes,
2317 exhausted: false,
2318 visited_nodes: 0,
2319 };
2320 let mut assignments = Vec::new();
2321 let mut stack = vec![root];
2322 while let Some(node) = stack.pop() {
2323 if !budget.visit() {
2324 return None;
2325 }
2326 if matches!(node.kind(), "assignment" | "annotated_assignment")
2327 && let Some(target) = node.child_by_field_name("left")
2328 && target.kind() == "identifier"
2329 && let Some(value) = node.child_by_field_name("right")
2330 {
2331 let scope = python_alias_scope(node);
2332 assignments.push((
2333 node_text(target, source).to_owned(),
2334 value,
2335 scope.start_byte(),
2336 scope.end_byte(),
2337 ));
2338 }
2339 let mut cursor = node.walk();
2340 stack.extend(node.named_children(&mut cursor));
2341 }
2342
2343 let mut aliases = Vec::<PythonExternalClassAlias>::new();
2344 loop {
2345 let before = aliases.len();
2346 for (name, value, scope_start_byte, scope_end_byte) in &assignments {
2347 if python_external_class_expression(
2348 *value,
2349 source,
2350 bindings,
2351 class_identities,
2352 &aliases,
2353 ) && !aliases.iter().any(|alias| {
2354 alias.name == *name
2355 && alias.scope_start_byte == *scope_start_byte
2356 && alias.scope_end_byte == *scope_end_byte
2357 }) {
2358 aliases.push(PythonExternalClassAlias {
2359 name: name.clone(),
2360 scope_start_byte: *scope_start_byte,
2361 scope_end_byte: *scope_end_byte,
2362 });
2363 }
2364 }
2365 if aliases.len() == before {
2366 return Some(aliases);
2367 }
2368 }
2369}
2370
2371fn python_alias_scope(mut node: Node<'_>) -> Node<'_> {
2372 while let Some(parent) = node.parent() {
2373 if matches!(
2374 parent.kind(),
2375 "module" | "class_definition" | "function_definition" | "lambda"
2376 ) {
2377 return parent;
2378 }
2379 node = parent;
2380 }
2381 node
2382}
2383
2384fn python_external_class_expression(
2385 expression: Node<'_>,
2386 source: &str,
2387 bindings: &[crate::imports::PythonImportBinding],
2388 class_identities: &[&str],
2389 aliases: &[PythonExternalClassAlias],
2390) -> bool {
2391 if python_imported_class_expression(expression, source, bindings, class_identities) {
2392 return true;
2393 }
2394 let Some((root, path)) = python_static_path(expression, source) else {
2395 return false;
2396 };
2397 path.len() == 1 && python_class_alias_visible_at(node_text(root, source), root, source, aliases)
2398}
2399
2400fn python_class_alias_visible_at(
2401 name: &str,
2402 reference: Node<'_>,
2403 source: &str,
2404 aliases: &[PythonExternalClassAlias],
2405) -> bool {
2406 if !aliases.iter().any(|alias| alias.name == name) {
2407 return false;
2408 }
2409 let mut current = reference;
2410 while let Some(scope) = nearest_scope(current) {
2411 if aliases.iter().any(|alias| {
2412 alias.name == name
2413 && alias.scope_start_byte == scope.start_byte()
2414 && alias.scope_end_byte == scope.end_byte()
2415 }) {
2416 return true;
2417 }
2418 match scope.kind() {
2419 "function_definition" | "lambda" => {
2420 let parameters = scope
2421 .child_by_field_name("parameters")
2422 .into_iter()
2423 .flat_map(|parameters| {
2424 let mut cursor = parameters.walk();
2425 parameters
2426 .named_children(&mut cursor)
2427 .filter_map(|parameter| python_parameter_name(parameter, source))
2428 .collect::<Vec<_>>()
2429 });
2430 let Some(inventory) =
2431 PythonLexicalScopeInventory::collect_bounded(scope, source, parameters, || {
2432 true
2433 })
2434 else {
2435 return true;
2436 };
2437 match inventory.name_resolution_at(name, reference) {
2438 PythonLexicalNameResolution::Local => return false,
2439 PythonLexicalNameResolution::Global => break,
2440 PythonLexicalNameResolution::Nonlocal
2441 | PythonLexicalNameResolution::Unbound => {}
2442 }
2443 }
2444 "module" => return false,
2445 "class_definition" => {}
2448 _ => unreachable!("nearest_scope returns only Python lexical scopes"),
2449 }
2450 current = scope;
2451 }
2452 aliases.iter().any(|alias| {
2453 alias.name == name
2454 && alias.scope_start_byte == 0
2455 && reference.start_byte() < alias.scope_end_byte
2456 })
2457}
2458
2459fn nearest_scope(mut node: Node<'_>) -> Option<Node<'_>> {
2460 while let Some(parent) = node.parent() {
2461 if matches!(
2462 parent.kind(),
2463 "module" | "class_definition" | "function_definition" | "lambda"
2464 ) {
2465 return Some(parent);
2466 }
2467 node = parent;
2468 }
2469 None
2470}
2471
2472fn python_class_alias_escapes_at(
2473 node: Node<'_>,
2474 source: &str,
2475 aliases: &[PythonExternalClassAlias],
2476) -> bool {
2477 let mut current = node;
2478 while let Some(parent) = current.parent() {
2479 if matches!(
2480 parent.kind(),
2481 "function_definition" | "lambda" | "class_definition" | "module"
2482 ) {
2483 return false;
2484 }
2485 if parent.kind() == "argument_list"
2486 && parent.parent().is_some_and(|call| call.kind() == "call")
2487 {
2488 return true;
2489 }
2490 if matches!(parent.kind(), "return_statement" | "yield") {
2491 return true;
2492 }
2493 if matches!(parent.kind(), "list" | "tuple" | "set" | "dictionary") {
2494 return true;
2495 }
2496 if matches!(parent.kind(), "assignment" | "annotated_assignment")
2497 && parent.child_by_field_name("right").is_some_and(|right| {
2498 right.start_byte() <= node.start_byte() && node.end_byte() <= right.end_byte()
2499 })
2500 {
2501 let Some(target) = parent.child_by_field_name("left") else {
2502 return true;
2503 };
2504 if target.kind() == "identifier"
2505 && python_class_alias_visible_at(node_text(target, source), target, source, aliases)
2506 {
2507 return false;
2508 }
2509 return true;
2510 }
2511 current = parent;
2512 }
2513 false
2514}
2515
2516fn python_setattr_may_replace_class_member(
2517 call: Node<'_>,
2518 source: &str,
2519 bindings: &[crate::imports::PythonImportBinding],
2520 class_identities: &[&str],
2521 member: &str,
2522 aliases: &[PythonExternalClassAlias],
2523) -> bool {
2524 let Some(function) = call.child_by_field_name("function") else {
2525 return false;
2526 };
2527 let imported_setattr = python_imported_path(function, source, bindings)
2528 .is_some_and(|path| path == ["builtins", "setattr"]);
2529 let builtin_setattr = python_static_path(function, source).is_some_and(|(root, path)| {
2530 path == ["setattr"]
2531 && !matches!(
2532 python_name_resolution_at(node_text(root, source), root, source),
2533 PythonLexicalNameResolution::Local | PythonLexicalNameResolution::Nonlocal
2534 )
2535 });
2536 if !imported_setattr && !builtin_setattr {
2537 return false;
2538 }
2539 let Some(arguments) = call.child_by_field_name("arguments") else {
2540 return false;
2541 };
2542 let mut cursor = arguments.walk();
2543 let positional = arguments
2544 .named_children(&mut cursor)
2545 .filter(|argument| argument.kind() != "keyword_argument")
2546 .collect::<Vec<_>>();
2547 let [object, name, ..] = positional.as_slice() else {
2548 return false;
2549 };
2550 if !python_external_class_expression(*object, source, bindings, class_identities, aliases) {
2551 return false;
2552 }
2553 python_plain_string_literal(*name, source).is_none_or(|name| name == member)
2554}
2555
2556fn python_mock_patch_may_replace_class_member(
2557 call: Node<'_>,
2558 source: &str,
2559 bindings: &[crate::imports::PythonImportBinding],
2560 class_identities: &[&str],
2561 member: &str,
2562 aliases: &[PythonExternalClassAlias],
2563) -> bool {
2564 let Some(function) = call.child_by_field_name("function") else {
2565 return false;
2566 };
2567 let Some(function_path) = python_imported_path(function, source, bindings) else {
2568 return false;
2569 };
2570 let patch_object = function_path == ["unittest", "mock", "patch", "object"];
2571 let patch_class_path = function_path == ["unittest", "mock", "patch"];
2572 if !patch_object && !patch_class_path {
2573 return false;
2574 }
2575 let Some(arguments) = call.child_by_field_name("arguments") else {
2576 return false;
2577 };
2578 let mut cursor = arguments.walk();
2579 let positional = arguments
2580 .named_children(&mut cursor)
2581 .filter(|argument| argument.kind() != "keyword_argument")
2582 .collect::<Vec<_>>();
2583 let Some(target) = positional.first().copied() else {
2584 return false;
2585 };
2586 if patch_object {
2587 let Some(name) = positional.get(1).copied() else {
2588 return false;
2589 };
2590 return python_external_class_expression(
2591 target,
2592 source,
2593 bindings,
2594 class_identities,
2595 aliases,
2596 ) && python_plain_string_literal(name, source).is_none_or(|name| name == member);
2597 }
2598 let Some(target_path) = python_plain_string_literal(target, source) else {
2599 return false;
2600 };
2601 class_identities
2602 .iter()
2603 .any(|class| target_path == *class || target_path == format!("{class}.{member}"))
2604}
2605
2606fn python_imported_path(
2607 expression: Node<'_>,
2608 source: &str,
2609 bindings: &[crate::imports::PythonImportBinding],
2610) -> Option<Vec<String>> {
2611 let (root, expression_path) = python_static_path(expression, source)?;
2612 let root_name = node_text(root, source);
2613 if matches!(
2614 python_name_resolution_at(root_name, root, source),
2615 PythonLexicalNameResolution::Local | PythonLexicalNameResolution::Nonlocal
2616 ) {
2617 return None;
2618 }
2619 let latest_import = bindings
2620 .iter()
2621 .filter(|binding| {
2622 binding.local_name == root_name
2623 && binding.start_byte <= root.start_byte()
2624 && binding.scope_start_byte <= root.start_byte()
2625 && root.end_byte() <= binding.scope_end_byte
2626 })
2627 .map(|binding| binding.start_byte)
2628 .max()?;
2629 let binding = bindings.iter().find(|binding| {
2630 binding.local_name == root_name
2631 && binding.start_byte == latest_import
2632 && binding.scope_start_byte <= root.start_byte()
2633 && root.end_byte() <= binding.scope_end_byte
2634 })?;
2635 let qualified_prefix = if binding.consumed_attributes > 0 {
2636 binding.qualified_name.split('.').next()?
2637 } else {
2638 binding.qualified_name.as_str()
2639 };
2640 let mut qualified_path = qualified_prefix
2641 .split('.')
2642 .map(str::to_owned)
2643 .collect::<Vec<_>>();
2644 qualified_path.extend(expression_path.into_iter().skip(1));
2645 Some(qualified_path)
2646}
2647
2648fn python_imported_class_expression(
2649 expression: Node<'_>,
2650 source: &str,
2651 bindings: &[crate::imports::PythonImportBinding],
2652 class_identities: &[&str],
2653) -> bool {
2654 let Some(path) = python_imported_path(expression, source, bindings) else {
2655 return false;
2656 };
2657 class_identities.iter().any(|identity| {
2658 identity
2659 .split('.')
2660 .map(str::to_owned)
2661 .eq(path.iter().cloned())
2662 })
2663}
2664
2665fn python_static_path<'tree>(
2666 node: Node<'tree>,
2667 source: &str,
2668) -> Option<(Node<'tree>, Vec<String>)> {
2669 let mut current = node;
2670 let mut attributes = Vec::new();
2671 loop {
2672 if current.kind() == "identifier" {
2673 attributes.reverse();
2674 let mut path = Vec::with_capacity(attributes.len() + 1);
2675 path.push(node_text(current, source).to_owned());
2676 path.extend(attributes);
2677 return Some((current, path));
2678 }
2679 if current.kind() != "attribute" {
2680 return None;
2681 }
2682 let attribute = current.child_by_field_name("attribute")?;
2683 attributes.push(node_text(attribute, source).to_owned());
2684 current = current.child_by_field_name("object")?;
2685 }
2686}
2687
2688fn module_binding_use(
2690 node: Node<'_>,
2691 source: &str,
2692 bindings: &[crate::imports::PythonImportBinding],
2693 table_aliases: &HashSet<String>,
2694 module: &str,
2695 member: &str,
2696) -> Option<PythonModuleMemberMutation> {
2697 let Some(parent) = node.parent() else {
2698 return Some(PythonModuleMemberMutation::Escape(node_source_range(node)));
2699 };
2700 if matches!(
2701 parent.kind(),
2702 "import_statement" | "import_from_statement" | "aliased_import" | "dotted_name"
2703 ) {
2704 return None;
2705 }
2706 let mut expression = node;
2707 while let Some(parent) = expression.parent() {
2708 if parent.kind() != "attribute"
2709 || parent
2710 .child_by_field_name("object")
2711 .is_none_or(|object| object.id() != expression.id())
2712 {
2713 break;
2714 }
2715 expression = parent;
2716 }
2717 if python_builtin_module_read(expression, source) {
2718 return None;
2719 }
2720 let imported_path = python_imported_path(expression, source, bindings);
2721 let table_alias_path = python_static_path(expression, source).and_then(|(root, suffix)| {
2722 let root_name = node_text(root, source);
2723 table_aliases.contains(root_name).then(|| {
2724 let mut resolved = vec![module.to_owned()];
2725 resolved.extend(suffix.into_iter().skip(1));
2726 resolved
2727 })
2728 });
2729 let Some(path) = imported_path.or(table_alias_path) else {
2730 return Some(PythonModuleMemberMutation::Escape(node_source_range(node)));
2731 };
2732 let path = path.join(".");
2733 let target = format!("{module}.{member}");
2734 if path == target {
2735 return is_binding_target(expression)
2736 .then(|| PythonModuleMemberMutation::Write(node_source_range(expression)));
2737 }
2738 if path == format!("{module}.__dict__") || path == module {
2739 return Some(PythonModuleMemberMutation::Escape(node_source_range(
2740 expression,
2741 )));
2742 }
2743 if path.starts_with(&format!("{module}.")) {
2744 None
2745 } else {
2746 Some(PythonModuleMemberMutation::Escape(node_source_range(node)))
2747 }
2748}
2749
2750#[derive(Clone, Copy, Debug, Eq, PartialEq)]
2752enum PythonTestHelperCall<'tree> {
2753 MockPatch,
2755 MockPatchDict,
2757 MonkeyPatchSetItem(Node<'tree>),
2760 MonkeyPatchDelItem(Node<'tree>),
2762}
2763
2764fn python_test_helper_call<'tree>(
2765 call: Node<'tree>,
2766 source: &str,
2767 bindings: &[crate::imports::PythonImportBinding],
2768) -> Option<PythonTestHelperCall<'tree>> {
2769 let function = call.child_by_field_name("function")?;
2770 if let Some(path) = python_imported_path(function, source, bindings)
2771 && let Some(patch) = path.windows(2).rposition(|pair| pair == ["mock", "patch"])
2772 {
2773 return match &path[patch + 2..] {
2774 [] => Some(PythonTestHelperCall::MockPatch),
2775 [method] if method == "dict" => Some(PythonTestHelperCall::MockPatchDict),
2776 _ => None,
2777 };
2778 }
2779 if function.kind() != "attribute" {
2780 return None;
2781 }
2782 let receiver = function.child_by_field_name("object")?;
2783 match node_text(function.child_by_field_name("attribute")?, source) {
2784 "setitem" => Some(PythonTestHelperCall::MonkeyPatchSetItem(receiver)),
2785 "delitem" => Some(PythonTestHelperCall::MonkeyPatchDelItem(receiver)),
2786 _ => None,
2787 }
2788}
2789
2790fn python_mock_patch_may_replace_module_member(
2791 call: Node<'_>,
2792 source: &str,
2793 bindings: &[crate::imports::PythonImportBinding],
2794 module: &str,
2795 member: &str,
2796) -> bool {
2797 if python_test_helper_call(call, source, bindings) != Some(PythonTestHelperCall::MockPatch) {
2798 return false;
2799 }
2800 let Some(arguments) = call.child_by_field_name("arguments") else {
2801 return false;
2802 };
2803 let mut cursor = arguments.walk();
2804 let Some(target) = arguments.named_children(&mut cursor).next() else {
2805 return false;
2806 };
2807 let expected = format!("{module}.{member}");
2808 python_plain_string_literal(target, source) == Some(expected.as_str())
2809}
2810
2811#[cfg(test)]
2812mod tests {
2813 use super::*;
2814
2815 fn facts(source: &str) -> PythonRuntimeReadFacts {
2816 let mut parser = tree_sitter::Parser::new();
2817 parser
2818 .set_language(&tree_sitter_python::LANGUAGE.into())
2819 .expect("python grammar loads");
2820 let tree = parser.parse(source, None).expect("fixture parses");
2821 extract_python_runtime_reads(tree.root_node(), source, 64)
2822 }
2823
2824 #[test]
2825 #[brokk_bifrost_test_macros::hardwon(
2826 "6b123df7cc68: quicksilver round 6 on brokk-bifrost-python: module alias exhaustion certified mutation-free"
2827 )]
2828 fn qs6_module_mutation_budget_is_not_absence() {
2829 let source = "import os\nos.system = fake\n";
2830 let tree = crate::declarations::parse_python_tree(source).unwrap();
2831 for budget in [0, 1, 2] {
2832 assert_eq!(
2833 python_module_member_mutation(
2834 tree.root_node(),
2835 source,
2836 "test",
2837 "os",
2838 "system",
2839 budget
2840 ),
2841 Some(PythonModuleMemberMutation::Budget),
2842 "budget={budget}"
2843 );
2844 }
2845 assert!(
2846 python_module_member_mutation(tree.root_node(), source, "test", "os", "system", 4096)
2847 .is_some()
2848 );
2849 }
2850
2851 #[test]
2854 #[brokk_bifrost_test_macros::hardwon(
2855 "6b123df7cc68: quicksilver round 6 on brokk-bifrost-python: runtime indices lost radix and separator literals"
2856 )]
2857 fn qs6_runtime_key_shapes() {
2858 use PythonRuntimeAccessKey::{Dynamic, Index, Property, Unsupported};
2859 for (key, expected) in [
2860 ("0", Index(0)),
2861 ("0x1", Index(1)),
2862 ("0b1", Index(1)),
2863 ("0o7", Index(7)),
2864 ("1_0", Index(10)),
2865 ("340282366920938463463374607431768211455", Index(u128::MAX)),
2866 ("340282366920938463463374607431768211456", Unsupported),
2867 ("1:3", Unsupported),
2868 ("1, 2", Unsupported),
2869 ("1,", Unsupported),
2870 ("(1, 2)", Unsupported),
2871 ("name", Dynamic),
2872 ("\"HOME\"", Property("HOME".into())),
2873 ] {
2874 let source = format!("import sys\nvalue = sys.argv[{key}]\n");
2875 let facts = facts(&source);
2876 assert!(facts.complete, "{facts:?}");
2877 assert_eq!(facts.reads.len(), 1, "{facts:?}");
2878 assert_eq!(facts.reads[0].access, expected, "{key}");
2879 }
2880 }
2881
2882 #[test]
2883 fn qs6_runtime_extraction_bounds() {
2884 let source = "pass\n".repeat(100);
2885 let tree = crate::declarations::parse_python_tree(&source).unwrap();
2886 let empty = extract_python_runtime_reads(tree.root_node(), &source, 1);
2887 assert!(empty.complete, "{empty:?}");
2888 assert!(empty.visited_nodes > 64, "{empty:?}");
2889 let source = "import sys\nx = sys.argv[1]\ny = sys.argv[2]\n";
2890 let tree = crate::declarations::parse_python_tree(source).unwrap();
2891 for cap in [0, 1, 2, 3] {
2892 let facts = extract_python_runtime_reads(tree.root_node(), source, cap);
2893 assert_eq!(facts.reads.len(), cap.min(2), "{facts:?}");
2894 assert_eq!(facts.complete, cap > 2, "{facts:?}");
2895 }
2896 }
2897
2898 #[test]
2900 fn imported_module_environment_subscript_is_a_pristine_keyed_read() {
2901 let source = "import os\n\n\ndef run():\n command = os.environ[\"DFB\"]\n os.system(command)\n";
2902 let facts = facts(source);
2903 assert!(facts.complete, "{facts:#?}");
2904 let [read] = facts.reads.as_slice() else {
2905 panic!("exactly one keyed read: {facts:#?}");
2906 };
2907 assert_eq!(read.root_name, "os");
2908 assert_eq!(read.container, "environ");
2909 assert_eq!(
2910 read.access,
2911 PythonRuntimeAccessKey::Property("DFB".to_owned())
2912 );
2913 assert_eq!(
2914 read.lexical_resolution,
2915 PythonRuntimeRootResolution::ImportedModule
2916 );
2917 assert_eq!(read.mutation, PythonRuntimeMutationEvidence::NoKnownWrite);
2918 assert_eq!(
2919 read.accessor,
2920 PythonRuntimeAccessorCoverage::NoKnownAccessorEffects,
2921 "the only call runs after the read"
2922 );
2923 assert!(facts.writes.is_empty(), "{facts:#?}");
2924 assert_eq!(
2927 &source[read.candidate_anchor.start_byte..read.candidate_anchor.end_byte],
2928 "os.environ"
2929 );
2930 assert_eq!(
2931 &source[read.range.start_byte..read.range.end_byte],
2932 "os.environ[\"DFB\"]"
2933 );
2934 }
2935
2936 #[test]
2937 fn imported_module_argv_index_is_a_keyed_read() {
2938 let facts = facts("import sys\n\n\ndef run():\n return sys.argv[1]\n");
2939 let [read] = facts.reads.as_slice() else {
2940 panic!("exactly one keyed read: {facts:#?}");
2941 };
2942 assert_eq!(read.root_name, "sys");
2943 assert_eq!(read.container, "argv");
2944 assert_eq!(read.access, PythonRuntimeAccessKey::Index(1));
2945 }
2946
2947 #[test]
2950 fn a_shadowed_module_name_is_not_a_read() {
2951 let facts = facts("import os\n\n\ndef run(os):\n return os.environ[\"DFB\"]\n");
2952 assert!(facts.reads.is_empty(), "{facts:#?}");
2953 }
2954
2955 #[test]
2956 fn a_local_container_name_is_not_a_read() {
2957 let facts =
2958 facts("def run():\n environ = {\"DFB\": \"x\"}\n return environ[\"DFB\"]\n");
2959 assert!(facts.reads.is_empty(), "{facts:#?}");
2960 assert!(facts.writes.is_empty(), "{facts:#?}");
2961 }
2962
2963 #[test]
2964 fn an_unrelated_module_spelled_os_is_not_a_read() {
2965 let facts = facts("from mypkg import os\n\n\ndef run():\n return os.environ[\"DFB\"]\n");
2966 assert!(facts.reads.is_empty(), "{facts:#?}");
2967 }
2968
2969 #[test]
2970 fn a_dynamic_key_stays_a_read_with_no_static_key() {
2971 let facts = facts("import os\n\n\ndef run(name):\n return os.environ[name]\n");
2972 let [read] = facts.reads.as_slice() else {
2973 panic!("one read: {facts:#?}");
2974 };
2975 assert_eq!(read.access, PythonRuntimeAccessKey::Dynamic);
2976 }
2977
2978 #[test]
2979 fn a_container_write_refutes_every_read_of_that_container() {
2980 let facts = facts(
2981 "import os\n\n\ndef run():\n os.environ[\"A\"] = \"1\"\n return os.environ[\"B\"]\n",
2982 );
2983 let [read] = facts.reads.as_slice() else {
2984 panic!("one read: {facts:#?}");
2985 };
2986 assert_eq!(read.mutation, PythonRuntimeMutationEvidence::KnownWrite);
2987 let [write] = facts.writes.as_slice() else {
2988 panic!("one write: {facts:#?}");
2989 };
2990 assert_eq!(write.target, PythonRuntimeWriteTarget::Key);
2991 assert_eq!(write.container.as_deref(), Some("environ"));
2992 }
2993
2994 #[test]
2995 fn a_container_method_receiver_refutes_the_pristine_claim() {
2996 let facts = facts(
2997 "import os\n\n\ndef run():\n os.environ.update({})\n return os.environ[\"B\"]\n",
2998 );
2999 let [read] = facts.reads.as_slice() else {
3000 panic!("one read: {facts:#?}");
3001 };
3002 assert_eq!(
3003 read.mutation,
3004 PythonRuntimeMutationEvidence::KnownWriteAndUnknownEffects
3005 );
3006 }
3007
3008 #[test]
3009 fn a_module_object_escape_refutes_every_container() {
3010 let facts =
3011 facts("import os\n\n\ndef run(sink):\n sink(os)\n return os.environ[\"B\"]\n");
3012 let [read] = facts.reads.as_slice() else {
3013 panic!("one read: {facts:#?}");
3014 };
3015 assert_eq!(
3016 read.mutation,
3017 PythonRuntimeMutationEvidence::KnownWriteAndUnknownEffects
3018 );
3019 assert!(
3020 facts
3021 .writes
3022 .iter()
3023 .any(|write| write.target == PythonRuntimeWriteTarget::ModuleEscape),
3024 "{facts:#?}"
3025 );
3026 }
3027
3028 #[test]
3029 fn a_call_before_the_read_leaves_accessor_coverage_open() {
3030 let facts = facts(
3031 "import os\n\n\ndef run(prepare):\n prepare()\n return os.environ[\"B\"]\n",
3032 );
3033 let [read] = facts.reads.as_slice() else {
3034 panic!("one read: {facts:#?}");
3035 };
3036 assert_eq!(
3037 read.accessor,
3038 PythonRuntimeAccessorCoverage::UnknownAccessorEffects
3039 );
3040 }
3041
3042 #[test]
3045 fn a_call_in_another_function_does_not_poison_the_read() {
3046 let source = "import os\n\n\ndef other(prepare):\n prepare()\n\n\ndef run():\n return os.environ[\"B\"]\n";
3047 let facts = facts(source);
3048 let [read] = facts.reads.as_slice() else {
3049 panic!("one read: {facts:#?}");
3050 };
3051 assert_eq!(
3052 read.accessor,
3053 PythonRuntimeAccessorCoverage::NoKnownAccessorEffects
3054 );
3055 }
3056
3057 #[test]
3058 fn a_call_later_in_the_same_loop_can_precede_the_read() {
3059 let source = "import os\n\n\ndef run(items, prepare):\n for item in items:\n value = os.environ[\"B\"]\n prepare(value)\n";
3060 let facts = facts(source);
3061 let [read] = facts.reads.as_slice() else {
3062 panic!("one read: {facts:#?}");
3063 };
3064 assert_eq!(
3065 read.accessor,
3066 PythonRuntimeAccessorCoverage::UnknownAccessorEffects
3067 );
3068 }
3069
3070 #[test]
3071 fn an_aliased_import_keeps_the_absolute_module_identity() {
3072 let facts =
3073 facts("import os as operating\n\n\ndef run():\n return operating.environ[\"B\"]\n");
3074 let [read] = facts.reads.as_slice() else {
3075 panic!("one read: {facts:#?}");
3076 };
3077 assert_eq!(read.binding_name, "operating");
3078 assert_eq!(read.root_name, "os");
3079 }
3080
3081 #[test]
3084 fn a_submodule_import_agrees_with_its_package_binding() {
3085 let facts =
3086 facts("import os\nimport os.path\n\n\ndef run():\n return os.environ[\"B\"]\n");
3087 let [read] = facts.reads.as_slice() else {
3088 panic!("one read: {facts:#?}");
3089 };
3090 assert_eq!(read.root_name, "os");
3091 }
3092
3093 #[test]
3094 fn a_module_scope_rebinding_of_the_import_name_is_not_a_read() {
3095 let facts = facts("import os\n\nos = None\n\n\ndef run():\n return os.environ[\"B\"]\n");
3096 assert!(facts.reads.is_empty(), "{facts:#?}");
3097 }
3098}
3099
3100#[cfg(test)]
3101mod module_member_tests {
3102 use super::*;
3103
3104 fn mutation(source: &str, member: &str) -> Option<PythonModuleMemberMutation> {
3105 let mut parser = tree_sitter::Parser::new();
3106 parser
3107 .set_language(&tree_sitter_python::LANGUAGE.into())
3108 .expect("python grammar loads");
3109 let tree = parser.parse(source, None).expect("fixture parses");
3110 python_module_member_mutation(tree.root_node(), source, "test", "os", member, 4096)
3111 }
3112
3113 #[test]
3114 fn a_member_assignment_refuses_that_member() {
3115 let found = mutation(
3116 "import os\n\n\ndef patch(fake):\n os.system = fake\n",
3117 "system",
3118 );
3119 assert!(
3120 matches!(found, Some(PythonModuleMemberMutation::Write(_))),
3121 "{found:?}"
3122 );
3123 assert_eq!(
3125 mutation(
3126 "import os\n\n\ndef patch(fake):\n os.getcwd = fake\n",
3127 "system"
3128 ),
3129 None
3130 );
3131 }
3132
3133 #[test]
3134 fn a_module_object_escape_refuses_every_member() {
3135 for source in [
3136 "import os\n\n\ndef leak(sink):\n sink(os)\n",
3137 "import os\n\n\nalias = os\n",
3138 "import os\nimport functools\n\n\ndef patch():\n setattr(os, \"system\", None)\n",
3139 ] {
3140 let found = mutation(source, "system");
3141 assert!(
3142 matches!(found, Some(PythonModuleMemberMutation::Escape(_))),
3143 "{source}: {found:?}"
3144 );
3145 }
3146 }
3147
3148 #[test]
3149 fn a_matching_module_table_read_counts_its_attribute_write() {
3150 let found = mutation(
3151 "import sys\n\n\ndef patch(fake):\n sys.modules[\"os\"].system = fake\n",
3152 "system",
3153 );
3154 assert!(
3155 matches!(found, Some(PythonModuleMemberMutation::Write(_))),
3156 "{found:?}"
3157 );
3158 }
3159
3160 #[test]
3161 fn a_file_that_never_binds_the_module_proves_the_absence() {
3162 assert_eq!(mutation("def run():\n return 1\n", "system"), None);
3163 }
3164}
3165
3166#[cfg(test)]
3167mod shadowing_tests {
3168 use super::*;
3169
3170 fn facts(source: &str) -> PythonRuntimeReadFacts {
3171 let mut parser = tree_sitter::Parser::new();
3172 parser
3173 .set_language(&tree_sitter_python::LANGUAGE.into())
3174 .expect("python grammar loads");
3175 let tree = parser.parse(source, None).expect("fixture parses");
3176 extract_python_runtime_reads(tree.root_node(), source, 64)
3177 }
3178
3179 #[test]
3182 fn a_parameter_spelling_the_module_name_is_not_an_escape() {
3183 let source = "import os\n\n\ndef read_shadowed(os):\n return os.environ[\"A\"]\n\n\ndef read_real():\n return os.environ[\"B\"]\n";
3184 let facts = facts(source);
3185 assert!(facts.writes.is_empty(), "{facts:#?}");
3186 let [read] = facts.reads.as_slice() else {
3187 panic!("only the unshadowed read is a candidate: {facts:#?}");
3188 };
3189 assert_eq!(read.mutation, PythonRuntimeMutationEvidence::NoKnownWrite);
3190 assert_eq!(
3191 &source[read.key_range.start_byte..read.key_range.end_byte],
3192 "\"B\""
3193 );
3194 }
3195
3196 #[test]
3197 fn a_parameter_spelling_the_module_name_does_not_refuse_a_member() {
3198 let source =
3199 "import os\n\n\ndef helper(os):\n return os\n\n\ndef run(c):\n os.system(c)\n";
3200 let mut parser = tree_sitter::Parser::new();
3201 parser
3202 .set_language(&tree_sitter_python::LANGUAGE.into())
3203 .expect("python grammar loads");
3204 let tree = parser.parse(source, None).expect("fixture parses");
3205 assert_eq!(
3206 python_module_member_mutation(tree.root_node(), source, "test", "os", "system", 4096),
3207 None
3208 );
3209 }
3210}
3211
3212#[cfg(test)]
3213mod external_class_mutation_tests {
3214 use super::*;
3215 use brokk_bifrost_test_macros::hardwon;
3216
3217 fn mutation(source: &str) -> Option<PythonModuleMemberMutation> {
3218 let mut parser = tree_sitter::Parser::new();
3219 parser
3220 .set_language(&tree_sitter_python::LANGUAGE.into())
3221 .expect("python grammar loads");
3222 let tree = parser.parse(source, None).expect("fixture parses");
3223 python_external_class_member_mutation(
3224 tree.root_node(),
3225 source,
3226 &["paramiko.client.SSHClient"],
3227 "set_missing_host_key_policy",
3228 4096,
3229 )
3230 }
3231
3232 #[test]
3233 fn exact_imported_class_member_writes_are_found() {
3234 for source in [
3235 "from paramiko.client import SSHClient as Client\n\nClient.set_missing_host_key_policy = fake\n",
3236 "import paramiko.client as client\n\nclient.SSHClient.set_missing_host_key_policy = fake\n",
3237 "from paramiko.client import SSHClient as Client\n\nsetattr(Client, 'set_missing_host_key_policy', fake)\n",
3238 "from paramiko.client import SSHClient as Client\n\nsetattr(Client, name, fake)\n",
3239 ] {
3240 assert!(
3241 matches!(mutation(source), Some(PythonModuleMemberMutation::Write(_))),
3242 "{source}"
3243 );
3244 }
3245 }
3246
3247 #[test]
3248 #[hardwon("#4411: class aliases and mock patch APIs left external method dispatch exact")]
3249 fn class_aliases_and_mock_patch_apis_refute_method_closure() {
3250 for source in [
3251 "from paramiko.client import SSHClient as C\nAlias = C\nAlias.set_missing_host_key_policy = fake\n",
3252 "from paramiko.client import SSHClient as C\nAlias = C\nsetattr(Alias, 'set_missing_host_key_policy', fake)\n",
3253 "from unittest import mock\nfrom paramiko.client import SSHClient as C\nmock.patch.object(C, 'set_missing_host_key_policy', fake)\n",
3254 "from unittest.mock import patch\nfrom paramiko.client import SSHClient as C\npatch.object(C, 'set_missing_host_key_policy', fake)\n",
3255 "from unittest import mock\nmock.patch('paramiko.client.SSHClient.set_missing_host_key_policy', fake)\n",
3256 ] {
3257 assert!(
3258 matches!(mutation(source), Some(PythonModuleMemberMutation::Write(_))),
3259 "{source}"
3260 );
3261 }
3262 }
3263
3264 #[test]
3265 #[hardwon("#4411: an escaped class alias left external method dispatch exact")]
3266 fn an_escaped_class_alias_refutes_method_closure() {
3267 assert!(matches!(
3268 mutation("from paramiko.client import SSHClient as C\nAlias = C\nconsume(Alias)\n"),
3269 Some(PythonModuleMemberMutation::Escape(_))
3270 ));
3271 }
3272
3273 #[test]
3274 fn constructor_uses_and_same_named_workspace_classes_are_not_writes() {
3275 assert_eq!(
3276 mutation(
3277 "from paramiko.client import SSHClient as Client\n\n\ndef connect():\n client = Client()\n client.set_missing_host_key_policy(policy)\n"
3278 ),
3279 None
3280 );
3281 assert_eq!(
3282 mutation(
3283 "class SSHClient:\n pass\n\nSSHClient.set_missing_host_key_policy = fake\n"
3284 ),
3285 None
3286 );
3287 }
3288}
3289
3290#[cfg(test)]
3291mod module_member_mutation_tests {
3292 use super::*;
3293 use brokk_bifrost_test_macros::hardwon;
3294
3295 fn mutation(source: &str) -> Option<PythonModuleMemberMutation> {
3296 mutation_in(source, "pkg.M")
3297 }
3298
3299 fn mutation_in(source: &str, source_module: &str) -> Option<PythonModuleMemberMutation> {
3300 mutation_for_module(source, source_module, "pkg.M")
3301 }
3302
3303 fn mutation_for_module(
3304 source: &str,
3305 source_module: &str,
3306 module: &str,
3307 ) -> Option<PythonModuleMemberMutation> {
3308 let mut parser = tree_sitter::Parser::new();
3309 parser
3310 .set_language(&tree_sitter_python::LANGUAGE.into())
3311 .expect("python grammar loads");
3312 let tree = parser.parse(source, None).expect("fixture parses");
3313 python_module_member_mutation(
3314 tree.root_node(),
3315 source,
3316 source_module,
3317 module,
3318 "NAME",
3319 4096,
3320 )
3321 }
3322
3323 #[test]
3324 #[hardwon(
3325 "#4622: writes through imported module aliases and mock patch APIs must keep a global class unknown"
3326 )]
3327 fn imported_module_member_mutation_covers_dynamic_write_forms() {
3328 for source in [
3329 "import pkg.M as target\ntarget.NAME = fake\n",
3330 "import pkg.M as target\ndel target.NAME\n",
3331 "import pkg.M as target\nsetattr(target, 'NAME', fake)\n",
3332 "import pkg.M as target\nsetattr(target, name, fake)\n",
3333 "import pkg.M as target\ntarget.__dict__['NAME'] = fake\n",
3334 "from unittest import mock\nmock.patch('pkg.M.NAME', fake)\n",
3335 "from unittest.mock import patch\npatch('pkg.M.NAME', fake)\n",
3336 "import pkg.M as target\nfrom unittest import mock\nmock.patch.object(target, 'NAME', fake)\n",
3337 ] {
3338 assert!(mutation(source).is_some(), "{source}");
3339 }
3340 assert_eq!(
3341 None,
3342 mutation("from unittest import mock\nmock.patch('pkg.Other.NAME', fake)\n")
3343 );
3344 }
3345
3346 #[test]
3347 #[hardwon("#4622: unrelated sys.modules reads blocked exact module-global class proofs")]
3348 fn sys_modules_reads_are_scoped_by_literal_module_identity() {
3349 assert_eq!(
3350 None,
3351 mutation("import sys\nother = sys.modules.get('torch')\n")
3352 );
3353 assert_eq!(None, mutation("import sys\nother = sys.modules['torch']\n"));
3354 assert!(matches!(
3355 mutation("import sys\nmodule = sys.modules.get('pkg.M')\nmodule.NAME = fake\n"),
3356 Some(PythonModuleMemberMutation::Write(_))
3357 ));
3358 assert!(matches!(
3359 mutation("import sys\nmodule = sys.modules['pkg.M']\nmodule.NAME = fake\n"),
3360 Some(PythonModuleMemberMutation::Write(_))
3361 ));
3362 assert_eq!(
3363 None,
3364 mutation("import sys\nmodule = sys.modules['pkg.M']\nconsume(module.NAME)\n")
3365 );
3366 assert!(matches!(
3367 mutation("import sys\nmodule = sys.modules[__name__]\nmodule.NAME = fake\n"),
3368 Some(PythonModuleMemberMutation::Write(_))
3369 ));
3370 assert_eq!(
3371 None,
3372 mutation_in("import sys\nmodule = sys.modules[__name__]\n", "pkg.Other")
3373 );
3374 for source in [
3375 "import sys\nsys.modules['pkg.M'] = fake\n",
3376 "import sys\nsys.modules.setdefault('pkg.M', fake)\n",
3377 "import sys\nsys.modules.update({'pkg.M': fake})\n",
3378 "import sys\nconsume(sys.modules)\n",
3379 ] {
3380 assert!(
3381 matches!(
3382 mutation(source),
3383 Some(PythonModuleMemberMutation::Escape(_))
3384 ),
3385 "{source}"
3386 );
3387 }
3388 for source in [
3389 "import sys\nmodule = sys.modules[module_name]\n",
3390 "import sys\ndel sys.modules['torch']\n",
3391 "import sys\nsys.modules.pop('torch')\n",
3392 "import sys\nsys.modules.clear()\n",
3393 "class Shim: pass\nimport sys\nsys.modules['shim'] = Shim()\n",
3394 "class Shim: pass\nimport sys\nsys.modules.update({'shim': Shim()})\n",
3395 "import sys\nimport types\nsys.modules['shim'] = types.ModuleType('shim')\n",
3396 "import sys\nsys.modules['literal'] = None\n",
3397 "import pkg.Other as shim\nimport sys\nsys.modules['shim'] = shim\n",
3398 ] {
3399 assert_eq!(mutation(source), None, "{source}");
3400 }
3401 }
3402
3403 #[test]
3404 #[hardwon("#8107: monkeypatch.setitem(sys.modules, ...) escaped every module workspace-wide")]
3405 fn sys_modules_test_helper_edits_are_table_operations() {
3406 for source in [
3407 "import sys\ndef test(monkeypatch):\n monkeypatch.setitem(sys.modules, 'dotenv', None)\n",
3408 "import sys\ndef test(monkeypatch):\n monkeypatch.delitem(sys.modules, 'dotenv', raising=False)\n",
3409 "import sys\nimport pytest\ndef test(mp: pytest.MonkeyPatch):\n mp.setitem(sys.modules, 'dotenv', None)\n",
3410 "import sys\nimport pytest\ndef test():\n mp = pytest.MonkeyPatch()\n mp.setitem(sys.modules, 'dotenv', None)\n",
3411 "import sys\nfrom unittest import mock\n@mock.patch.dict(sys.modules, {'dotenv': None})\ndef test():\n pass\n",
3412 "import sys\nfrom unittest.mock import patch\ndef test():\n with patch.dict(sys.modules, values={'dotenv': None}, clear=True, extra=None):\n pass\n",
3413 ] {
3414 assert_eq!(mutation(source), None, "{source}");
3415 }
3416 for source in [
3417 "import sys\ndef test(monkeypatch, fake):\n monkeypatch.setitem(sys.modules, 'pkg.M', fake)\n",
3419 "import pkg.M as target\nimport sys\ndef test(monkeypatch):\n monkeypatch.setitem(sys.modules, 'shim', target)\n",
3420 "import pkg.M as target\nimport sys\nfrom unittest import mock\nmock.patch.dict(sys.modules, {'shim': target})\n",
3421 "import pkg.M as target\nimport sys\nfrom unittest import mock\nmock.patch.dict(sys.modules, values={'shim': target})\n",
3422 "import pkg.M as target\nimport sys\nfrom unittest import mock\nmock.patch.dict(sys.modules, shim=target)\n",
3423 "import sys\ndef test(mp):\n mp.setitem(sys.modules, 'dotenv', None)\n",
3425 "import sys\ndef test(monkeypatch):\n monkeypatch = other\n monkeypatch.setitem(sys.modules, 'dotenv', None)\n",
3426 "import sys\ndef test(monkeypatch):\n monkeypatch.setitem(registry, sys.modules, None)\n",
3428 "import sys\ndef test(monkeypatch):\n monkeypatch.setattr(sys.modules, 'dotenv', None)\n",
3429 ] {
3430 assert!(
3431 matches!(
3432 mutation(source),
3433 Some(PythonModuleMemberMutation::Escape(_))
3434 ),
3435 "{source}"
3436 );
3437 }
3438 }
3439
3440 #[test]
3441 #[hardwon("#4622: sys.modules values must be followed by module identity")]
3442 fn dynamic_module_table_reads_follow_only_writes_through_the_possible_alias() {
3443 assert!(matches!(
3444 mutation("import pkg.M as target\nimport sys\nsys.modules['shim'] = target\n"),
3445 Some(PythonModuleMemberMutation::Escape(_))
3446 ));
3447 assert_eq!(
3448 mutation("import sys\ntarget = sys.modules[module_name]\nconsume(target.NAME)\n"),
3449 None
3450 );
3451 assert_eq!(
3452 mutation(
3453 "import sys\ntarget = sys.modules[module_name]\nconsume(getattr(target, 'NAME'))\nconsume(hasattr(target, 'NAME'))\n"
3454 ),
3455 None
3456 );
3457 assert_eq!(
3458 mutation("import sys\nconsume(sys.modules[module_name].NAME)\n"),
3459 None
3460 );
3461 assert!(matches!(
3462 mutation("import sys\ntarget = sys.modules[module_name]\ntarget.NAME = fake\n"),
3463 Some(PythonModuleMemberMutation::Write(_))
3464 ));
3465 }
3466
3467 #[test]
3468 #[hardwon("#4622: a dominating string filter excludes unrelated dynamic sys.modules aliases")]
3469 fn dynamic_sys_modules_keys_respect_dominating_string_filters() {
3470 assert_eq!(
3471 mutation_for_module(
3472 "import sys\ndef patch():\n for module_name in module_names:\n if 'huggingface' in module_name or module_name.startswith('tqdm'):\n target = sys.modules[module_name]\n target.NAME = fake\n",
3473 "pkg.M",
3474 "backend.backends",
3475 ),
3476 None
3477 );
3478 assert!(matches!(
3479 mutation_for_module(
3480 "import sys\ndef patch():\n for module_name in module_names:\n if module_name.startswith('backend'):\n target = sys.modules[module_name]\n target.NAME = fake\n",
3481 "pkg.M",
3482 "backend.backends",
3483 ),
3484 Some(PythonModuleMemberMutation::Write(_))
3485 ));
3486 assert!(matches!(
3487 mutation_for_module(
3488 "import sys\ndef patch(allowed):\n for module_name in module_names:\n if module_name in allowed:\n target = sys.modules[module_name]\n target.NAME = fake\n",
3489 "pkg.M",
3490 "backend.backends",
3491 ),
3492 Some(PythonModuleMemberMutation::Write(_))
3493 ));
3494 assert!(matches!(
3495 mutation_for_module(
3496 "import sys\ndef patch():\n for module_name in module_names:\n if 'huggingface' in module_name:\n module_name = 'backend.backends'\n target = sys.modules[module_name]\n target.NAME = fake\n",
3497 "pkg.M",
3498 "backend.backends",
3499 ),
3500 Some(PythonModuleMemberMutation::Write(_))
3501 ));
3502 assert_eq!(
3503 mutation_for_module(
3504 "import sys\ndef patch():\n for module_name in module_names:\n if not ('huggingface' in module_name):\n continue\n target = sys.modules[module_name]\n target.NAME = fake\n",
3505 "pkg.M",
3506 "backend.backends",
3507 ),
3508 None
3509 );
3510 }
3511}