generated from nationalarchives/da-template
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
More config fixup, refactors and tests
- Loading branch information
1 parent
4f83a59
commit 4152a31
Showing
8 changed files
with
240 additions
and
126 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,73 +1,80 @@ | ||
| from unittest.mock import patch | ||
|
|
||
| import boto3 | ||
| from moto import mock_ssm | ||
| from flask import Flask | ||
| from opensearchpy import AWSV4SignerAuth | ||
|
|
||
| from app.main.aws.open_search import ( | ||
| generate_open_search_client_from_aws_params, | ||
| get_open_search_index_from_aws_params, | ||
| generate_open_search_client_from_current_app_config, | ||
| get_open_search_http_auth, | ||
| ) | ||
|
|
||
|
|
||
| @mock_ssm | ||
| def test_get_open_search_index_from_aws_params(): | ||
| ssm_client = boto3.client("ssm") | ||
| ssm_client.put_parameter( | ||
| Name="ENVIRONMENT_NAME", | ||
| Value="test_env", | ||
| Type="String", | ||
| Overwrite=True, | ||
| ) | ||
| ssm_client.put_parameter( | ||
| Name="/test_env/AWS_OPEN_SEARCH_INDEX", | ||
| Value="test_index", | ||
| Type="String", | ||
| Overwrite=True, | ||
| ) | ||
| @patch("app.main.aws.open_search.get_open_search_http_auth") | ||
| @patch("app.main.aws.open_search.OpenSearch") | ||
| def test_generate_open_search_client_from_current_app_config( | ||
| mock_opensearch, mock_get_open_search_http_auth | ||
| ): | ||
| """ | ||
| Given a Flask application and the necessary configuration set for OpenSearch | ||
| When generating an OpenSearch client using the current application's configuration | ||
| Then an OpenSearch client should be created with the expected configuration | ||
| """ | ||
| app = Flask("test") | ||
| app.config["AWS_OPEN_SEARCH_HOST"] = "mock_host" | ||
|
|
||
| assert get_open_search_index_from_aws_params() == "test_index" | ||
| with app.app_context(): | ||
| assert ( | ||
| generate_open_search_client_from_current_app_config() | ||
| == mock_opensearch.return_value | ||
| ) | ||
| mock_opensearch.assert_called_once_with( | ||
| hosts=[{"host": "mock_host", "port": 443}], | ||
| http_auth=mock_get_open_search_http_auth.return_value, | ||
| use_ssl=True, | ||
| verify_certs=True, | ||
| http_compress=True, | ||
| ssl_assert_hostname=False, | ||
| ssl_show_warn=True, | ||
| ) | ||
|
|
||
|
|
||
| @mock_ssm | ||
| @patch("app.main.aws.open_search.OpenSearch") | ||
| def test_generate_open_search_client_from_aws_params(mock_open_search): | ||
| ssm_client = boto3.client("ssm") | ||
| ssm_client.put_parameter( | ||
| Name="ENVIRONMENT_NAME", | ||
| Value="test_env", | ||
| Type="String", | ||
| Overwrite=True, | ||
| ) | ||
| ssm_client.put_parameter( | ||
| Name="/test_env/AWS_OPEN_SEARCH_HOST", | ||
| Value="mock_opensearch_host", | ||
| Type="String", | ||
| Overwrite=True, | ||
| ) | ||
| ssm_client.put_parameter( | ||
| Name="/test_env/AWS_OPEN_SEARCH_USERNAME", | ||
| Value="mock_username", | ||
| Type="String", | ||
| Overwrite=True, | ||
| ) | ||
| ssm_client.put_parameter( | ||
| Name="/test_env/AWS_OPEN_SEARCH_PASSWORD", | ||
| Value="mock_password", | ||
| Type="String", | ||
| Overwrite=True, | ||
| ) | ||
| def test_get_open_search_http_auth(): | ||
| """ | ||
| Given a Flask application with AWS_OPEN_SEARCH_USERNAME, AWS_OPEN_SEARCH_PASSWORD in the config | ||
| When calling get_open_search_http_auth with iam as `False` | ||
| Then a tuple of the username and password is returned | ||
| """ | ||
| app = Flask("test") | ||
| app.config["AWS_OPEN_SEARCH_USERNAME"] = "test_username" | ||
| app.config[ | ||
| "AWS_OPEN_SEARCH_PASSWORD" | ||
| ] = "test_password" # pragma: allowlist secret | ||
|
|
||
| with app.app_context(): | ||
| assert get_open_search_http_auth(iam=False) == ( | ||
| "test_username", | ||
| "test_password", | ||
| ) | ||
|
|
||
|
|
||
| @patch("app.main.aws.open_search.boto3.Session") | ||
| def test_get_open_search_http_auth_iam(mock_boto3_session): | ||
| """ | ||
| Given a mock Boto3 Session and the current app's configuration set for OpenSearch | ||
| When calling get_open_search_http_auth with iam as "True" | ||
| Then it should return an AWSV4SignerAuth object created using the mock AWS | ||
| credentials and AWS region from the current app's configuration. | ||
| """ | ||
| app = Flask("test") | ||
| app.config["AWS_REGION"] = "us-east-1" | ||
|
|
||
| mock_session_instance = mock_boto3_session.return_value | ||
| mock_session_instance.get_credentials.return_value = "mock_credentials" | ||
|
|
||
| assert ( | ||
| generate_open_search_client_from_aws_params() | ||
| == mock_open_search.return_value | ||
| ) | ||
| with app.app_context(): | ||
| auth = get_open_search_http_auth(iam=True) | ||
|
|
||
| mock_open_search.assert_called_once_with( | ||
| hosts=[{"host": "mock_opensearch_host", "port": 443}], | ||
| http_auth=("mock_username", "mock_password"), | ||
| use_ssl=True, | ||
| verify_certs=True, | ||
| http_compress=True, | ||
| ssl_assert_hostname=False, | ||
| ssl_show_warn=True, | ||
| ) | ||
| assert isinstance(auth, AWSV4SignerAuth) | ||
| assert auth.credentials == "mock_credentials" | ||
| assert auth.region == "us-east-1" | ||
| assert auth.service == "es" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.