Fix PythonOperator when DAG has hyphen in name (#42993)

airflow/utils/file.py

@@ -389,6 +389,6 @@ def get_unique_dag_module_name(file_path: str) -> str:
     """Return a unique module name in the format unusual_prefix_{sha1 of module's file path}_{original module name}."""
     if isinstance(file_path, str):
         path_hash = hashlib.sha1(file_path.encode("utf-8")).hexdigest()
-        org_mod_name = Path(file_path).stem
+        org_mod_name = re2.sub(r"[.-]", "_", Path(file_path).stem)
         return MODIFIED_DAG_MODULE_NAME.format(path_hash=path_hash, module_name=org_mod_name)
     raise ValueError("file_path should be a string to generate unique module name")

tests/utils/test_file.py

@@ -211,3 +211,21 @@ def test_get_modules_from_invalid_file(self):
         modules = list(file_utils.iter_airflow_imports(file_path))
 
         assert len(modules) == 0
+
+
+@pytest.mark.parametrize(
+    "edge_filename, expected_modification",
+    [
+        ("test_dag.py", "unusual_prefix_mocked_path_hash_sha1_test_dag"),
+        ("test-dag.py", "unusual_prefix_mocked_path_hash_sha1_test_dag"),
+        ("test-dag-1.py", "unusual_prefix_mocked_path_hash_sha1_test_dag_1"),
+        ("test-dag_1.py", "unusual_prefix_mocked_path_hash_sha1_test_dag_1"),
+        ("test-dag.dev.py", "unusual_prefix_mocked_path_hash_sha1_test_dag_dev"),
+        ("test_dag.prod.py", "unusual_prefix_mocked_path_hash_sha1_test_dag_prod"),
+    ],
+)
+def test_get_unique_dag_module_name(edge_filename, expected_modification):
+    with mock.patch("hashlib.sha1") as mocked_sha1:
+        mocked_sha1.return_value.hexdigest.return_value = "mocked_path_hash_sha1"
+        modify_module_name = file_utils.get_unique_dag_module_name(edge_filename)
+        assert modify_module_name == expected_modification