diff --git a/samcli/commands/build/build_context.py b/samcli/commands/build/build_context.py index 817547402c..8dc51e31e1 100644 --- a/samcli/commands/build/build_context.py +++ b/samcli/commands/build/build_context.py @@ -408,6 +408,10 @@ def manifest_path_override(self) -> Optional[str]: def mode(self) -> Optional[str]: return self._mode + @property + def use_base_dir(self) -> bool: + return self._use_raw_codeuri + @property def resources_to_build(self) -> ResourcesToBuildCollector: """ diff --git a/samcli/lib/providers/provider.py b/samcli/lib/providers/provider.py index ea816af7b9..3c07a082cb 100644 --- a/samcli/lib/providers/provider.py +++ b/samcli/lib/providers/provider.py @@ -603,22 +603,22 @@ def get_parent_stack(child_stack: "Stack", stacks: List["Stack"]) -> Optional["S return None @staticmethod - def get_stack_by_logical_id(logical_id: str, stacks: List["Stack"]) -> Optional["Stack"]: + def get_stack_by_full_path(full_path: str, stacks: List["Stack"]) -> Optional["Stack"]: """ - Return the stack with given logical id + Return the stack with given full path Parameters ---------- - logical_id str - logical_id of the stack + full_path str + full path of the stack like ChildStack/ChildChildStack stacks : List[Stack] a list of stack for searching Returns ------- Stack - The stack with the given logical id + The stack with the given full path """ for stack in stacks: - if stack.name == logical_id: + if stack.stack_path == full_path: return stack return None diff --git a/samcli/lib/providers/sam_function_provider.py b/samcli/lib/providers/sam_function_provider.py index bcf18e4adf..a63e953862 100644 --- a/samcli/lib/providers/sam_function_provider.py +++ b/samcli/lib/providers/sam_function_provider.py @@ -616,7 +616,9 @@ def _locate_layer_from_nested( # pylint: disable=too-many-return-statements LOG.debug("Search layer %s in child stack", layer_reference) child_stacks = Stack.get_child_stacks(stack, stacks) - child_stack = Stack.get_stack_by_logical_id(layer_stack_reference, child_stacks) + stack_prefix = stack.stack_path + "/" if stack.stack_path else "" + stack_path = stack_prefix + layer_stack_reference + child_stack = Stack.get_stack_by_full_path(stack_path, child_stacks) if not child_stack: LOG.debug("Child stack not found, layer can not be located in templates") return None diff --git a/samcli/lib/sync/flows/generic_api_sync_flow.py b/samcli/lib/sync/flows/generic_api_sync_flow.py index 16acda723e..37155916b9 100644 --- a/samcli/lib/sync/flows/generic_api_sync_flow.py +++ b/samcli/lib/sync/flows/generic_api_sync_flow.py @@ -1,9 +1,9 @@ """SyncFlow interface for HttpApi and RestApi""" import logging from pathlib import Path -from typing import Any, Dict, List, Optional, TYPE_CHECKING, cast +from typing import Any, Dict, List, Optional, TYPE_CHECKING -from samcli.lib.sync.sync_flow import SyncFlow, ResourceAPICall +from samcli.lib.sync.sync_flow import SyncFlow, ResourceAPICall, get_definition_path from samcli.lib.providers.provider import Stack, get_resource_by_id, ResourceIdentifier # BuildContext and DeployContext will only be imported for type checking to improve performance @@ -20,7 +20,7 @@ class GenericApiSyncFlow(SyncFlow): _api_client: Any _api_identifier: str - _definition_uri: Optional[str] + _definition_uri: Optional[Path] _stacks: List[Stack] _swagger_body: Optional[bytes] @@ -65,19 +65,21 @@ def gather_resources(self) -> None: def _process_definition_file(self) -> Optional[bytes]: if self._definition_uri is None: return None - with open(self._definition_uri, "rb") as swagger_file: + with open(str(self._definition_uri), "rb") as swagger_file: swagger_body = swagger_file.read() return swagger_body - def _get_definition_file(self, api_identifier: str) -> Optional[str]: + def _get_definition_file(self, api_identifier: str) -> Optional[Path]: api_resource = get_resource_by_id(self._stacks, ResourceIdentifier(api_identifier)) - if api_resource is None: + if not api_resource: return None - properties = api_resource.get("Properties", {}) - definition_file = properties.get("DefinitionUri") - if self._build_context.base_dir and definition_file: - definition_file = str(Path(self._build_context.base_dir).joinpath(definition_file)) - return cast(Optional[str], definition_file) + return get_definition_path( + api_resource, + self._api_identifier, + self._build_context.use_base_dir, + self._build_context.base_dir, + self._stacks, + ) def compare_remote(self) -> bool: return False diff --git a/samcli/lib/sync/flows/stepfunctions_sync_flow.py b/samcli/lib/sync/flows/stepfunctions_sync_flow.py index 09479288f8..35bd31bc7d 100644 --- a/samcli/lib/sync/flows/stepfunctions_sync_flow.py +++ b/samcli/lib/sync/flows/stepfunctions_sync_flow.py @@ -1,11 +1,11 @@ """Base SyncFlow for StepFunctions""" import logging from pathlib import Path -from typing import Any, Dict, List, TYPE_CHECKING, cast, Optional +from typing import Any, Dict, List, TYPE_CHECKING, Optional from samcli.lib.providers.provider import Stack, get_resource_by_id, ResourceIdentifier -from samcli.lib.sync.sync_flow import SyncFlow, ResourceAPICall +from samcli.lib.sync.sync_flow import SyncFlow, ResourceAPICall, get_definition_path from samcli.lib.sync.exceptions import InfraSyncRequiredError from samcli.lib.providers.exceptions import MissingLocalDefinition @@ -21,7 +21,7 @@ class StepFunctionsSyncFlow(SyncFlow): _state_machine_identifier: str _stepfunctions_client: Any _stacks: List[Stack] - _definition_uri: Optional[str] + _definition_uri: Optional[Path] _states_definition: Optional[str] def __init__( @@ -75,18 +75,20 @@ def gather_resources(self) -> None: def _process_definition_file(self) -> Optional[str]: if self._definition_uri is None: return None - with open(self._definition_uri, "r", encoding="utf-8") as states_file: + with open(str(self._definition_uri), "r", encoding="utf-8") as states_file: states_data = states_file.read() return states_data - def _get_definition_file(self, state_machine_identifier: str) -> Optional[str]: - if self._resource is None: + def _get_definition_file(self, state_machine_identifier: str) -> Optional[Path]: + if not self._resource: return None - properties = self._resource.get("Properties", {}) - definition_file = properties.get("DefinitionUri") - if self._build_context.base_dir: - definition_file = str(Path(self._build_context.base_dir).joinpath(definition_file)) - return cast(Optional[str], definition_file) + return get_definition_path( + self._resource, + state_machine_identifier, + self._build_context.use_base_dir, + self._build_context.base_dir, + self._stacks, + ) def compare_remote(self) -> bool: # Not comparing with remote right now, instead only making update api calls diff --git a/samcli/lib/sync/sync_flow.py b/samcli/lib/sync/sync_flow.py index 6885679187..3f7b5953a1 100644 --- a/samcli/lib/sync/sync_flow.py +++ b/samcli/lib/sync/sync_flow.py @@ -3,6 +3,7 @@ from abc import ABC, abstractmethod from enum import Enum +from pathlib import Path from threading import Lock from typing import Any, Dict, List, NamedTuple, Optional, TYPE_CHECKING, cast, Set from boto3.session import Session @@ -315,3 +316,40 @@ def execute(self) -> List["SyncFlow"]: dependencies = self.gather_dependencies() LOG.debug("%sFinished", self.log_prefix) return dependencies + + +def get_definition_path( + resource: Dict, identifier: str, use_base_dir: bool, base_dir: str, stacks: List[Stack] +) -> Optional[Path]: + """ + A helper method used by non-function sync flows to resolve definition file path + that are relative to the child stack to absolute path for nested stacks + + Parameters + ------- + resource: Dict + The resource's template dict + identifier: str + The logical ID identifier of the resource + use_base_dir: bool + Whether or not the base_dir option was used + base_dir: str + Base directory if provided, otherwise the root template directory + stacks: List[Stack] + The list of stacks for the application + + Returns + ------- + Optional[Path] + A resolved absolute path for the definition file + """ + properties = resource.get("Properties", {}) + definition_file = properties.get("DefinitionUri") + definition_path = None + if definition_file: + definition_path = Path(base_dir).joinpath(definition_file) + if not use_base_dir: + child_stack = Stack.get_stack_by_full_path(ResourceIdentifier(identifier).stack_path, stacks) + if child_stack: + definition_path = Path(child_stack.location).parent.joinpath(definition_file) + return definition_path diff --git a/tests/integration/sync/test_sync_code.py b/tests/integration/sync/test_sync_code.py index d34a2efefc..c2ae4ceb47 100644 --- a/tests/integration/sync/test_sync_code.py +++ b/tests/integration/sync/test_sync_code.py @@ -410,7 +410,6 @@ def test_sync_nested_function_layer_race_condition(self): self.assertIn("extra_message", lambda_response) self.assertEqual(lambda_response.get("message"), "10") - @pytest.mark.skip(reason="Currently not properly supported") def test_sync_code_nested_rest_api(self): shutil.rmtree( TestSyncCodeBase.temp_dir.joinpath("child_stack").joinpath("child_child_stack").joinpath("apigateway"), @@ -447,7 +446,6 @@ def test_sync_code_nested_rest_api(self): rest_api = self.stack_resources.get(AWS_APIGATEWAY_RESTAPI)[0] self.assertEqual(self._get_api_message(rest_api), '{"message": "hello 2"}') - @pytest.mark.skip(reason="Currently not properly supported") def test_sync_code_nested_state_machine(self): shutil.rmtree( TestSyncCodeBase.temp_dir.joinpath("child_stack").joinpath("child_child_stack").joinpath("statemachine"), diff --git a/tests/unit/commands/buildcmd/test_build_context.py b/tests/unit/commands/buildcmd/test_build_context.py index ef9f2ce913..dde06145d2 100644 --- a/tests/unit/commands/buildcmd/test_build_context.py +++ b/tests/unit/commands/buildcmd/test_build_context.py @@ -467,6 +467,7 @@ def test_must_return_many_functions_to_build( self.assertEqual(context.stacks, [stack]) self.assertEqual(context.manifest_path_override, os.path.abspath("manifest_path")) self.assertEqual(context.mode, "buildmode") + self.assertFalse(context.use_base_dir) self.assertFalse(context.is_building_specific_resource) resources_to_build = context.resources_to_build self.assertEqual(resources_to_build.functions, [func1, func2, func6]) diff --git a/tests/unit/commands/local/lib/test_provider.py b/tests/unit/commands/local/lib/test_provider.py index 9dbca89e97..6e544566f2 100644 --- a/tests/unit/commands/local/lib/test_provider.py +++ b/tests/unit/commands/local/lib/test_provider.py @@ -802,26 +802,39 @@ def test_get_resource_full_path_by_id(self, resource_id, expected_full_path): class TestGetStack(TestCase): - root_stack = Stack("", "", "template.yaml", None, {}) - child_stack = Stack("", "child", "template.yaml", None, {}) + root_stack = Stack("", "Root", "template.yaml", None, {}) + child_stack = Stack("Root", "Child", "root_stack/template.yaml", None, {}) + child_child_stack = Stack("Root/Child", "ChildChild", "root_stack/child_stack/template.yaml", None, {}) def test_get_parent_stack(self): - stack = Stack.get_parent_stack(self.child_stack, [self.root_stack, self.child_stack]) + stack = Stack.get_parent_stack(self.child_stack, [self.root_stack, self.child_stack, self.child_child_stack]) self.assertEqual(stack, self.root_stack) - stack = Stack.get_parent_stack(self.root_stack, [self.root_stack, self.child_stack]) + stack = Stack.get_parent_stack(self.root_stack, [self.root_stack, self.child_stack, self.child_child_stack]) self.assertIsNone(stack) - def test_get_stack_by_logical_id(self): - stack = Stack.get_stack_by_logical_id("child", [self.root_stack, self.child_stack]) + def test_get_stack_by_full_path(self): + stack = Stack.get_stack_by_full_path("Root/Child", [self.root_stack, self.child_stack, self.child_child_stack]) self.assertEqual(stack, self.child_stack) - stack = Stack.get_stack_by_logical_id("not_exist", [self.root_stack, self.child_stack]) + stack = Stack.get_stack_by_full_path("Root", [self.root_stack, self.child_stack, self.child_child_stack]) + self.assertEqual(stack, self.root_stack) + + stack = Stack.get_stack_by_full_path("Child/Child", [self.root_stack, self.child_stack, self.child_child_stack]) self.assertIsNone(stack) def test_get_child_stacks(self): - stack_list = Stack.get_child_stacks(self.root_stack, [self.root_stack, self.child_stack]) + stack_list = Stack.get_child_stacks( + self.root_stack, [self.root_stack, self.child_stack, self.child_child_stack] + ) self.assertEqual(stack_list, [self.child_stack]) - stack_list = Stack.get_child_stacks(self.child_stack, [self.root_stack, self.child_stack]) + stack_list = Stack.get_child_stacks( + self.child_stack, [self.root_stack, self.child_stack, self.child_child_stack] + ) + self.assertEqual(stack_list, [self.child_child_stack]) + + stack_list = Stack.get_child_stacks( + self.child_child_stack, [self.root_stack, self.child_stack, self.child_child_stack] + ) self.assertEqual(stack_list, []) diff --git a/tests/unit/lib/sync/flows/test_http_api_sync_flow.py b/tests/unit/lib/sync/flows/test_http_api_sync_flow.py index 02da1a9c81..b4790c6637 100644 --- a/tests/unit/lib/sync/flows/test_http_api_sync_flow.py +++ b/tests/unit/lib/sync/flows/test_http_api_sync_flow.py @@ -1,6 +1,7 @@ from unittest import TestCase from unittest.mock import MagicMock, mock_open, patch from pathlib import Path +from samcli.lib.providers.provider import Stack from samcli.lib.sync.flows.http_api_sync_flow import HttpApiSyncFlow from samcli.lib.providers.exceptions import MissingLocalDefinition @@ -45,31 +46,51 @@ def test_sync_direct(self, session_mock): sync_flow._api_client.reimport_api.assert_called_once_with(ApiId="PhysicalApi1", Body='{"key": "value"}') @patch("samcli.lib.sync.flows.generic_api_sync_flow.get_resource_by_id") - @patch("samcli.lib.sync.flows.generic_api_sync_flow.Path.joinpath") - def test_get_definition_file(self, join_path_mock, get_resource_mock): + @patch("samcli.lib.sync.flows.generic_api_sync_flow.get_definition_path") + def test_get_definition_file(self, get_path_mock, get_resource_mock): sync_flow = self.create_sync_flow() - sync_flow._build_context.base_dir = None - join_path_mock.return_value = "test_uri" + sync_flow._build_context.use_base_dir = False + sync_flow._build_context.base_dir = "base_dir" get_resource_mock.return_value = {"Properties": {"DefinitionUri": "test_uri"}} - result_uri = sync_flow._get_definition_file("test") + get_path_mock.return_value = Path("base_dir").joinpath("test_uri") + + result_uri = sync_flow._get_definition_file(sync_flow._api_identifier) - self.assertEqual(result_uri, "test_uri") + get_path_mock.assert_called_with( + {"Properties": {"DefinitionUri": "test_uri"}}, + sync_flow._api_identifier, + False, + "base_dir", + sync_flow._stacks, + ) + self.assertEqual(result_uri, Path("base_dir").joinpath("test_uri")) - get_resource_mock.return_value = {"Properties": {}} - result_uri = sync_flow._get_definition_file("test") + get_resource_mock.return_value = {} + result_uri = sync_flow._get_definition_file(sync_flow._api_identifier) self.assertEqual(result_uri, None) @patch("samcli.lib.sync.flows.generic_api_sync_flow.get_resource_by_id") - def test_get_definition_file_with_base_dir(self, get_resource_mock): + @patch("samcli.lib.sync.flows.generic_api_sync_flow.get_definition_path") + def test_get_definition_file_with_base_dir(self, get_path_mock, get_resource_mock): sync_flow = self.create_sync_flow() + sync_flow._build_context.use_base_dir = True sync_flow._build_context.base_dir = "base_dir" get_resource_mock.return_value = {"Properties": {"DefinitionUri": "test_uri"}} - result_uri = sync_flow._get_definition_file("test") + get_path_mock.return_value = Path("base_dir").joinpath("test_uri") - self.assertEqual(result_uri, str(Path("base_dir").joinpath("test_uri"))) + result_uri = sync_flow._get_definition_file(sync_flow._api_identifier) + + get_path_mock.assert_called_with( + {"Properties": {"DefinitionUri": "test_uri"}}, + sync_flow._api_identifier, + True, + "base_dir", + sync_flow._stacks, + ) + self.assertEqual(result_uri, Path("base_dir").joinpath("test_uri")) def test_process_definition_file(self): sync_flow = self.create_sync_flow() diff --git a/tests/unit/lib/sync/flows/test_rest_api_sync_flow.py b/tests/unit/lib/sync/flows/test_rest_api_sync_flow.py index af338bee5c..4a68eae98a 100644 --- a/tests/unit/lib/sync/flows/test_rest_api_sync_flow.py +++ b/tests/unit/lib/sync/flows/test_rest_api_sync_flow.py @@ -7,7 +7,7 @@ from samcli.lib.utils.colors import Colored from samcli.lib.sync.flows.rest_api_sync_flow import RestApiSyncFlow from samcli.lib.providers.exceptions import MissingLocalDefinition -from samcli.lib.providers.provider import ResourceIdentifier +from samcli.lib.providers.provider import ResourceIdentifier, Stack class TestRestApiSyncFlow(TestCase): @@ -263,31 +263,51 @@ def test_delete_deployment_failure(self, session_mock): ) @patch("samcli.lib.sync.flows.generic_api_sync_flow.get_resource_by_id") - @patch("samcli.lib.sync.flows.generic_api_sync_flow.Path.joinpath") - def test_get_definition_file(self, join_path_mock, get_resource_mock): + @patch("samcli.lib.sync.flows.generic_api_sync_flow.get_definition_path") + def test_get_definition_file(self, get_path_mock, get_resource_mock): sync_flow = self.create_sync_flow() - sync_flow._build_context.base_dir = None - join_path_mock.return_value = "test_uri" + sync_flow._build_context.use_base_dir = False + sync_flow._build_context.base_dir = "base_dir" get_resource_mock.return_value = {"Properties": {"DefinitionUri": "test_uri"}} - result_uri = sync_flow._get_definition_file("test") + get_path_mock.return_value = Path("base_dir").joinpath("test_uri") + + result_uri = sync_flow._get_definition_file(sync_flow._api_identifier) - self.assertEqual(result_uri, "test_uri") + get_path_mock.assert_called_with( + {"Properties": {"DefinitionUri": "test_uri"}}, + sync_flow._api_identifier, + False, + "base_dir", + sync_flow._stacks, + ) + self.assertEqual(result_uri, Path("base_dir").joinpath("test_uri")) - get_resource_mock.return_value = {"Properties": {}} - result_uri = sync_flow._get_definition_file("test") + get_resource_mock.return_value = {} + result_uri = sync_flow._get_definition_file(sync_flow._api_identifier) self.assertEqual(result_uri, None) @patch("samcli.lib.sync.flows.generic_api_sync_flow.get_resource_by_id") - def test_get_definition_file_with_base_dir(self, get_resource_mock): + @patch("samcli.lib.sync.flows.generic_api_sync_flow.get_definition_path") + def test_get_definition_file_with_base_dir(self, get_path_mock, get_resource_mock): sync_flow = self.create_sync_flow() + sync_flow._build_context.use_base_dir = True sync_flow._build_context.base_dir = "base_dir" get_resource_mock.return_value = {"Properties": {"DefinitionUri": "test_uri"}} - result_uri = sync_flow._get_definition_file("test") + get_path_mock.return_value = Path("base_dir").joinpath("test_uri") - self.assertEqual(result_uri, str(Path("base_dir").joinpath("test_uri"))) + result_uri = sync_flow._get_definition_file(sync_flow._api_identifier) + + get_path_mock.assert_called_with( + {"Properties": {"DefinitionUri": "test_uri"}}, + sync_flow._api_identifier, + True, + "base_dir", + sync_flow._stacks, + ) + self.assertEqual(result_uri, Path("base_dir").joinpath("test_uri")) def test_process_definition_file(self): sync_flow = self.create_sync_flow() diff --git a/tests/unit/lib/sync/flows/test_stepfunctions_sync_flow.py b/tests/unit/lib/sync/flows/test_stepfunctions_sync_flow.py index 6d4bebf2f2..6af77cd60b 100644 --- a/tests/unit/lib/sync/flows/test_stepfunctions_sync_flow.py +++ b/tests/unit/lib/sync/flows/test_stepfunctions_sync_flow.py @@ -1,6 +1,7 @@ from unittest import TestCase -from unittest.mock import ANY, MagicMock, mock_open, patch +from unittest.mock import MagicMock, mock_open, patch from pathlib import Path +from samcli.lib.providers.provider import Stack from samcli.lib.sync.flows.stepfunctions_sync_flow import StepFunctionsSyncFlow from samcli.lib.sync.exceptions import InfraSyncRequiredError @@ -54,32 +55,50 @@ def test_sync_direct(self, session_mock): stateMachineArn="PhysicalId1", definition='{"key": "value"}' ) - @patch("samcli.lib.sync.flows.stepfunctions_sync_flow.get_resource_by_id") - @patch("samcli.lib.sync.flows.stepfunctions_sync_flow.Path.joinpath") - def test_get_definition_file(self, join_path_mock, get_resource_mock): + @patch("samcli.lib.sync.flows.stepfunctions_sync_flow.get_definition_path") + def test_get_definition_file(self, get_path_mock): sync_flow = self.create_sync_flow() - sync_flow._build_context.base_dir = None - join_path_mock.return_value = "test_uri" + sync_flow._build_context.use_base_dir = False + sync_flow._build_context.base_dir = "base_dir" sync_flow._resource = {"Properties": {"DefinitionUri": "test_uri"}} - result_uri = sync_flow._get_definition_file("test") + get_path_mock.return_value = Path("base_dir").joinpath("test_uri") + + result_uri = sync_flow._get_definition_file(sync_flow._state_machine_identifier) - self.assertEqual(result_uri, "test_uri") + get_path_mock.assert_called_with( + {"Properties": {"DefinitionUri": "test_uri"}}, + sync_flow._state_machine_identifier, + False, + "base_dir", + sync_flow._stacks, + ) + self.assertEqual(result_uri, Path("base_dir").joinpath("test_uri")) - sync_flow._resource = {"Properties": {}} - result_uri = sync_flow._get_definition_file("test") + sync_flow._resource = {} + result_uri = sync_flow._get_definition_file(sync_flow._state_machine_identifier) self.assertEqual(result_uri, None) - @patch("samcli.lib.sync.flows.stepfunctions_sync_flow.get_resource_by_id") - def test_get_definition_file_with_base_dir(self, get_resource_mock): + @patch("samcli.lib.sync.flows.stepfunctions_sync_flow.get_definition_path") + def test_get_definition_file_with_base_dir(self, get_path_mock): sync_flow = self.create_sync_flow() + sync_flow._build_context.use_base_dir = True sync_flow._build_context.base_dir = "base_dir" sync_flow._resource = {"Properties": {"DefinitionUri": "test_uri"}} - result_uri = sync_flow._get_definition_file("test") + get_path_mock.return_value = Path("base_dir").joinpath("test_uri") - self.assertEqual(result_uri, str(Path("base_dir").joinpath("test_uri"))) + result_uri = sync_flow._get_definition_file(sync_flow._state_machine_identifier) + + get_path_mock.assert_called_with( + {"Properties": {"DefinitionUri": "test_uri"}}, + sync_flow._state_machine_identifier, + True, + "base_dir", + sync_flow._stacks, + ) + self.assertEqual(result_uri, Path("base_dir").joinpath("test_uri")) def test_process_definition_file(self): sync_flow = self.create_sync_flow() diff --git a/tests/unit/lib/sync/test_sync_flow.py b/tests/unit/lib/sync/test_sync_flow.py index 1b161bc1cf..2bffa3799c 100644 --- a/tests/unit/lib/sync/test_sync_flow.py +++ b/tests/unit/lib/sync/test_sync_flow.py @@ -1,9 +1,9 @@ -from samcli.lib.providers.provider import ResourceIdentifier +from pathlib import Path +from samcli.lib.providers.provider import ResourceIdentifier, Stack from unittest import TestCase -from unittest.mock import MagicMock, call, patch, Mock +from unittest.mock import MagicMock, patch, Mock -from samcli.lib.sync.sync_flow import SyncFlow, ResourceAPICall, ApiCallTypes -from samcli.lib.utils.lock_distributor import LockChain +from samcli.lib.sync.sync_flow import SyncFlow, ResourceAPICall, ApiCallTypes, get_definition_path from parameterized import parameterized @@ -148,3 +148,21 @@ def test_hash(self): sync_flow._equality_keys = MagicMock() sync_flow._equality_keys.return_value = "A" self.assertEqual(hash(sync_flow), hash((type(sync_flow), "A"))) + + @patch("samcli.lib.sync.sync_flow.Stack.get_stack_by_full_path") + def test_get_definition_path(self, get_stack_mock): + resource = {"Properties": {"DefinitionUri": "test_uri"}} + get_stack_mock.return_value = Stack("parent_path", "stack_name", "location/template.yaml", None, {}) + + definition_path = get_definition_path(resource, "identifier", False, "base_dir", []) + self.assertEqual(definition_path, Path("location").joinpath("test_uri")) + + resource = {"Properties": {"DefinitionUri": ""}} + definition_path = get_definition_path(resource, "identifier", False, "base_dir", []) + self.assertEqual(definition_path, None) + + def test_get_definition_file_with_base_dir(self): + resource = {"Properties": {"DefinitionUri": "test_uri"}} + + definition_path = get_definition_path(resource, "identifier", True, "base_dir", []) + self.assertEqual(definition_path, Path("base_dir").joinpath("test_uri"))