Unify resource mapping json schema across cpp and python usage (#5799)
* Unify resource mapping json schema across cpp and python usage Signed-off-by: onecent1101 <liug@amazon.com> * Update based on feedback Signed-off-by: onecent1101 <liug@amazon.com> * Add empty line at end of file Signed-off-by: onecent1101 <liug@amazon.com>
This commit is contained in:
Vincent Liu
GitHub
@@ -154,10 +154,20 @@ if(PAL_TRAIT_BUILD_TESTS_SUPPORTED)
|
||||
AZ::AWSNativeSDKInit
|
||||
Gem::AWSCore.Static
|
||||
)
|
||||
|
||||
ly_add_googletest(
|
||||
NAME Gem::AWSCore.Tests
|
||||
)
|
||||
|
||||
ly_add_target_files(
|
||||
TARGETS
|
||||
AWSCore.Tests
|
||||
FILES
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/Tools/ResourceMappingTool/resource_mapping_schema.json
|
||||
OUTPUT_SUBDIRECTORY
|
||||
Gems/AWSCore
|
||||
)
|
||||
|
||||
if (PAL_TRAIT_BUILD_HOST_TOOLS)
|
||||
ly_add_target(
|
||||
NAME AWSCore.Editor.Tests ${PAL_TRAIT_TEST_TARGET_TYPE}
|
||||
@@ -189,4 +199,13 @@ if(PAL_TRAIT_BUILD_TESTS_SUPPORTED)
|
||||
endif()
|
||||
endif()
|
||||
|
||||
ly_add_target_files(
|
||||
TARGETS
|
||||
AWSCore
|
||||
FILES
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/Tools/ResourceMappingTool/resource_mapping_schema.json
|
||||
OUTPUT_SUBDIRECTORY
|
||||
Gems/AWSCore
|
||||
)
|
||||
|
||||
ly_install_directory(DIRECTORIES Tools/ResourceMappingTool)
|
||||
|
||||
@@ -22,63 +22,6 @@ namespace AWSCore
|
||||
static constexpr const char ResourceMappingTypeKeyName[] = "Type";
|
||||
static constexpr const char ResourceMappingVersionKeyName[] = "Version";
|
||||
|
||||
// TODO: move this into an independent file under AWSCore gem, if resource mapping tool can reuse it
|
||||
static constexpr const char ResourceMappingJsonSchema[] =
|
||||
R"({
|
||||
"$schema": "http://json-schema.org/draft-04/schema",
|
||||
"type": "object",
|
||||
"title": "The AWS Resource Mapping Root schema",
|
||||
"required": ["AWSResourceMappings", "AccountId", "Region", "Version"],
|
||||
"properties": {
|
||||
"AWSResourceMappings": {
|
||||
"type": "object",
|
||||
"title": "The AWSResourceMappings schema",
|
||||
"patternProperties": {
|
||||
"^.+$": {
|
||||
"type": "object",
|
||||
"title": "The AWS Resource Entry schema",
|
||||
"required": ["Type", "Name/ID"],
|
||||
"properties": {
|
||||
"Type": {
|
||||
"$ref": "#/NonEmptyString"
|
||||
},
|
||||
"Name/ID": {
|
||||
"$ref": "#/NonEmptyString"
|
||||
},
|
||||
"AccountId": {
|
||||
"$ref": "#/AccountIdString"
|
||||
},
|
||||
"Region": {
|
||||
"$ref": "#/RegionString"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"additionalProperties": false
|
||||
},
|
||||
"AccountId": {
|
||||
"$ref": "#/AccountIdString"
|
||||
},
|
||||
"Region": {
|
||||
"$ref": "#/RegionString"
|
||||
},
|
||||
"Version": {
|
||||
"pattern": "^[0-9]{1}.[0-9]{1}.[0-9]{1}$"
|
||||
}
|
||||
},
|
||||
"AccountIdString": {
|
||||
"type": "string",
|
||||
"pattern": "^[0-9]{12}$|EMPTY|^$"
|
||||
},
|
||||
"NonEmptyString": {
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
"RegionString": {
|
||||
"type": "string",
|
||||
"pattern": "^[a-z]{2}-[a-z]{4,9}-[0-9]{1}$"
|
||||
},
|
||||
"additionalProperties": false
|
||||
})";
|
||||
|
||||
static constexpr const char ResourceMapppingJsonSchemaFilePath[] =
|
||||
"Gems/AWSCore/resource_mapping_schema.json";
|
||||
} // namespace AWSCore
|
||||
|
||||
@@ -12,6 +12,7 @@
|
||||
#include <AzCore/Serialization/Json/JsonUtils.h>
|
||||
#include <AzCore/Settings/SettingsRegistry.h>
|
||||
#include <AzCore/Settings/SettingsRegistryImpl.h>
|
||||
#include <AzCore/Utils/Utils.h>
|
||||
#include <AzFramework/StringFunc/StringFunc.h>
|
||||
|
||||
#include <AWSCoreInternalBus.h>
|
||||
@@ -244,14 +245,16 @@ namespace AWSCore
|
||||
|
||||
bool AWSResourceMappingManager::ValidateJsonDocumentAgainstSchema(const rapidjson::Document& jsonDocument)
|
||||
{
|
||||
rapidjson::Document jsonSchemaDocument;
|
||||
if (jsonSchemaDocument.Parse(ResourceMappingJsonSchema).HasParseError())
|
||||
AZ::IO::Path executablePath = AZ::IO::PathView(AZ::Utils::GetExecutableDirectory());
|
||||
AZ::IO::Path jsonSchemaPath = (executablePath / ResourceMapppingJsonSchemaFilePath).LexicallyNormal();
|
||||
AZ::Outcome<rapidjson::Document, AZStd::string> readJsonOutcome = AZ::JsonSerializationUtils::ReadJsonFile(jsonSchemaPath.c_str());
|
||||
if (!readJsonOutcome.IsSuccess() || readJsonOutcome.TakeValue().ObjectEmpty())
|
||||
{
|
||||
AZ_Error(AWSResourceMappingManagerName, false, ResourceMappingFileInvalidSchemaErrorMessage);
|
||||
return false;
|
||||
}
|
||||
|
||||
auto jsonSchema = rapidjson::SchemaDocument(jsonSchemaDocument);
|
||||
auto jsonSchema = rapidjson::SchemaDocument(readJsonOutcome.TakeValue());
|
||||
rapidjson::SchemaValidator validator(jsonSchema);
|
||||
|
||||
if (!jsonDocument.Accept(validator))
|
||||
|
||||
@@ -121,7 +121,7 @@ public:
|
||||
|
||||
void SetUp() override
|
||||
{
|
||||
AWSCoreFixture::SetUp();
|
||||
AWSCoreFixture::SetUpFixture(false);
|
||||
|
||||
m_normalizedSourceProjectFolder = AZStd::string::format("%s/%s%s/", AZ::Test::GetCurrentExecutablePath().c_str(),
|
||||
"AWSResourceMappingManager", AZ::Uuid::CreateRandom().ToString<AZStd::string>(false, false).c_str());
|
||||
@@ -142,7 +142,7 @@ public:
|
||||
m_resourceMappingManager->DeactivateManager();
|
||||
m_resourceMappingManager.reset();
|
||||
|
||||
AWSCoreFixture::TearDown();
|
||||
AWSCoreFixture::TearDownFixture(false);
|
||||
}
|
||||
|
||||
// AWSCoreInternalRequestBus interface implementation
|
||||
|
||||
@@ -8,6 +8,7 @@
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <AzCore/Component/ComponentApplication.h>
|
||||
#include <AzCore/Memory/PoolAllocator.h>
|
||||
#include <AzCore/UnitTest/TestTypes.h>
|
||||
#include <AzCore/Settings/SettingsRegistryImpl.h>
|
||||
@@ -111,6 +112,11 @@ public:
|
||||
~AWSCoreFixture() override = default;
|
||||
|
||||
void SetUp() override
|
||||
{
|
||||
SetUpFixture();
|
||||
}
|
||||
|
||||
void SetUpFixture(bool mockSettingsRegistry = true)
|
||||
{
|
||||
AZ::AllocatorInstance<AZ::ThreadPoolAllocator>::Create();
|
||||
AZ::AllocatorInstance<AZ::PoolAllocator>::Create();
|
||||
@@ -120,14 +126,33 @@ public:
|
||||
AZ::IO::FileIOBase::SetInstance(nullptr);
|
||||
AZ::IO::FileIOBase::SetInstance(m_localFileIO);
|
||||
|
||||
m_settingsRegistry = AZStd::make_unique<AZ::SettingsRegistryImpl>();
|
||||
AZ::SettingsRegistry::Register(m_settingsRegistry.get());
|
||||
if (mockSettingsRegistry)
|
||||
{
|
||||
m_settingsRegistry = AZStd::make_unique<AZ::SettingsRegistryImpl>();
|
||||
AZ::SettingsRegistry::Register(m_settingsRegistry.get());
|
||||
}
|
||||
else
|
||||
{
|
||||
m_app = AZStd::make_unique<AZ::ComponentApplication>();
|
||||
}
|
||||
}
|
||||
|
||||
void TearDown() override
|
||||
{
|
||||
AZ::SettingsRegistry::Unregister(m_settingsRegistry.get());
|
||||
m_settingsRegistry.reset();
|
||||
TearDownFixture();
|
||||
}
|
||||
|
||||
void TearDownFixture(bool mockSettingsRegistry = true)
|
||||
{
|
||||
if (mockSettingsRegistry)
|
||||
{
|
||||
AZ::SettingsRegistry::Unregister(m_settingsRegistry.get());
|
||||
m_settingsRegistry.reset();
|
||||
}
|
||||
else
|
||||
{
|
||||
m_app.reset();
|
||||
}
|
||||
|
||||
AZ::IO::FileIOBase::SetInstance(nullptr);
|
||||
|
||||
@@ -173,4 +198,5 @@ private:
|
||||
|
||||
protected:
|
||||
AZStd::unique_ptr<AZ::SettingsRegistryImpl> m_settingsRegistry;
|
||||
AZStd::unique_ptr<AZ::ComponentApplication> m_app;
|
||||
};
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
{
|
||||
"$schema": "http://json-schema.org/draft-04/schema",
|
||||
"type": "object",
|
||||
"title": "O3DE AWS Resource mapping file schema",
|
||||
"required": ["AWSResourceMappings", "AccountId", "Region", "Version"],
|
||||
"properties": {
|
||||
"AWSResourceMappings": {
|
||||
"type": "object",
|
||||
"title": "AWS resource mappings schema",
|
||||
"patternProperties": {
|
||||
"^.+$": {
|
||||
"type": "object",
|
||||
"title": "AWS resource entry schema",
|
||||
"required": ["Type", "Name/ID"],
|
||||
"properties": {
|
||||
"Type": {
|
||||
"$ref": "#/NonEmptyString"
|
||||
},
|
||||
"Name/ID": {
|
||||
"$ref": "#/NonEmptyString"
|
||||
},
|
||||
"AccountId": {
|
||||
"$ref": "#/AccountIdString"
|
||||
},
|
||||
"Region": {
|
||||
"$ref": "#/RegionString"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"additionalProperties": false
|
||||
},
|
||||
"AccountId": {
|
||||
"$ref": "#/AccountIdString"
|
||||
},
|
||||
"Region": {
|
||||
"$ref": "#/RegionString"
|
||||
},
|
||||
"Version": {
|
||||
"pattern": "^[0-9]{1}.[0-9]{1}.[0-9]{1}$"
|
||||
}
|
||||
},
|
||||
"AccountIdString": {
|
||||
"type": "string",
|
||||
"pattern": "^[0-9]{12}$|EMPTY|^$"
|
||||
},
|
||||
"NonEmptyString": {
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
"RegionString": {
|
||||
"type": "string",
|
||||
"pattern": "^[a-z]{2}-[a-z]{4,9}-[0-9]{1}$"
|
||||
},
|
||||
"additionalProperties": false
|
||||
}
|
||||
@@ -10,6 +10,7 @@ import logging
|
||||
import sys
|
||||
|
||||
from utils import environment_utils
|
||||
from utils import json_utils
|
||||
from utils import file_utils
|
||||
|
||||
# arguments setup
|
||||
@@ -74,6 +75,15 @@ if __name__ == "__main__":
|
||||
except FileNotFoundError:
|
||||
logger.warning("Failed to load style sheet for resource mapping tool")
|
||||
|
||||
try:
|
||||
schema_path: str = file_utils.join_path(file_utils.get_parent_directory_path(__file__),
|
||||
'resource_mapping_schema.json')
|
||||
json_utils.load_resource_mapping_json_schema(schema_path)
|
||||
except (FileNotFoundError, ValueError, KeyError) as e:
|
||||
logger.error(f"Failed to load schema file {e}")
|
||||
environment_utils.cleanup_qt_environment()
|
||||
exit(-1)
|
||||
|
||||
logger.info("Initializing configuration manager ...")
|
||||
configuration_manager: ConfigurationManager = ConfigurationManager()
|
||||
configuration_error: bool = not configuration_manager.setup(arguments.profile, arguments.config_path)
|
||||
|
||||
@@ -68,12 +68,65 @@ class TestFileUtils(TestCase):
|
||||
self._mock_path.cwd.assert_called_once()
|
||||
assert actual_path_name == TestFileUtils._expected_path_name
|
||||
|
||||
def test_get_parent_directory_path_return_expected_path_name(self) -> None:
|
||||
self._mock_path.return_value.parent = TestFileUtils._expected_path_name
|
||||
def test_get_parent_directory_path_return_empty_when_invalid_input(self) -> None:
|
||||
mocked_path: MagicMock = self._mock_path.return_value
|
||||
mocked_path.exists.return_value = False
|
||||
|
||||
actual_path_name: str = file_utils.get_parent_directory_path("dummy")
|
||||
self._mock_path.assert_called_once()
|
||||
assert actual_path_name == TestFileUtils._expected_path_name
|
||||
assert actual_path_name == ""
|
||||
|
||||
def test_get_parent_directory_path_return_empty_when_parent_invalid(self) -> None:
|
||||
mocked_path: MagicMock = self._mock_path.return_value
|
||||
mocked_path.exists.return_value = True
|
||||
mocked_parent_path: MagicMock = MagicMock()
|
||||
mocked_path.parent = mocked_parent_path
|
||||
mocked_parent_path.exists.return_value = False
|
||||
|
||||
actual_path_name: str = file_utils.get_parent_directory_path("dummy")
|
||||
self._mock_path.assert_called()
|
||||
assert actual_path_name == ""
|
||||
|
||||
def test_get_parent_directory_path_return_expected_path_when_parent_valid(self) -> None:
|
||||
mocked_path: MagicMock = self._mock_path.return_value
|
||||
mocked_path.exists.return_value = True
|
||||
mocked_parent_path: MagicMock = MagicMock()
|
||||
mocked_path.parent = mocked_parent_path
|
||||
mocked_parent_path.exists.return_value = True
|
||||
mocked_parent_path.resolve.return_value = TestFileUtils._expected_file_name
|
||||
|
||||
actual_path_name: str = file_utils.get_parent_directory_path("dummy")
|
||||
self._mock_path.assert_called()
|
||||
assert actual_path_name == TestFileUtils._expected_file_name
|
||||
|
||||
def test_get_parent_directory_path_return_empty_when_level_two_parent_invalid(self) -> None:
|
||||
mocked_path: MagicMock = self._mock_path.return_value
|
||||
mocked_path.exists.return_value = True
|
||||
mocked_parent_path1: MagicMock = MagicMock()
|
||||
mocked_path.parent = mocked_parent_path1
|
||||
mocked_parent_path1.exists.return_value = True
|
||||
mocked_parent_path2: MagicMock = MagicMock()
|
||||
mocked_parent_path1.parent = mocked_parent_path2
|
||||
mocked_parent_path2.exists.return_value = False
|
||||
|
||||
actual_path_name: str = file_utils.get_parent_directory_path("dummy", 2)
|
||||
self._mock_path.assert_called()
|
||||
assert actual_path_name == ""
|
||||
|
||||
def test_get_parent_directory_path_return_expected_path_when_level_two_parent_valid(self) -> None:
|
||||
mocked_path: MagicMock = self._mock_path.return_value
|
||||
mocked_path.exists.return_value = True
|
||||
mocked_parent_path1: MagicMock = MagicMock()
|
||||
mocked_path.parent = mocked_parent_path1
|
||||
mocked_parent_path1.exists.return_value = True
|
||||
mocked_parent_path2: MagicMock = MagicMock()
|
||||
mocked_parent_path1.parent = mocked_parent_path2
|
||||
mocked_parent_path2.exists.return_value = True
|
||||
mocked_parent_path2.resolve.return_value = TestFileUtils._expected_file_name
|
||||
|
||||
actual_path_name: str = file_utils.get_parent_directory_path("dummy", 2)
|
||||
self._mock_path.assert_called()
|
||||
assert actual_path_name == TestFileUtils._expected_file_name
|
||||
|
||||
def test_find_files_with_suffix_under_directory_return_expected_file_name(self) -> None:
|
||||
mocked_path: MagicMock = self._mock_path.return_value
|
||||
|
||||
@@ -10,6 +10,7 @@ from typing import (Dict, List)
|
||||
from unittest import TestCase
|
||||
from unittest.mock import (MagicMock, mock_open, patch)
|
||||
|
||||
from utils import file_utils
|
||||
from utils import json_utils
|
||||
from model import constants
|
||||
from model.resource_mapping_attributes import (ResourceMappingAttributes, ResourceMappingAttributesBuilder,
|
||||
@@ -49,6 +50,10 @@ class TestJsonUtils(TestCase):
|
||||
}
|
||||
|
||||
def setUp(self) -> None:
|
||||
schema_path: str = file_utils.join_path(file_utils.get_parent_directory_path(__file__, 4),
|
||||
'resource_mapping_schema.json')
|
||||
json_utils.load_resource_mapping_json_schema(schema_path)
|
||||
|
||||
self._mock_open = mock_open()
|
||||
open_patcher: patch = patch("utils.json_utils.open", self._mock_open)
|
||||
self.addCleanup(open_patcher.stop)
|
||||
|
||||
@@ -33,8 +33,29 @@ def get_current_directory_path() -> str:
|
||||
return str(pathlib.Path.cwd())
|
||||
|
||||
|
||||
def get_parent_directory_path(file_path: str) -> str:
|
||||
return pathlib.Path(file_path).parent
|
||||
def get_parent_directory_path(file_path: str, level: int = 1) -> str:
|
||||
"""
|
||||
Get parent directory path based on requested file path
|
||||
:param file_path: The requested file path
|
||||
:param level: The level of parent directory, default value is 1
|
||||
:return The string value of parent directory path if exist; otherwise empty string
|
||||
"""
|
||||
if not pathlib.Path(file_path).exists():
|
||||
return ""
|
||||
|
||||
result: pathlib.Path = pathlib.Path(file_path).parent
|
||||
current_level: int = 1
|
||||
while current_level < level:
|
||||
current_level += 1
|
||||
if result.exists():
|
||||
result = result.parent
|
||||
else:
|
||||
return ""
|
||||
|
||||
if not result.exists():
|
||||
return ""
|
||||
else:
|
||||
return result.resolve()
|
||||
|
||||
|
||||
def find_files_with_suffix_under_directory(dir_path: str, suffix: str) -> List[str]:
|
||||
|
||||
@@ -19,18 +19,29 @@ Json Utils provide related functions to read/write/serialize/deserialize json fo
|
||||
"""
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# resource mapping json content constants
|
||||
_RESOURCE_MAPPING_JSON_KEY_NAME: str = "AWSResourceMappings"
|
||||
_RESOURCE_MAPPING_TYPE_JSON_KEY_NAME: str = "Type"
|
||||
_RESOURCE_MAPPING_NAMEID_JSON_KEY_NAME: str = "Name/ID"
|
||||
_RESOURCE_MAPPING_REGION_JSON_KEY_NAME: str = "Region"
|
||||
_RESOURCE_MAPPING_VERSION_JSON_KEY_NAME: str = "Version"
|
||||
_RESOURCE_MAPPING_JSON_FORMAT_VERSION: str = "1.1.0"
|
||||
|
||||
RESOURCE_MAPPING_ACCOUNTID_JSON_KEY_NAME: str = "AccountId"
|
||||
RESOURCE_MAPPING_ACCOUNTID_TEMPLATE_VALUE: str = "EMPTY"
|
||||
_RESOURCE_MAPPING_ACCOUNTID_PATTERN: str = f"^[0-9]{{12}}$|{RESOURCE_MAPPING_ACCOUNTID_TEMPLATE_VALUE}|^$"
|
||||
_RESOURCE_MAPPING_REGION_PATTERN: str = "^[a-z]{2}-[a-z]{4,9}-[0-9]{1}$"
|
||||
_RESOURCE_MAPPING_VERSION_PATTERN: str = "^[0-9]{1}.[0-9]{1}.[0-9]{1}$"
|
||||
|
||||
# resource mapping json schema constants
|
||||
_RESOURCE_MAPPING_SCHEMA_ACCOUNTID_KEY_NAME: str = "AccountIdString"
|
||||
_RESOURCE_MAPPING_SCHEMA_REGION_KEY_NAME: str = "RegionString"
|
||||
_RESOURCE_MAPPING_SCHEMA_REQUIRED_PROPERTIES_KEY_NAME: str = "required"
|
||||
_RESOURCE_MAPPING_SCHEMA_PROPERTIES_KEY_NAME: str = "properties"
|
||||
_RESOURCE_MAPPING_SCHEMA_PATTERN_PROPERTIES_KEY_NAME: str = "patternProperties"
|
||||
_RESOURCE_MAPPING_SCHEMA_PROPERTY_PATTERN_KEY_NAME: str = "pattern"
|
||||
_RESOURCE_MAPPING_SCHEMA: Dict[str, any] = {}
|
||||
_RESOURCE_MAPPING_SCHEMA_REQUIRED_ROOT_PROPERTIES: List[str] = []
|
||||
_RESOURCE_MAPPING_SCHEMA_REQUIRED_RESOURCE_PROPERTIES: List[str] = []
|
||||
_RESOURCE_MAPPING_SCHEMA_ACCOUNTID_PATTERN: str = ""
|
||||
_RESOURCE_MAPPING_SCHEMA_REGION_PATTERN: str = ""
|
||||
_RESOURCE_MAPPING_SCHEMA_VERSION_PATTERN: str = ""
|
||||
|
||||
def _add_validation_error_message(errors: Dict[int, List[str]], row: int, error_message: str) -> None:
|
||||
if row in errors.keys():
|
||||
@@ -119,6 +130,26 @@ def convert_json_dict_to_resources(json_dict: Dict[str, any]) -> List[ResourceMa
|
||||
return resources
|
||||
|
||||
|
||||
def load_resource_mapping_json_schema(schema_path: str) -> None:
|
||||
global _RESOURCE_MAPPING_SCHEMA, _RESOURCE_MAPPING_SCHEMA_REQUIRED_ROOT_PROPERTIES, _RESOURCE_MAPPING_SCHEMA_REQUIRED_RESOURCE_PROPERTIES,\
|
||||
_RESOURCE_MAPPING_SCHEMA_ACCOUNTID_PATTERN, _RESOURCE_MAPPING_SCHEMA_REGION_PATTERN, _RESOURCE_MAPPING_SCHEMA_VERSION_PATTERN
|
||||
|
||||
if not _RESOURCE_MAPPING_SCHEMA:
|
||||
# assume schema should be in correct format, and manually load expected pattern; tool will log error if schema is invalid
|
||||
_RESOURCE_MAPPING_SCHEMA = read_from_json_file(schema_path)
|
||||
_RESOURCE_MAPPING_SCHEMA_REQUIRED_ROOT_PROPERTIES = _RESOURCE_MAPPING_SCHEMA[_RESOURCE_MAPPING_SCHEMA_REQUIRED_PROPERTIES_KEY_NAME]
|
||||
schema_properties: Dict[str, any] = _RESOURCE_MAPPING_SCHEMA[_RESOURCE_MAPPING_SCHEMA_PROPERTIES_KEY_NAME]
|
||||
schema_pattern_properties: Dict[str, any] = \
|
||||
schema_properties[_RESOURCE_MAPPING_JSON_KEY_NAME][_RESOURCE_MAPPING_SCHEMA_PATTERN_PROPERTIES_KEY_NAME]
|
||||
_RESOURCE_MAPPING_SCHEMA_REQUIRED_RESOURCE_PROPERTIES = list(schema_pattern_properties.values())[0][_RESOURCE_MAPPING_SCHEMA_REQUIRED_PROPERTIES_KEY_NAME]
|
||||
_RESOURCE_MAPPING_SCHEMA_ACCOUNTID_PATTERN = \
|
||||
_RESOURCE_MAPPING_SCHEMA[_RESOURCE_MAPPING_SCHEMA_ACCOUNTID_KEY_NAME][_RESOURCE_MAPPING_SCHEMA_PROPERTY_PATTERN_KEY_NAME]
|
||||
_RESOURCE_MAPPING_SCHEMA_REGION_PATTERN = \
|
||||
_RESOURCE_MAPPING_SCHEMA[_RESOURCE_MAPPING_SCHEMA_REGION_KEY_NAME][_RESOURCE_MAPPING_SCHEMA_PROPERTY_PATTERN_KEY_NAME]
|
||||
_RESOURCE_MAPPING_SCHEMA_VERSION_PATTERN = \
|
||||
schema_properties[_RESOURCE_MAPPING_VERSION_JSON_KEY_NAME][_RESOURCE_MAPPING_SCHEMA_PROPERTY_PATTERN_KEY_NAME]
|
||||
|
||||
|
||||
def read_from_json_file(file_name: str) -> Dict[str, any]:
|
||||
try:
|
||||
json_dict: Dict[str, any] = {}
|
||||
@@ -166,20 +197,20 @@ def validate_resources_according_to_json_schema(resources: List[ResourceMappingA
|
||||
invalid_resources, row_count,
|
||||
error_messages.INVALID_FORMAT_DUPLICATED_KEY_ERROR_MESSAGE.format(resource.key_name))
|
||||
|
||||
if not re.match(_RESOURCE_MAPPING_ACCOUNTID_PATTERN, resource.account_id):
|
||||
if not re.match(_RESOURCE_MAPPING_SCHEMA_ACCOUNTID_PATTERN, resource.account_id):
|
||||
_add_validation_error_message(
|
||||
invalid_resources, row_count,
|
||||
error_messages.INVALID_FORMAT_UNEXPECTED_VALUE_IN_TABLE_ERROR_MESSAGE.format(
|
||||
resource.account_id,
|
||||
RESOURCE_MAPPING_ACCOUNTID_JSON_KEY_NAME,
|
||||
_RESOURCE_MAPPING_ACCOUNTID_PATTERN))
|
||||
_RESOURCE_MAPPING_SCHEMA_ACCOUNTID_PATTERN))
|
||||
|
||||
if not re.match(_RESOURCE_MAPPING_REGION_PATTERN, resource.region):
|
||||
if not re.match(_RESOURCE_MAPPING_SCHEMA_REGION_PATTERN, resource.region):
|
||||
_add_validation_error_message(
|
||||
invalid_resources, row_count,
|
||||
error_messages.INVALID_FORMAT_UNEXPECTED_VALUE_IN_TABLE_ERROR_MESSAGE.format(
|
||||
resource.region, _RESOURCE_MAPPING_REGION_JSON_KEY_NAME,
|
||||
_RESOURCE_MAPPING_REGION_PATTERN))
|
||||
_RESOURCE_MAPPING_SCHEMA_REGION_PATTERN))
|
||||
|
||||
row_count += 1
|
||||
|
||||
@@ -187,30 +218,32 @@ def validate_resources_according_to_json_schema(resources: List[ResourceMappingA
|
||||
|
||||
|
||||
def validate_json_dict_according_to_json_schema(json_dict: Dict[str, any]) -> None:
|
||||
_validate_required_key_in_json_dict(json_dict, "root", _RESOURCE_MAPPING_VERSION_JSON_KEY_NAME)
|
||||
_validate_required_key_in_json_dict(json_dict, "root", _RESOURCE_MAPPING_JSON_KEY_NAME)
|
||||
# The reason we keep this manual json schema validation is python missing supportive feature in default libs
|
||||
# When it is ready, we should be able to replace this with straightforward lib function call
|
||||
root_property: str
|
||||
for root_property in _RESOURCE_MAPPING_SCHEMA_REQUIRED_ROOT_PROPERTIES:
|
||||
_validate_required_key_in_json_dict(json_dict, "root", root_property)
|
||||
|
||||
_validate_required_key_in_json_dict(json_dict, "root", RESOURCE_MAPPING_ACCOUNTID_JSON_KEY_NAME)
|
||||
if not re.match(_RESOURCE_MAPPING_ACCOUNTID_PATTERN, json_dict[RESOURCE_MAPPING_ACCOUNTID_JSON_KEY_NAME]):
|
||||
if not re.match(_RESOURCE_MAPPING_SCHEMA_ACCOUNTID_PATTERN, json_dict[RESOURCE_MAPPING_ACCOUNTID_JSON_KEY_NAME]):
|
||||
raise ValueError(error_messages.INVALID_FORMAT_UNEXPECTED_VALUE_IN_FILE_ERROR_MESSAGE.format(
|
||||
json_dict[RESOURCE_MAPPING_ACCOUNTID_JSON_KEY_NAME],
|
||||
f"root/{RESOURCE_MAPPING_ACCOUNTID_JSON_KEY_NAME}",
|
||||
_RESOURCE_MAPPING_ACCOUNTID_PATTERN))
|
||||
_RESOURCE_MAPPING_SCHEMA_ACCOUNTID_PATTERN))
|
||||
|
||||
_validate_required_key_in_json_dict(json_dict, "root", _RESOURCE_MAPPING_REGION_JSON_KEY_NAME)
|
||||
if not re.match(_RESOURCE_MAPPING_REGION_PATTERN, json_dict[_RESOURCE_MAPPING_REGION_JSON_KEY_NAME]):
|
||||
if not re.match(_RESOURCE_MAPPING_SCHEMA_REGION_PATTERN, json_dict[_RESOURCE_MAPPING_REGION_JSON_KEY_NAME]):
|
||||
raise ValueError(error_messages.INVALID_FORMAT_UNEXPECTED_VALUE_IN_FILE_ERROR_MESSAGE.format(
|
||||
json_dict[_RESOURCE_MAPPING_REGION_JSON_KEY_NAME],
|
||||
f"root/{_RESOURCE_MAPPING_REGION_JSON_KEY_NAME}",
|
||||
_RESOURCE_MAPPING_REGION_PATTERN))
|
||||
_RESOURCE_MAPPING_SCHEMA_REGION_PATTERN))
|
||||
|
||||
json_resources: Dict[str, any] = json_dict[_RESOURCE_MAPPING_JSON_KEY_NAME]
|
||||
if json_resources:
|
||||
resource_key: str
|
||||
resource_value: Dict[str, str]
|
||||
for resource_key, resource_value in json_resources.items():
|
||||
_validate_required_key_in_json_dict(resource_value, resource_key, _RESOURCE_MAPPING_TYPE_JSON_KEY_NAME)
|
||||
_validate_required_key_in_json_dict(resource_value, resource_key, _RESOURCE_MAPPING_NAMEID_JSON_KEY_NAME)
|
||||
resource_property: str
|
||||
for resource_property in _RESOURCE_MAPPING_SCHEMA_REQUIRED_RESOURCE_PROPERTIES:
|
||||
_validate_required_key_in_json_dict(resource_value, resource_key, resource_property)
|
||||
|
||||
|
||||
def write_into_json_file(file_name: str, json_dict: Dict[str, any]) -> None:
|
||||
|
||||
Reference in New Issue
Block a user