-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Introduces tests for cron_job_ingest_events.py plus refactoring for e…
…asier tests
- Loading branch information
Mark Kasaboski
committed
Nov 6, 2024
1 parent
7c3f637
commit e042697
Showing
3 changed files
with
269 additions
and
47 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,196 @@ | ||
| from datetime import datetime, timedelta | ||
| import json | ||
| import os | ||
| import pytest | ||
| import sys | ||
|
|
||
| from typing import Any | ||
| from unittest.mock import MagicMock | ||
| from unittest.mock import Mock | ||
| from unittest.mock import PropertyMock | ||
| from unittest.mock import patch | ||
|
|
||
|
|
||
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../../python")) | ||
| from constants import CRON_JOB_THRESHOLD_SINCE_LAST_FETCH | ||
| from constants import KV_COLLECTION_NAME | ||
| from cron_job_ingest_events import fetch_and_print_feed_results | ||
| from cron_job_ingest_events import get_api_credentials | ||
| from cron_job_ingest_events import get_collection_value | ||
| from cron_job_ingest_events import main | ||
| from cron_job_ingest_events import save_collection_value | ||
|
|
||
|
|
||
| def test_get_collection_value_expect_none() -> None: | ||
| app = MagicMock() | ||
| assert get_collection_value(app=app, key="some_key") is None | ||
|
|
||
|
|
||
| def test_get_collection_value_expect_result() -> None: | ||
| app = MagicMock() | ||
| app.service.kvstore.__contains__.side_effect = lambda x: x == KV_COLLECTION_NAME | ||
| app.service.kvstore[KV_COLLECTION_NAME].data.query.return_value = [ | ||
| { | ||
| "_key": "some_key", | ||
| "value": "some_value", | ||
| }, | ||
| ] | ||
|
|
||
| assert get_collection_value(app=app, key="some_key") == "some_value" | ||
|
|
||
|
|
||
| def test_save_collection_value_expect_insert() -> None: | ||
| key = "some_key" | ||
| value = "some_value" | ||
| app = MagicMock() | ||
| save_collection_value(app=app, key=key, value=value) | ||
| app.service.kvstore[KV_COLLECTION_NAME].data.insert.assert_called_once_with( | ||
| json.dumps({"_key": key, "value": value}) | ||
| ) | ||
|
|
||
|
|
||
| def test_save_collection_value_expect_update() -> None: | ||
| key = "some_key" | ||
| value = "update_value" | ||
| app = MagicMock() | ||
| app.service.kvstore.__contains__.side_effect = lambda x: x == KV_COLLECTION_NAME | ||
| app.service.kvstore[KV_COLLECTION_NAME].data.query.return_value = [ | ||
| { | ||
| "_key": key, | ||
| "value": "old_value", | ||
| }, | ||
| ] | ||
| save_collection_value(app=app, key=key, value=value) | ||
| app.service.kvstore[KV_COLLECTION_NAME].data.update.assert_called_once_with( | ||
| id=key, | ||
| data=json.dumps({"value": value}), | ||
| ) | ||
|
|
||
|
|
||
| def test_get_api_credentials_expect_exception() -> None: | ||
| app = MagicMock() | ||
|
|
||
| with pytest.raises(Exception, match="API key not found"): | ||
| get_api_credentials(app=app) | ||
|
|
||
| api_key_item = Mock() | ||
| type(api_key_item.content).username = PropertyMock(return_value="api_key") | ||
| type(api_key_item).clear_password = PropertyMock(return_value="some_api_key") | ||
| app.service.storage_passwords.list.return_value = [api_key_item] | ||
|
|
||
| with pytest.raises(Exception, match="Tenant ID not found"): | ||
| get_api_credentials(app=app) | ||
|
|
||
|
|
||
| def test_get_api_credentials_expect_api_key_and_tenant_id() -> None: | ||
| app = MagicMock() | ||
|
|
||
| api_key_item = Mock() | ||
| type(api_key_item.content).username = PropertyMock(return_value="api_key") | ||
| type(api_key_item).clear_password = PropertyMock(return_value="some_api_key") | ||
|
|
||
| tenant_id_item = Mock() | ||
| type(tenant_id_item.content).username = PropertyMock(return_value="tenant_id") | ||
| type(tenant_id_item).clear_password = PropertyMock(return_value=11111) | ||
|
|
||
| app.service.storage_passwords.list.return_value = [api_key_item, tenant_id_item] | ||
|
|
||
| api_key, tenant_id = get_api_credentials(app=app) | ||
| assert api_key == "some_api_key" | ||
| assert tenant_id == 11111 | ||
|
|
||
|
|
||
| def test_fetch_and_print_feed_results_expect_exception() -> None: | ||
| logger = MagicMock() | ||
| app = MagicMock() | ||
| with patch.object( | ||
| app.service, | ||
| "confs", | ||
| {"flare": {"endpoints": {"me_feed_endpoint": "/firework/v2/me/feed"}}}, | ||
| ): | ||
| fetch_and_print_feed_results( | ||
| logger=logger, app=app, api_key="some_key", tenant_id=11111 | ||
| ) | ||
|
|
||
| logger.error.assert_called_once_with("Exception=Failed to fetch API Token") | ||
|
|
||
|
|
||
| def test_fetch_and_print_feed_results_expect_non_200_response() -> None: | ||
| logger = MagicMock() | ||
| app = MagicMock() | ||
|
|
||
| with patch("cron_job_ingest_events.FlareAPI") as MockFlareAPI: | ||
| response_mock = Mock() | ||
| type(response_mock).status_code = PropertyMock(return_value=400) | ||
| type(response_mock).text = PropertyMock(return_value="Bad Request") | ||
|
|
||
| flare_api_mock_instance = MockFlareAPI.return_value | ||
| flare_api_mock_instance.retrieve_feed.return_value = [response_mock] | ||
|
|
||
| with patch("time.sleep", return_value=None): | ||
| with patch.object( | ||
| app.service, | ||
| "confs", | ||
| {"flare": {"endpoints": {"me_feed_endpoint": "/firework/v2/me/feed"}}}, | ||
| ): | ||
| fetch_and_print_feed_results( | ||
| logger=logger, app=app, api_key="some_key", tenant_id=11111 | ||
| ) | ||
|
|
||
| logger.error.assert_called_once_with("Bad Request") | ||
|
|
||
|
|
||
| def test_fetch_and_print_feed_results_expect_feed_response(capfd: Any) -> None: | ||
| logger = MagicMock() | ||
| app = MagicMock() | ||
|
|
||
| with patch("cron_job_ingest_events.FlareAPI") as MockFlareAPI: | ||
| response_mock = Mock() | ||
| type(response_mock).status_code = PropertyMock(return_value=200) | ||
| response_mock.json.return_value = { | ||
| "next": "some_next_value", | ||
| "items": [ | ||
| { | ||
| "actor": "this guy", | ||
| }, | ||
| { | ||
| "actor": "some other guy", | ||
| }, | ||
| ], | ||
| } | ||
|
|
||
| flare_api_mock_instance = MockFlareAPI.return_value | ||
| flare_api_mock_instance.retrieve_feed.return_value = [response_mock] | ||
|
|
||
| with patch("time.sleep", return_value=None): | ||
| with patch.object( | ||
| app.service, | ||
| "confs", | ||
| {"flare": {"endpoints": {"me_feed_endpoint": "/firework/v2/me/feed"}}}, | ||
| ): | ||
| fetch_and_print_feed_results( | ||
| logger=logger, app=app, api_key="some_key", tenant_id=11111 | ||
| ) | ||
| captured = capfd.readouterr() | ||
| assert ( | ||
| captured.out | ||
| == '{"actor": "this guy"}\n{"actor": "some other guy"}\n' | ||
| ) | ||
|
|
||
| def test_main_expect_early_return() -> None: | ||
| logger = MagicMock() | ||
| app = MagicMock() | ||
|
|
||
| with patch('cron_job_ingest_events.get_last_fetched', return_value=datetime.now() - timedelta(minutes=5)): | ||
| main(logger=logger, app=app) | ||
| logger.debug.assert_called_once_with(f"Fetched events less than {int(CRON_JOB_THRESHOLD_SINCE_LAST_FETCH.seconds / 60)} minutes ago, exiting") | ||
|
|
||
| def test_main_expect_normal_run() -> None: | ||
| logger = MagicMock() | ||
| app = MagicMock() | ||
|
|
||
| with patch('cron_job_ingest_events.get_last_fetched', return_value=datetime.now() - timedelta(minutes=11)): | ||
| with patch('cron_job_ingest_events.get_api_credentials', return_value=("some_api_key", "some_tenant_id")): | ||
| with patch('cron_job_ingest_events.fetch_and_print_feed_results') as fetch_and_print_feed_results_mock: | ||
| main(logger=logger, app=app) | ||
| fetch_and_print_feed_results_mock.assert_called_once_with(logger=logger, app=app, api_key="some_api_key", tenant_id="some_tenant_id") |