[pysrc2cpg] Fix **kwargs handling, walrus operator and comprehension test coverage (#5910)

* [pysrc2cpg] Fix **kwargs handling, add match pattern AST nodes, and test coverage

- Fix **kwargs unpacking to preserve dict argument for taint tracking
- Convert match statement patterns to proper AST nodes in case body blocks
- Add walrus operator tests verifying expression semantics
- Add comprehensive comprehension tests (list, set, dict, generator)

* Address review feedback: fix test conventions and kwargs arg name

- Rename (named expression) to (assignment expression) in walrus test
- Remove lazy and explicit test.py filename from new test blocks
- Use val keywordDictArgName = "<keyword_dict>" instead of "**" string literal
- Revert match pattern AST node changes (needs proper semantic design)
- Fix walrus operator test to match actual generated code

* Fix scalafmt formatting in PythonAstVisitor

Author: allsmog

joern-cli/frontends/pysrc2cpg/src/main/scala/io/joern/pysrc2cpg/PythonAstVisitor.scala

@@ -1,6 +1,6 @@
 package io.joern.pysrc2cpg
 
-import PythonAstVisitor.{logger, metaClassSuffix, noLineAndColumn}
+import PythonAstVisitor.{keywordDictArgName, logger, metaClassSuffix, noLineAndColumn}
 import io.joern.pysrc2cpg.memop.*
 import io.joern.pysrc2cpg.memop.MemoryOperation.{Del, Load, Store}
 import io.joern.x2cpg.frontendspecific.pysrc2cpg.Constants.builtinPrefix
@@ -1444,7 +1444,6 @@ class PythonAstVisitor(
     createNAryOperatorCall(boolOpToCodeAndFullName(boolOp.op), operandNodes, lineAndColOf(boolOp))
   }
 
-  // TODO test
   def convert(namedExpr: ast.NamedExpr): NewNode = {
     val targetNode = convert(namedExpr.target)
     val valueNode  = convert(namedExpr.value)
@@ -1848,13 +1847,14 @@ class PythonAstVisitor(
     */
   def convert(call: ast.Call): nodes.NewNode = {
     val argumentNodes = call.args.map(convert).toSeq
-    val keywordArgNodes = call.keywords.flatMap { keyword =>
+    val keywordArgNodes = call.keywords.map { keyword =>
       if (keyword.arg.isDefined) {
-        Some((keyword.arg.get, convert(keyword.value)))
+        (keyword.arg.get, convert(keyword.value))
       } else {
         // keyword.arg == None. This is the case for func(**dict) style arguments.
-        // TODO implement handling for this case.
-        None
+        // We use a synthetic argument name to preserve the unpacked dict as an argument
+        // in the CPG so that data flow tracking can follow through it.
+        (keywordDictArgName, convert(keyword.value))
       }
     }
 
@@ -2184,8 +2184,9 @@ class PythonAstVisitor(
 object PythonAstVisitor {
   private val logger = LoggerFactory.getLogger(getClass)
 
-  val typingPrefix    = "typing."
-  val metaClassSuffix = "<meta>"
+  val typingPrefix       = "typing."
+  val metaClassSuffix    = "<meta>"
+  val keywordDictArgName = "<keyword_dict>"
 
   val noLineAndColumn = LineAndColumn(-1, -1, -1, -1, -1, -1)
 

joern-cli/frontends/pysrc2cpg/src/test/scala/io/joern/pysrc2cpg/cpg/AssignCpgTests.scala

@@ -203,6 +203,33 @@ class AssignCpgTests extends PySrc2CpgFixture with Matchers {
     }
   }
 
+  "walrus operator (assignment expression)" should {
+    val cpg = code("""
+        |if (x := 10) > 5:
+        |  print(x)
+        |""".stripMargin)
+
+    "create an assignment node for the walrus operator" in {
+      val assignCall = cpg.call.methodFullName(Operators.assignment).codeExact("x = 10").head
+      assignCall.code shouldBe "x = 10"
+      assignCall.dispatchType shouldBe DispatchTypes.STATIC_DISPATCH
+    }
+
+    "have correct assignment target and value" in {
+      val assignCall = cpg.call.methodFullName(Operators.assignment).codeExact("x = 10").head
+      assignCall.argument
+        .argumentIndex(1)
+        .isIdentifier
+        .head
+        .code shouldBe "x"
+      assignCall.argument
+        .argumentIndex(2)
+        .isLiteral
+        .head
+        .code shouldBe "10"
+    }
+  }
+
   "augmented assign" should {
     val cpg = code("""x += y""".stripMargin)
 

joern-cli/frontends/pysrc2cpg/src/test/scala/io/joern/pysrc2cpg/cpg/CallCpgTests.scala

@@ -194,6 +194,36 @@ class CallCpgTests extends PySrc2CpgFixture(withOssDataflow = false) {
     }
   }
 
+  "call with **kwargs unpacking" should {
+    val cpg = code("""func(a, **my_dict)""".stripMargin)
+
+    "test call node properties" in {
+      val callNode = cpg.call.codeExact("func(a, **my_dict)").head
+      callNode.name shouldBe "func"
+      callNode.dispatchType shouldBe DispatchTypes.DYNAMIC_DISPATCH
+    }
+
+    "test that the unpacked dict appears as an argument" in {
+      val callNode  = cpg.call.codeExact("func(a, **my_dict)").head
+      val kwargsArg = callNode.argument.isIdentifier.nameExact("my_dict").head
+      kwargsArg.code shouldBe "my_dict"
+      kwargsArg.argumentIndex shouldBe -1
+      kwargsArg.argumentName shouldBe Some("<keyword_dict>")
+    }
+  }
+
+  "call on member with **kwargs unpacking" should {
+    val cpg = code("""x.func(a, **my_dict)""".stripMargin)
+
+    "test that the unpacked dict appears as an argument" in {
+      val callNode  = cpg.call.codeExact("x.func(a, **my_dict)").head
+      val kwargsArg = callNode.argument.isIdentifier.nameExact("my_dict").head
+      kwargsArg.code shouldBe "my_dict"
+      kwargsArg.argumentIndex shouldBe -1
+      kwargsArg.argumentName shouldBe Some("<keyword_dict>")
+    }
+  }
+
   "call from a function defined from an imported module" should {
 
     lazy val cpg = code(

joern-cli/frontends/pysrc2cpg/src/test/scala/io/joern/pysrc2cpg/cpg/ComprehensionCpgTests.scala

@@ -0,0 +1,58 @@
+package io.joern.pysrc2cpg.cpg
+
+import io.joern.pysrc2cpg.testfixtures.PySrc2CpgFixture
+import io.shiftleft.codepropertygraph.generated.Operators
+import io.shiftleft.semanticcpg.language.*
+
+class ComprehensionCpgTests extends PySrc2CpgFixture(withOssDataflow = false) {
+
+  "list comprehension" should {
+    val cpg = code("""x = [i * 2 for i in range(10)]""".stripMargin)
+
+    "create a call to range" in {
+      cpg.call.codeExact("range(10)").head.name shouldBe "range"
+    }
+
+    "create a multiplication operation" in {
+      cpg.call.methodFullName(Operators.multiplication).codeExact("i * 2").size shouldBe 1
+    }
+  }
+
+  "set comprehension" should {
+    val cpg = code("""x = {i * 2 for i in range(10)}""".stripMargin)
+
+    "create a call to range" in {
+      cpg.call.codeExact("range(10)").head.name shouldBe "range"
+    }
+
+    "create a multiplication operation" in {
+      cpg.call.methodFullName(Operators.multiplication).codeExact("i * 2").size shouldBe 1
+    }
+  }
+
+  "dict comprehension" should {
+    val cpg = code("""x = {k: v for k, v in items.items()}""".stripMargin)
+
+    "create a call to items()" in {
+      cpg.call.codeExact("items.items()").head.name shouldBe "items"
+    }
+
+    "have identifiers for k and v" in {
+      cpg.identifier.nameExact("k").size should be > 0
+      cpg.identifier.nameExact("v").size should be > 0
+    }
+  }
+
+  "generator expression" should {
+    val cpg = code("""x = sum(i * 2 for i in range(10))""".stripMargin)
+
+    "create a call to sum" in {
+      cpg.call.codeExact("sum(i * 2 for i in range(10))").head.name shouldBe "sum"
+    }
+
+    "create a call to range" in {
+      cpg.call.codeExact("range(10)").head.name shouldBe "range"
+    }
+  }
+
+}