Fix AKS permission error in restricted env (#1051)

## Description
~shutil.copy includes permission copying via chmod.
If the user lacks permission to run chmod, a PermissionError occurs.
To avoid this, we split the operation into two steps:
first, copy the file contents; then, copy metadata if feasible without
raising exceptions.
Step 1: Copy file contents (no metadata)
Step 2: Copy file metadata (permission bits and other metadata) without
raising exception~

use shutil.copyfile(...) instead of shutil.copy(...) to avoid running
chmod

## Related Issue(s)

closes: #1008

## Breaking Change?

No

## Checklist

- [ ] I have made corresponding changes to the documentation (if
required)
- [ ] I have added tests that prove my fix is effective or that my
feature works

cosmos/cache.py

@@ -107,8 +107,8 @@ def _update_partial_parse_cache(latest_partial_parse_filepath: Path, cache_dir:
     manifest_path = get_partial_parse_path(cache_dir).parent / DBT_MANIFEST_FILE_NAME
     latest_manifest_filepath = latest_partial_parse_filepath.parent / DBT_MANIFEST_FILE_NAME
 
-    shutil.copy(str(latest_partial_parse_filepath), str(cache_path))
-    shutil.copy(str(latest_manifest_filepath), str(manifest_path))
+    shutil.copyfile(str(latest_partial_parse_filepath), str(cache_path))
+    shutil.copyfile(str(latest_manifest_filepath), str(manifest_path))
 
 
 def patch_partial_parse_content(partial_parse_filepath: Path, project_path: Path) -> bool:

tests/test_cache.py

@@ -4,13 +4,18 @@
 import time
 from datetime import datetime
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import call, patch
 
 import pytest
 from airflow import DAG
 from airflow.utils.task_group import TaskGroup
 
-from cosmos.cache import _copy_partial_parse_to_project, _create_cache_identifier, _get_latest_partial_parse
+from cosmos.cache import (
+    _copy_partial_parse_to_project,
+    _create_cache_identifier,
+    _get_latest_partial_parse,
+    _update_partial_parse_cache,
+)
 from cosmos.constants import DBT_PARTIAL_PARSE_FILE_NAME, DBT_TARGET_DIR_NAME
 
 START_DATE = datetime(2024, 4, 16)
@@ -74,7 +79,6 @@ def test_get_latest_partial_parse(tmp_path):
 
 @patch("cosmos.cache.msgpack.unpack", side_effect=ValueError)
 def test__copy_partial_parse_to_project_msg_fails_msgpack(mock_unpack, tmp_path, caplog):
-    # setup
     caplog.set_level(logging.INFO)
     source_dir = tmp_path / DBT_TARGET_DIR_NAME
     source_dir.mkdir()
@@ -86,3 +90,25 @@ def test__copy_partial_parse_to_project_msg_fails_msgpack(mock_unpack, tmp_path,
         _copy_partial_parse_to_project(partial_parse_filepath, Path(tmp_dir))
 
     assert "Unable to patch the partial_parse.msgpack file due to ValueError()" in caplog.text
+
+
+@patch("cosmos.cache.shutil.copyfile")
+@patch("cosmos.cache.get_partial_parse_path")
+def test_update_partial_parse_cache(mock_get_partial_parse_path, mock_copyfile):
+    mock_get_partial_parse_path.side_effect = lambda cache_dir: cache_dir / "partial_parse.yml"
+
+    latest_partial_parse_filepath = Path("/path/to/latest_partial_parse.yml")
+    cache_dir = Path("/path/to/cache_directory")
+
+    # Expected paths
+    cache_path = cache_dir / "partial_parse.yml"
+    manifest_path = cache_dir / "manifest.json"
+
+    _update_partial_parse_cache(latest_partial_parse_filepath, cache_dir)
+
+    # Assert shutil.copyfile was called twice with the correct arguments
+    calls = [
+        call(str(latest_partial_parse_filepath), str(cache_path)),
+        call(str(latest_partial_parse_filepath.parent / "manifest.json"), str(manifest_path)),
+    ]
+    mock_copyfile.assert_has_calls(calls)