From 1539e4400934db21a122b771dbb599a2ef40fe2b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bence=20H=C3=A9zs=C5=91?= Date: Wed, 12 Aug 2026 13:56:51 +0200 Subject: [PATCH 1/2] use resourcetype params for AWS resource enumeration --- core/utils_aws.py | 142 +++++++++++-- tests/test_utils_aws.py | 443 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 564 insertions(+), 21 deletions(-) diff --git a/core/utils_aws.py b/core/utils_aws.py index be68639..cb151a4 100644 --- a/core/utils_aws.py +++ b/core/utils_aws.py @@ -16,32 +16,86 @@ AWS_RETRY_CONFIG = Config(retries={"mode": "adaptive", "max_attempts": 8}) +# Upper bound on the items collected for a single resource type. Some list +# operations (e.g. describe_images, describe_snapshots) return every publicly +# shared resource in the region when called without an owner filter, which would +# otherwise stall the scan and blow up the raw-data file. +MAX_ITEMS_PER_RESOURCE_TYPE = 50_000 + + +def extract_result_path(container: Any, result_path: list) -> list: + if isinstance(result_path, str): + result_path = [result_path] + value = container + for key in result_path: + if not isinstance(value, dict): + return [] + value = value.get(key) + return value if isinstance(value, list) else [] + def paginate( client: Any, operation_name: str, - result_key: str, + result_path: list, + max_items: int = MAX_ITEMS_PER_RESOURCE_TYPE, **kwargs: Any, ) -> list: items: list = [] for page in client.get_paginator(operation_name).paginate(**kwargs): - items.extend(page.get(result_key, [])) + items.extend(extract_result_path(page, result_path)) + if len(items) >= max_items: + return items[:max_items] return items def paginate_or_call( client: Any, operation_name: str, - result_key: str, + result_path: list, + max_items: int = MAX_ITEMS_PER_RESOURCE_TYPE, **kwargs: Any, ) -> list: """paginate() when boto3 supports it for this operation, else a single call.""" if client.can_paginate(operation_name): - return paginate(client, operation_name, result_key, **kwargs) + return paginate(client, operation_name, result_path, max_items, **kwargs) response = getattr(client, operation_name)(**kwargs) - if not isinstance(response, dict): - return [] - return response.get(result_key, []) + return extract_result_path(response, result_path)[:max_items] + + +def parse_resource_type_params(raw_params: Any) -> dict: + if isinstance(raw_params, dict): + return raw_params + try: + parsed = json.loads(raw_params or "{}") + except (TypeError, ValueError): + return {} + return parsed if isinstance(parsed, dict) else {} + + +def resolve_aws_call_spec(resource_type_code: str, params: dict) -> dict | None: + service = params.get("service") + operation = params.get("operation") + if service and operation: + return { + "source": "params", + "service": service, + "operation": operation, + "result_path": list(params.get("result_path") or []), + "kwargs": dict(params.get("kwargs") or {}), + } + + parts = resource_type_code.split(".") + if len(parts) >= 4 and parts[0] == "AWS": + return { + "source": "code", + "service": parts[1], + "operation": parts[2], + "result_path": [part.strip() for part in parts[3:]], + "kwargs": {}, + } + + return None def convert_datetime(obj: Any) -> Any: @@ -78,7 +132,11 @@ def build_aws_resource_inventory( # Load the ResourceType mapping resource_type_mapping = { - item["code"]: {"id": item["id"], "name": item["name"]} + item["code"]: { + "id": item["id"], + "name": item["name"], + "params": item.get("params"), + } for item in load_data("resourcetype") if item["csp"] == 2 and item["status"] == "t" } @@ -89,19 +147,49 @@ def build_aws_resource_inventory( # Aggregate resources by type and location aggregated_resources = defaultdict(int) + # How each catalogue row was resolved, summarised into run.log once the + # sweep is done. Without it a scan result cannot be traced back to the + # master data that produced it. + spec_sources = defaultdict(int) + # Iterate through each resource type in the JSON for idx, (resource_type_code, resource_info) in enumerate( resource_type_mapping.items(), start=1 ): - parts = resource_type_code.split(".") - if len(parts) != 4 or parts[0] != "AWS": - # logger.warning(f"Invalid resource type format: {resource_type_code}. Skipping.") + params = parse_resource_type_params(resource_info.get("params")) + spec = resolve_aws_call_spec(resource_type_code, params) + if spec is None: + if len(resource_type_code.split(".")) == 2: + # Service-level placeholder (e.g. AWS.iam) with nothing to + # call yet -- intentional, so keep it off the console. + spec_sources["placeholder"] += 1 + logger.debug( + "No call spec for placeholder resource type %s. Skipping.", + resource_type_code, + ) + else: + spec_sources["invalid"] += 1 + logger.warning( + "Invalid resource type format: %s. Skipping.", + resource_type_code, + ) continue - # Extract service name, operation name, and result key - service_name, operation_name, result_key = parts[1], parts[2], parts[3] + service_name = spec["service"] + operation_name = spec["operation"] + result_path = spec["result_path"] + call_kwargs = spec["kwargs"] - # logger.info(f"Processing service {service_name} with operation {operation_name}") + spec_sources[spec["source"]] += 1 + logger.debug( + "Resolved %s from %s -> %s.%s result_path=%s kwargs=%s", + resource_type_code, + spec["source"], + service_name, + operation_name, + result_path, + call_kwargs, + ) try: client = session.client( @@ -111,7 +199,20 @@ def build_aws_resource_inventory( # logger.error(f"Operation {operation_name} does not exist for service {service_name}") continue - resources = paginate_or_call(client, operation_name, result_key.strip()) + resources = paginate_or_call( + client, + operation_name, + result_path, + MAX_ITEMS_PER_RESOURCE_TYPE, + **call_kwargs, + ) + if len(resources) >= MAX_ITEMS_PER_RESOURCE_TYPE: + logger.warning( + "Item cap of %d reached for %s.%s; results truncated.", + MAX_ITEMS_PER_RESOURCE_TYPE, + service_name, + operation_name, + ) # Aggregate the resources for resource in resources: @@ -141,6 +242,17 @@ def build_aws_resource_inventory( ) continue + logger.info( + "Resource type resolution: %d of %d rows scanned (%d from params, " + "%d from code), %d placeholders skipped, %d invalid.", + spec_sources["params"] + spec_sources["code"], + len(resource_type_mapping), + spec_sources["params"], + spec_sources["code"], + spec_sources["placeholder"], + spec_sources["invalid"], + ) + # Save raw data to a JSON file raw_data = convert_datetime(raw_data) diff --git a/tests/test_utils_aws.py b/tests/test_utils_aws.py index 8ccf9de..7aa3112 100644 --- a/tests/test_utils_aws.py +++ b/tests/test_utils_aws.py @@ -10,9 +10,12 @@ from core.utils_aws import ( convert_datetime, + extract_result_path, get_missing_months_aws, paginate, paginate_or_call, + parse_resource_type_params, + resolve_aws_call_spec, ) @@ -270,6 +273,369 @@ def test_failed_service_is_skipped_and_logged_at_debug( self.assertFalse(any(r.levelno >= logging.WARNING for r in cm.records)) +class ParseResourceTypeParamsTests(unittest.TestCase): + def test_parses_json_string(self): + self.assertEqual( + parse_resource_type_params('{"service": "s3"}'), {"service": "s3"} + ) + + def test_empty_object_string(self): + self.assertEqual(parse_resource_type_params("{}"), {}) + + def test_none_and_empty_string_degrade_to_empty_dict(self): + self.assertEqual(parse_resource_type_params(None), {}) + self.assertEqual(parse_resource_type_params(""), {}) + + def test_malformed_json_degrades_to_empty_dict(self): + self.assertEqual(parse_resource_type_params("{not json"), {}) + + def test_non_object_json_degrades_to_empty_dict(self): + self.assertEqual(parse_resource_type_params("[1, 2]"), {}) + + def test_dict_passes_through(self): + self.assertEqual( + parse_resource_type_params({"service": "ec2"}), {"service": "ec2"} + ) + + +class ResolveAwsCallSpecTests(unittest.TestCase): + def test_four_part_code_with_no_params(self): + spec = resolve_aws_call_spec("AWS.s3.list_buckets.Buckets", {}) + self.assertEqual( + spec, + { + "source": "code", + "service": "s3", + "operation": "list_buckets", + "result_path": ["Buckets"], + "kwargs": {}, + }, + ) + + def test_five_part_code_with_no_params_keeps_full_result_path(self): + spec = resolve_aws_call_spec( + "AWS.cloudfront.list_distributions.DistributionList.Items", {} + ) + self.assertEqual(spec["result_path"], ["DistributionList", "Items"]) + self.assertEqual(spec["service"], "cloudfront") + + def test_params_take_precedence_over_code(self): + spec = resolve_aws_call_spec( + "AWS.wrong.wrong_op.Wrong", + { + "service": "cloudfront", + "operation": "list_distributions", + "result_path": ["DistributionList", "Items"], + }, + ) + self.assertEqual( + spec, + { + "source": "params", + "service": "cloudfront", + "operation": "list_distributions", + "result_path": ["DistributionList", "Items"], + "kwargs": {}, + }, + ) + + def test_params_with_kwargs(self): + spec = resolve_aws_call_spec( + "AWS.ec2.describe_snapshots.Snapshots", + { + "service": "ec2", + "operation": "describe_snapshots", + "result_path": ["Snapshots"], + "kwargs": {"OwnerIds": ["self"]}, + }, + ) + self.assertEqual(spec["kwargs"], {"OwnerIds": ["self"]}) + + def test_unknown_params_keys_are_ignored(self): + spec = resolve_aws_call_spec( + "AWS.s3.list_buckets.Buckets", + { + "service": "s3", + "operation": "list_buckets", + "result_path": ["Buckets"], + "future_key": "whatever", + }, + ) + self.assertNotIn("future_key", spec) + self.assertEqual(spec["service"], "s3") + + def test_params_missing_result_path_defaults_to_empty(self): + spec = resolve_aws_call_spec( + "AWS.iam", {"service": "iam", "operation": "list_users"} + ) + self.assertEqual(spec["result_path"], []) + + def test_other_cloud_params_fall_back_to_code(self): + spec = resolve_aws_call_spec( + "AWS.s3.list_buckets.Buckets", {"kind": "functionapp"} + ) + self.assertEqual(spec["service"], "s3") + + def test_partial_params_fall_back_to_code(self): + spec = resolve_aws_call_spec("AWS.s3.list_buckets.Buckets", {"service": "s3"}) + self.assertEqual(spec["operation"], "list_buckets") + + def test_source_records_which_branch_resolved_the_row(self): + from_code = resolve_aws_call_spec("AWS.s3.list_buckets.Buckets", {}) + from_params = resolve_aws_call_spec( + "AWS.s3.list_buckets.Buckets", + {"service": "s3", "operation": "list_buckets"}, + ) + self.assertEqual(from_code["source"], "code") + self.assertEqual(from_params["source"], "params") + + def test_two_part_placeholder_is_unresolvable(self): + self.assertIsNone(resolve_aws_call_spec("AWS.iam", {})) + + def test_malformed_code_is_unresolvable(self): + self.assertIsNone(resolve_aws_call_spec("AWS.ec2.describe_instances", {})) + self.assertIsNone(resolve_aws_call_spec("Azure.a.b.c", {})) + self.assertIsNone(resolve_aws_call_spec("", {})) + + +class ExtractResultPathTests(unittest.TestCase): + def test_single_key(self): + self.assertEqual(extract_result_path({"Items": [1, 2]}, ["Items"]), [1, 2]) + + def test_nested_key(self): + self.assertEqual( + extract_result_path({"A": {"B": ["x"]}}, ["A", "B"]), + ["x"], + ) + + def test_missing_intermediate_key_returns_empty(self): + self.assertEqual(extract_result_path({"A": {}}, ["A", "B"]), []) + self.assertEqual(extract_result_path({}, ["A", "B"]), []) + + def test_non_dict_intermediate_returns_empty(self): + self.assertEqual(extract_result_path({"A": "scalar"}, ["A", "B"]), []) + + def test_non_list_final_value_returns_empty(self): + self.assertEqual(extract_result_path({"A": {"B": 5}}, ["A", "B"]), []) + self.assertEqual(extract_result_path({"A": {"B": {}}}, ["A", "B"]), []) + + def test_empty_path_returns_empty(self): + self.assertEqual(extract_result_path({"Items": [1]}, []), []) + + def test_non_dict_container_returns_empty(self): + self.assertEqual(extract_result_path(None, ["Items"]), []) + + +class BuildAwsResourceInventorySpecTests(unittest.TestCase): + """The scanner drives the resolved spec, not the raw code string.""" + + def _row(self, id_, code, name, params="{}"): + return { + "code": code, + "id": id_, + "name": name, + "csp": 2, + "status": "t", + "params": params, + } + + def _run(self, rows, poc_return=None): + """Run the scanner with load_data/boto3/paginate_or_call mocked out. + + Returns (paginate_or_call mock, captured log records). A plain handler + is used instead of assertLogs because a clean scan logs nothing at all. + """ + records: list[logging.LogRecord] = [] + + class _Capture(logging.Handler): + def emit(self, record): + records.append(record) + + aws_logger = logging.getLogger("core.engine.aws") + handler = _Capture() + previous_level = aws_logger.level + aws_logger.addHandler(handler) + aws_logger.setLevel(logging.DEBUG) + + try: + with tempfile.TemporaryDirectory() as tmp: + report_path = os.path.join(tmp, "report") + raw_data_path = os.path.join(tmp, "raw") + os.makedirs(os.path.join(report_path, "data"), exist_ok=True) + os.makedirs(raw_data_path, exist_ok=True) + + with ( + patch("core.utils_aws.load_data", return_value=rows), + patch("core.utils_aws.boto3.Session"), + patch("core.utils_aws.connect") as mock_connect, + patch("core.utils_aws.paginate_or_call") as mock_poc, + ): + mock_connect.return_value.__enter__.return_value = MagicMock() + mock_poc.return_value = [] if poc_return is None else poc_return + + from core.utils_aws import build_aws_resource_inventory + + build_aws_resource_inventory( + 2, + {"accessKey": "AK", "secretKey": "SK", "region": "us-east-1"}, + report_path, + raw_data_path, + ) + return mock_poc, records + finally: + aws_logger.removeHandler(handler) + aws_logger.setLevel(previous_level) + + def test_params_drive_the_call_for_nested_result_path(self): + rows = [ + self._row( + 1, + "AWS.cloudfront.list_distributions.DistributionList.Items", + "CloudFront", + '{"service": "cloudfront", "operation": "list_distributions", ' + '"result_path": ["DistributionList", "Items"]}', + ) + ] + mock_poc, records = self._run(rows, poc_return=[{"Id": "E1"}]) + + args, kwargs = mock_poc.call_args + self.assertEqual(args[1], "list_distributions") + self.assertEqual(args[2], ["DistributionList", "Items"]) + self.assertEqual(kwargs, {}) + self.assertFalse(any(r.levelno >= logging.WARNING for r in records)) + + def test_params_kwargs_are_forwarded_to_the_call(self): + rows = [ + self._row( + 1, + "AWS.ec2.describe_snapshots.Snapshots", + "EBS Snapshot", + '{"service": "ec2", "operation": "describe_snapshots", ' + '"result_path": ["Snapshots"], "kwargs": {"OwnerIds": ["self"]}}', + ) + ] + mock_poc, _ = self._run(rows) + + _, kwargs = mock_poc.call_args + self.assertEqual(kwargs, {"OwnerIds": ["self"]}) + + def test_code_is_used_when_params_are_empty(self): + rows = [self._row(1, "AWS.s3.list_buckets.Buckets", "S3")] + mock_poc, _ = self._run(rows) + + args, _ = mock_poc.call_args + self.assertEqual(args[1], "list_buckets") + self.assertEqual(args[2], ["Buckets"]) + + def test_two_part_placeholder_is_skipped_at_debug(self): + rows = [self._row(1, "AWS.iam", "IAM")] + mock_poc, records = self._run(rows) + + mock_poc.assert_not_called() + self.assertFalse(any(r.levelno >= logging.WARNING for r in records)) + self.assertTrue( + any( + "AWS.iam" in r.getMessage() + for r in records + if r.levelno == logging.DEBUG + ) + ) + + def test_malformed_code_is_skipped_with_warning(self): + rows = [self._row(1, "AWS.ec2.describe_instances", "Broken")] + mock_poc, records = self._run(rows) + + mock_poc.assert_not_called() + self.assertTrue( + any( + r.levelno == logging.WARNING + and "AWS.ec2.describe_instances" in r.getMessage() + for r in records + ) + ) + + def test_item_cap_hit_logs_warning(self): + rows = [self._row(1, "AWS.ec2.describe_images.Images", "AMI")] + with patch("core.utils_aws.MAX_ITEMS_PER_RESOURCE_TYPE", 3): + _, records = self._run(rows, poc_return=[{"ImageId": "ami"}] * 3) + + self.assertTrue( + any( + r.levelno == logging.WARNING + and "describe_images" in r.getMessage() + and "ec2" in r.getMessage() + for r in records + ) + ) + + def test_per_row_debug_line_names_the_spec_source(self): + rows = [ + self._row( + 1, + "AWS.cloudfront.list_distributions.DistributionList.Items", + "CloudFront", + '{"service": "cloudfront", "operation": "list_distributions", ' + '"result_path": ["DistributionList", "Items"]}', + ), + self._row(2, "AWS.s3.list_buckets.Buckets", "S3"), + ] + _, records = self._run(rows) + messages = [r.getMessage() for r in records if r.levelno == logging.DEBUG] + + self.assertTrue( + any("AWS.cloudfront" in m and "from params" in m for m in messages), + messages, + ) + self.assertTrue( + any("AWS.s3.list_buckets" in m and "from code" in m for m in messages), + messages, + ) + + def test_summary_line_counts_every_resolution_outcome(self): + rows = [ + self._row( + 1, + "AWS.cloudfront.list_distributions.DistributionList.Items", + "CloudFront", + '{"service": "cloudfront", "operation": "list_distributions", ' + '"result_path": ["DistributionList", "Items"]}', + ), + self._row(2, "AWS.s3.list_buckets.Buckets", "S3"), + self._row(3, "AWS.iam", "IAM"), + self._row(4, "AWS.ec2.describe_instances", "Broken"), + ] + _, records = self._run(rows) + summary = [ + r.getMessage() + for r in records + if r.levelno == logging.INFO + and "Resource type resolution" in r.getMessage() + ] + + self.assertEqual(len(summary), 1, records) + self.assertEqual( + summary[0], + "Resource type resolution: 2 of 4 rows scanned (1 from params, " + "1 from code), 1 placeholders skipped, 1 invalid.", + ) + + def test_summary_is_info_so_it_stays_off_the_default_console(self): + rows = [self._row(1, "AWS.s3.list_buckets.Buckets", "S3")] + _, records = self._run(rows) + + summary = [r for r in records if "Resource type resolution" in r.getMessage()] + self.assertEqual([r.levelno for r in summary], [logging.INFO]) + + def test_item_cap_is_passed_to_paginate_or_call(self): + from core.utils_aws import MAX_ITEMS_PER_RESOURCE_TYPE + + rows = [self._row(1, "AWS.s3.list_buckets.Buckets", "S3")] + mock_poc, _ = self._run(rows) + + args, _ = mock_poc.call_args + self.assertEqual(args[3], MAX_ITEMS_PER_RESOURCE_TYPE) + + class PaginateTests(unittest.TestCase): def _fake_client(self, pages): """Build a stub client whose paginator yields the given pages.""" @@ -283,15 +649,49 @@ def test_collects_items_across_pages(self): client = self._fake_client( [{"Items": [1, 2, 3]}, {"Items": [4, 5]}, {"Items": [6]}] ) - self.assertEqual(paginate(client, "any_op", "Items"), [1, 2, 3, 4, 5, 6]) + self.assertEqual(paginate(client, "any_op", ["Items"]), [1, 2, 3, 4, 5, 6]) def test_missing_result_key_treated_as_empty(self): client = self._fake_client([{"Items": [1]}, {}]) - self.assertEqual(paginate(client, "any_op", "Items"), [1]) + self.assertEqual(paginate(client, "any_op", ["Items"]), [1]) + + def test_walks_nested_result_path_per_page(self): + client = self._fake_client( + [ + {"DistributionList": {"Items": ["d1", "d2"]}}, + {"DistributionList": {"Items": ["d3"]}}, + ] + ) + self.assertEqual( + paginate(client, "list_distributions", ["DistributionList", "Items"]), + ["d1", "d2", "d3"], + ) + + def test_page_missing_intermediate_key_yields_nothing_for_that_page(self): + client = self._fake_client( + [{"DistributionList": {"Items": ["d1"]}}, {}, {"DistributionList": {}}] + ) + self.assertEqual( + paginate(client, "list_distributions", ["DistributionList", "Items"]), + ["d1"], + ) + + def test_bare_string_result_path_is_treated_as_single_key(self): + client = self._fake_client([{"Volumes": ["v1", "v2"]}]) + self.assertEqual(paginate(client, "describe_volumes", "Volumes"), ["v1", "v2"]) + + def test_stops_and_truncates_at_max_items(self): + pages = [{"Items": list(range(4))} for _ in range(10)] + client = self._fake_client(pages) + + result = paginate(client, "any_op", ["Items"], 10) + + self.assertEqual(len(result), 10) + self.assertEqual(result, list(range(4)) * 2 + [0, 1]) def test_forwards_kwargs_to_paginator(self): client = self._fake_client([{"Items": []}]) - paginate(client, "any_op", "Items", MaxResults=50) + paginate(client, "any_op", ["Items"], MaxResults=50) client.get_paginator.return_value.paginate.assert_called_once_with( MaxResults=50 ) @@ -322,7 +722,7 @@ def test_uses_paginator_when_available(self): client.can_paginate.return_value = True client.get_paginator.return_value = paginator - self.assertEqual(paginate_or_call(client, "list_things", "Items"), [1, 2, 3]) + self.assertEqual(paginate_or_call(client, "list_things", ["Items"]), [1, 2, 3]) client.can_paginate.assert_called_once_with("list_things") def test_falls_back_to_single_call_when_not_paginable(self): @@ -330,7 +730,7 @@ def test_falls_back_to_single_call_when_not_paginable(self): client.can_paginate.return_value = False client.list_things.return_value = {"Items": ["a", "b"]} - self.assertEqual(paginate_or_call(client, "list_things", "Items"), ["a", "b"]) + self.assertEqual(paginate_or_call(client, "list_things", ["Items"]), ["a", "b"]) client.list_things.assert_called_once_with() def test_non_dict_response_returns_empty_list(self): @@ -338,7 +738,38 @@ def test_non_dict_response_returns_empty_list(self): client.can_paginate.return_value = False client.list_things.return_value = None - self.assertEqual(paginate_or_call(client, "list_things", "Items"), []) + self.assertEqual(paginate_or_call(client, "list_things", ["Items"]), []) + + def test_walks_nested_result_path_on_single_call(self): + client = MagicMock() + client.can_paginate.return_value = False + client.get_apps.return_value = { + "ApplicationsResponse": {"Applications": ["a1", "a2"]} + } + + self.assertEqual( + paginate_or_call( + client, "get_apps", ["ApplicationsResponse", "Applications"] + ), + ["a1", "a2"], + ) + + def test_forwards_kwargs_to_single_call(self): + client = MagicMock() + client.can_paginate.return_value = False + client.describe_images.return_value = {"Images": []} + + paginate_or_call(client, "describe_images", ["Images"], Owners=["self"]) + client.describe_images.assert_called_once_with(Owners=["self"]) + + def test_truncates_single_call_at_max_items(self): + client = MagicMock() + client.can_paginate.return_value = False + client.list_things.return_value = {"Items": list(range(20))} + + self.assertEqual( + paginate_or_call(client, "list_things", ["Items"], 5), [0, 1, 2, 3, 4] + ) if __name__ == "__main__": From 68562579e13c38f52dd238d270ac418755676126 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bence=20H=C3=A9zs=C5=91?= Date: Fri, 14 Aug 2026 18:22:29 +0200 Subject: [PATCH 2/2] load the egress resource registry from master data --- assets/template/egress.html | 3 + core/utils_db.py | 3 + core/utils_egress.py | 182 ++++++++++- core/utils_egress_aws.py | 248 ++++++++------- core/utils_egress_azure.py | 70 +---- core/utils_report_egress.py | 20 +- main.py | 1 + tests/test_utils_and_main.py | 8 +- tests/test_utils_egress.py | 174 ++++++++++- tests/test_utils_egress_aws.py | 461 ++++++++++++++++++++++++++-- tests/test_utils_egress_azure.py | 112 ++++++- tests/test_utils_egress_registry.py | 281 +++++++++++++++++ tests/test_utils_report_egress.py | 115 ++++++- 13 files changed, 1465 insertions(+), 213 deletions(-) create mode 100644 tests/test_utils_egress_registry.py diff --git a/assets/template/egress.html b/assets/template/egress.html index 483b6c8..5353767 100644 --- a/assets/template/egress.html +++ b/assets/template/egress.html @@ -36,6 +36,9 @@ .category-block { background-color: rgba(83, 155, 255, 1); } + .category-file { + background-color: rgba(245, 158, 11, 1); + } .category-database { background-color: rgba(168, 85, 247, 1); } diff --git a/core/utils_db.py b/core/utils_db.py index 1d18e49..2ef045d 100644 --- a/core/utils_db.py +++ b/core/utils_db.py @@ -10,7 +10,10 @@ ALLOWED_TABLES = { "resourcetype", + "resourcetype_data", "resource_inventory", + "egress_inventory", + "egress_inventory_tier", "cost_inventory", "risk_inventory", "scoring_data", diff --git a/core/utils_egress.py b/core/utils_egress.py index 83036d7..afb1ae2 100644 --- a/core/utils_egress.py +++ b/core/utils_egress.py @@ -5,11 +5,69 @@ from typing import Any from datetime import datetime, timezone +from .utils_db import connect, load_data + logger = logging.getLogger("core.engine.egress") GIB = 1024**3 +def _parse_params(raw_params: Any) -> dict[str, Any]: + if isinstance(raw_params, dict): + return raw_params + try: + parsed = json.loads(raw_params or "{}") + except (TypeError, ValueError): + return {} + return parsed if isinstance(parsed, dict) else {} + + +def load_egress_registry(csp: int) -> dict[str, dict[str, Any]]: + # Master data is the only source of truth: a missing table or an empty + # result fails the egress stage rather than producing a wrong estimate. + data_rows = load_data("resourcetype_data") + resource_types = load_data("resourcetype") + + types_by_id = {rt["id"]: rt for rt in resource_types} + + registry: dict[str, dict[str, Any]] = {} + for row in data_rows: + if row["status"] != "t": + continue + resource_type = types_by_id.get(row["resource_type"]) + if resource_type is None or resource_type["csp"] != csp: + continue + + # resourcetype.status is deliberately not filtered. Rows such as EBS + # Snapshots and Managed Disks ship as 'f' so they stay out of the + # inventory catalogue, but they still carry data that has to be egressed + # -- excluding them would silently drop a whole category from the + # estimate. + code = resource_type["code"] + # Azure matches against ARM resource.type, which is case-insensitive; + # AWS codes are handed to the collectors as the row identifier verbatim. + key = code.strip().lower() if csp == 1 else code + + params = _parse_params(row["params"]) + registry[key] = { + **params, + # Carried so the egress_inventory writer has the FK without a + # reverse lookup -- the Azure key is lowercased, the code is not. + "resource_type_id": resource_type["id"], + "category": row["data_category"], + "label": resource_type["name"], + "strategy": row["strategy"], + } + + if not registry: + raise ValueError( + f"No enabled resourcetype_data rows for CSP {csp}; " + "cannot build the egress registry." + ) + + return registry + + def new_row( resource_id: str, name: str, resource_type: str, label: str, category: str ) -> dict[str, Any]: @@ -59,11 +117,117 @@ def compute_totals( } +def resource_type_id_for( + resource_type_ids: dict[str, int], row_type: str +) -> int | None: + # AWS rows carry the code verbatim; Azure rows carry the ARM resource.type, + # which the registry keys in lowercase. + return resource_type_ids.get(row_type) or resource_type_ids.get( + row_type.strip().lower() + ) + + +def public_row(row: dict[str, Any]) -> dict[str, Any]: + # The label is dropped: it is resourcetype.name, already in the database. + return { + "id": row["id"], + "name": row["name"], + "code": row["type"], + "category": row["category"], + "size_bytes": row["size_bytes"], + "size_unknown": row["size_unknown"], + "tier_bytes": row["tier_bytes"], + "flags": row["flags"], + "notes": row["notes"], + } + + +def write_egress_inventory( + rows: list[dict[str, Any]], + resource_type_ids: dict[str, int], + archive_tiers: set[str], + db_path: str, +) -> None: + conn = connect(db_path) + try: + cursor = conn.cursor() + for row in rows: + resource_type_id = resource_type_id_for(resource_type_ids, row["type"]) + if resource_type_id is None: + # Should not happen: every row is built from a registry entry. + logger.warning( + "No resourcetype id for %s; row not stored.", row["type"] + ) + continue + cursor.execute( + "INSERT INTO egress_inventory " + "(resource_type, name, size_bytes, size_unknown, flags, notes) " + "VALUES (?, ?, ?, ?, ?, ?)", + ( + resource_type_id, + row["name"], + row["size_bytes"], + int(bool(row["size_unknown"])), + json.dumps(row["flags"]), + json.dumps(row["notes"]), + ), + ) + inventory_id = cursor.lastrowid + for tier, size_bytes in (row["tier_bytes"] or {}).items(): + cursor.execute( + "INSERT INTO egress_inventory_tier " + "(egress_inventory_id, tier, size_bytes, is_archive) " + "VALUES (?, ?, ?, ?)", + (inventory_id, tier, size_bytes, int(tier in archive_tiers)), + ) + conn.commit() + finally: + conn.close() + + +def read_egress_inventory(db_path: str) -> tuple[list[dict[str, Any]], set[str]]: + # Rebuilds the row shape the report already consumes, so everything + # downstream of _load_estimate is unchanged. + resource_types = {rt["id"]: rt for rt in load_data("resourcetype", db_path=db_path)} + categories = { + rtd["resource_type"]: rtd["data_category"] + for rtd in load_data("resourcetype_data", db_path=db_path) + } + + tiers_by_row: dict[int, dict[str, int]] = {} + archive_tiers: set[str] = set() + for tier_row in load_data("egress_inventory_tier", db_path=db_path): + tiers_by_row.setdefault(tier_row["egress_inventory_id"], {})[ + tier_row["tier"] + ] = tier_row["size_bytes"] + if tier_row["is_archive"]: + archive_tiers.add(tier_row["tier"]) + + rows = [] + for record in load_data("egress_inventory", db_path=db_path): + resource_type = resource_types[record["resource_type"]] + row = new_row( + record["id"], + record["name"], + resource_type["code"], + resource_type["name"], + categories.get(record["resource_type"], ""), + ) + row["size_bytes"] = record["size_bytes"] + row["size_unknown"] = bool(record["size_unknown"]) + row["tier_bytes"] = tiers_by_row.get(record["id"]) or None + row["flags"] = json.loads(record["flags"] or "[]") + row["notes"] = json.loads(record["notes"] or "[]") + rows.append(row) + return rows, archive_tiers + + def estimate_egress( cloud_service_provider: int, provider_details: dict[str, Any], raw_data_path: str, *, + report_path: str, name: str, exit_strategy: int, assessment_type: int, @@ -82,6 +246,11 @@ def estimate_egress( f"Unsupported cloud service provider: {cloud_service_provider}" ) + resource_type_ids = { + key: entry["resource_type_id"] + for key, entry in load_egress_registry(cloud_service_provider).items() + } + json_payload = { "meta": { "name": name, @@ -93,11 +262,20 @@ def estimate_egress( ), }, "data": { - "resources": rows, + "resources": [public_row(row) for row in rows], "totals": compute_totals(rows, archive_tiers), }, } - json_path = os.path.join(raw_data_path, "egress_estimate.json") + # The estimate is stored per assessment; the JSON stays as the raw + # artifact alongside the other raw dumps. + write_egress_inventory( + rows, + resource_type_ids, + archive_tiers, + os.path.join(report_path, "data", "assessment.db"), + ) + + json_path = os.path.join(raw_data_path, "egress_inventory_raw_data.json") with open(json_path, "w", encoding="utf-8") as json_file: json.dump(json_payload, json_file, indent=4) diff --git a/core/utils_egress_aws.py b/core/utils_egress_aws.py index ef25ef9..0f253fa 100644 --- a/core/utils_egress_aws.py +++ b/core/utils_egress_aws.py @@ -5,8 +5,12 @@ from datetime import datetime, timedelta, timezone from botocore.exceptions import BotoCoreError, ClientError -from .utils_aws import AWS_RETRY_CONFIG, paginate -from .utils_egress import GIB, format_bytes, new_row +from .utils_aws import ( + AWS_RETRY_CONFIG, + extract_result_path, + paginate_or_call, +) +from .utils_egress import GIB, format_bytes, load_egress_registry, new_row logger = logging.getLogger("core.engine.egress.aws") @@ -41,39 +45,6 @@ "DeepArchiveStagingStorage": "Deep Archive", } -EGRESS_RESOURCE_REGISTRY = { - "AWS.s3.list_buckets.Buckets": { - "category": "object", - "label": "S3 Bucket", - "strategy": "s3_bucket_metrics", - }, - "AWS.ec2.describe_volumes.Volumes": { - "category": "block", - "label": "EBS Volume", - "strategy": "ebs_volumes", - }, - "AWS.ec2.describe_snapshots.Snapshots": { - "category": "block", - "label": "EBS Snapshot", - "strategy": "ebs_snapshots", - }, - "AWS.rds.describe_db_instances.DBInstances": { - "category": "database", - "label": "RDS Instance", - "strategy": "rds_instances", - }, - "AWS.dynamodb.list_tables.TableNames": { - "category": "database", - "label": "DynamoDB Table", - "strategy": "dynamodb_tables", - }, - "AWS.backup.list_backup_vaults.BackupVaultList": { - "category": "backup", - "label": "Backup Vault", - "strategy": "backup_vaults", - }, -} - def fetch_latest_metric_values( cloudwatch: Any, @@ -129,6 +100,67 @@ def fetch_latest_metric_values( return None +SIZE_UNITS = { + "B": 1, + "KiB": 1024, + "MiB": 1024**2, + "GiB": GIB, + "TiB": 1024**4, +} + + +def _nested_get(item: dict[str, Any], path: Any) -> Any: + if isinstance(path, str): + path = [path] + value: Any = item + for key in path: + if not isinstance(value, dict): + return None + value = value.get(key) + return value + + +def _resource_identity(item: dict[str, Any], sizing: dict[str, Any]) -> tuple[str, str]: + # The row id prefers an ARN when the service reports one; the display name + # comes from a tag when the service tags resources, else from a field. + identifier = item[sizing["id_field"]] + name_tag = sizing.get("name_tag") + if name_tag: + name = next( + ( + tag["Value"] + for tag in item.get(sizing.get("tags_field", "Tags"), []) + if tag["Key"] == name_tag + ), + identifier, + ) + else: + name = item.get(sizing.get("name_field") or sizing["id_field"], identifier) + arn_field = sizing.get("arn_field") + row_id = item.get(arn_field, identifier) if arn_field else identifier + return row_id, name + + +def _enumeration_client(session: Any, region: str, entry: dict[str, Any]) -> Any: + return session.client( + entry["enumeration"]["service"], region_name=region, config=AWS_RETRY_CONFIG + ) + + +def _enumerate(session: Any, region: str, entry: dict[str, Any]) -> tuple[Any, list]: + # service/operation/result_path/kwargs all come from resourcetype_data, so a + # catalogue change does not need a code change. Sizing stays in the strategy. + enumeration = entry["enumeration"] + client = _enumeration_client(session, region, entry) + items = paginate_or_call( + client, + enumeration["operation"], + enumeration["result_path"], + **enumeration.get("kwargs", {}), + ) + return client, items + + def _bucket_region(s3_client: Any, bucket_name: str) -> str: location = s3_client.get_bucket_location(Bucket=bucket_name).get( "LocationConstraint" @@ -140,15 +172,21 @@ def _bucket_region(s3_client: Any, bucket_name: str) -> str: return location -def _list_buckets_in_region(s3_client: Any, region: str) -> list[str]: +def _list_buckets_in_region( + s3_client: Any, region: str, enumeration: dict[str, Any] +) -> list[str]: + # The region filter is strategy logic, but the call itself is master data. + list_buckets = getattr(s3_client, enumeration["operation"]) + result_path = enumeration["result_path"] + kwargs = enumeration.get("kwargs", {}) try: - response = s3_client.list_buckets(BucketRegion=region) - return [bucket["Name"] for bucket in response.get("Buckets", [])] + response = list_buckets(BucketRegion=region, **kwargs) + return [bucket["Name"] for bucket in extract_result_path(response, result_path)] except (BotoCoreError, ClientError) as e: logger.debug("ListBuckets with BucketRegion filter failed: %s", str(e)) bucket_names = [] - for bucket in s3_client.list_buckets().get("Buckets", []): + for bucket in extract_result_path(list_buckets(**kwargs), result_path): name = bucket["Name"] bucket_region = bucket.get("BucketRegion") if bucket_region is None: @@ -165,12 +203,12 @@ def _list_buckets_in_region(s3_client: Any, region: str) -> list[str]: def _collect_s3_buckets( session: Any, region: str, code: str, entry: dict[str, Any] ) -> list[dict[str, Any]]: - s3_client = session.client("s3", region_name=region, config=AWS_RETRY_CONFIG) + s3_client = _enumeration_client(session, region, entry) cloudwatch = session.client( "cloudwatch", region_name=region, config=AWS_RETRY_CONFIG ) - bucket_names = _list_buckets_in_region(s3_client, region) + bucket_names = _list_buckets_in_region(s3_client, region, entry["enumeration"]) rows = [] for name in bucket_names: @@ -244,49 +282,41 @@ def _collect_s3_buckets( return rows -def _collect_ebs_volumes( +def _collect_list_item_size( session: Any, region: str, code: str, entry: dict[str, Any] ) -> list[dict[str, Any]]: - ec2_client = session.client("ec2", region_name=region, config=AWS_RETRY_CONFIG) + sizing = entry["sizing"] + multiplier = SIZE_UNITS[sizing.get("size_unit", "GiB")] + _, items = _enumerate(session, region, entry) rows = [] - for volume in paginate(ec2_client, "describe_volumes", "Volumes"): - name = next( - (tag["Value"] for tag in volume.get("Tags", []) if tag["Key"] == "Name"), - volume["VolumeId"], - ) - row = new_row(volume["VolumeId"], name, code, entry["label"], entry["category"]) - size_gb = volume.get("Size") - if size_gb: - row["size_bytes"] = int(size_gb) * GIB - row["flags"].append("allocated (upper bound)") + for item in items: + row_id, name = _resource_identity(item, sizing) + row = new_row(row_id, name, code, entry["label"], entry["category"]) + size = _nested_get(item, sizing["size_field"]) + if size: + row["size_bytes"] = int(size) * multiplier + row["flags"].extend(sizing.get("flags", [])) + row["notes"].extend(sizing.get("notes", [])) else: row["size_unknown"] = True rows.append(row) return rows -def _collect_ebs_snapshots( +def _collect_not_sizeable( session: Any, region: str, code: str, entry: dict[str, Any] ) -> list[dict[str, Any]]: - ec2_client = session.client("ec2", region_name=region, config=AWS_RETRY_CONFIG) + sizing = entry["sizing"] + _, items = _enumerate(session, region, entry) rows = [] - for snapshot in paginate( - ec2_client, "describe_snapshots", "Snapshots", OwnerIds=["self"] - ): - row = new_row( - snapshot["SnapshotId"], - snapshot["SnapshotId"], - code, - entry["label"], - entry["category"], - ) - size_gb = snapshot.get("VolumeSize") - if size_gb: - row["size_bytes"] = int(size_gb) * GIB - row["flags"].append("allocated (upper bound)") - row["notes"].append("incremental – shares blocks with sibling snapshots") - else: - row["size_unknown"] = True + for item in items: + row_id, name = _resource_identity(item, sizing) + row = new_row(row_id, name, code, entry["label"], entry["category"]) + row["flags"].extend(sizing.get("flags", [])) + count_field = sizing.get("count_field") + count = _nested_get(item, count_field) if count_field else None + if count: + row["notes"].append(sizing["count_note"].format(count=count)) rows.append(row) return rows @@ -294,7 +324,7 @@ def _collect_ebs_snapshots( def _collect_rds_instances( session: Any, region: str, code: str, entry: dict[str, Any] ) -> list[dict[str, Any]]: - rds_client = session.client("rds", region_name=region, config=AWS_RETRY_CONFIG) + _, instances = _enumerate(session, region, entry) cloudwatch = session.client( "cloudwatch", region_name=region, config=AWS_RETRY_CONFIG ) @@ -302,9 +332,7 @@ def _collect_rds_instances( rows = [] specs = [] spec_rows: dict[str, tuple[dict[str, Any], int]] = {} - for index, instance in enumerate( - paginate(rds_client, "describe_db_instances", "DBInstances") - ): + for index, instance in enumerate(instances): identifier = instance["DBInstanceIdentifier"] row = new_row( instance.get("DBInstanceArn", identifier), @@ -348,11 +376,9 @@ def _collect_rds_instances( def _collect_dynamodb_tables( session: Any, region: str, code: str, entry: dict[str, Any] ) -> list[dict[str, Any]]: - dynamodb_client = session.client( - "dynamodb", region_name=region, config=AWS_RETRY_CONFIG - ) + dynamodb_client, table_names = _enumerate(session, region, entry) rows = [] - for table_name in paginate(dynamodb_client, "list_tables", "TableNames"): + for table_name in table_names: try: table = dynamodb_client.describe_table(TableName=table_name).get( "Table", {} @@ -383,39 +409,19 @@ def _collect_dynamodb_tables( return rows -def _collect_backup_vaults( - session: Any, region: str, code: str, entry: dict[str, Any] -) -> list[dict[str, Any]]: - backup_client = session.client( - "backup", region_name=region, config=AWS_RETRY_CONFIG - ) - rows = [] - for vault in paginate(backup_client, "list_backup_vaults", "BackupVaultList"): - vault_name = vault["BackupVaultName"] - row = new_row( - vault.get("BackupVaultArn", vault_name), - vault_name, - code, - entry["label"], - entry["category"], - ) - row["flags"].append("backup vault – not sized") - recovery_points = vault.get("NumberOfRecoveryPoints") - if recovery_points: - row["notes"].append( - f"{recovery_points} recovery points (cannot be exported directly)" - ) - rows.append(row) - return rows - - +# Strategy names stay stable so master data and engine can be deployed +# independently; several of them share one parameterised collector. The +# service-neutral names are the ones to use for new master-data rows. _STRATEGY_COLLECTORS = { "s3_bucket_metrics": _collect_s3_buckets, - "ebs_volumes": _collect_ebs_volumes, - "ebs_snapshots": _collect_ebs_snapshots, "rds_instances": _collect_rds_instances, "dynamodb_tables": _collect_dynamodb_tables, - "backup_vaults": _collect_backup_vaults, + "list_item_size": _collect_list_item_size, + "not_sizeable": _collect_not_sizeable, + # Kept so existing rows keep working; prefer the two names above. + "ebs_volumes": _collect_list_item_size, + "ebs_snapshots": _collect_list_item_size, + "backup_vaults": _collect_not_sizeable, } @@ -431,13 +437,29 @@ def collect_aws_egress( ) rows = [] - for code, entry in EGRESS_RESOURCE_REGISTRY.items(): - collector = _STRATEGY_COLLECTORS[entry["strategy"]] + for code, entry in load_egress_registry(2).items(): + collector = _STRATEGY_COLLECTORS.get(entry["strategy"]) + if collector is None: + # Master data can ship a strategy ahead of the engine; skip that one + # resource type rather than aborting the whole egress run. + logger.warning( + "Unknown egress strategy %r for %s; skipping.", + entry["strategy"], + code, + ) + continue try: rows.extend(collector(session, region, code, entry)) except Exception as e: - logger.debug( - "Egress collection failed for %s: %s", code, str(e), exc_info=True + # One failing service must not abort the run, but a whole resource + # type dropping out of the estimate has to be visible in run.log. + logger.warning( + "Egress collection failed for %s (%s: %s); " + "this resource type is missing from the estimate.", + code, + type(e).__name__, + str(e), + exc_info=True, ) return rows, ARCHIVE_TIERS diff --git a/core/utils_egress_azure.py b/core/utils_egress_azure.py index 6cada1e..fb360bb 100644 --- a/core/utils_egress_azure.py +++ b/core/utils_egress_azure.py @@ -6,7 +6,7 @@ from azure.identity import ClientSecretCredential from azure.mgmt.resource import ResourceManagementClient -from .utils_egress import GIB, format_bytes, new_row +from .utils_egress import GIB, format_bytes, load_egress_registry, new_row logger = logging.getLogger("core.engine.egress.azure") @@ -22,64 +22,14 @@ ARCHIVE_TIERS = {"Archive"} -EGRESS_RESOURCE_REGISTRY = { - "microsoft.storage/storageaccounts": { - "category": "object", - "label": "Storage Account", - "strategy": "storage_account_metrics", - }, - "microsoft.compute/disks": { - "category": "block", - "label": "Managed Disk", - "strategy": "allocated_size_property", - "api_version": "2024-03-02", - "size_property": "diskSizeGB", - }, - "microsoft.compute/snapshots": { - "category": "block", - "label": "Snapshot", - "strategy": "allocated_size_property", - "api_version": "2024-03-02", - "size_property": "diskSizeGB", - }, - "microsoft.sql/servers/databases": { - "category": "database", - "label": "SQL Database", - "strategy": "monitor_metric", - "metrics": ["storage"], - }, - "microsoft.documentdb/databaseaccounts": { - "category": "database", - "label": "Cosmos DB Account", - "strategy": "monitor_metric", - "metrics": ["DataUsage", "IndexUsage"], - }, - "microsoft.dbforpostgresql/flexibleservers": { - "category": "database", - "label": "PostgreSQL Flexible Server", - "strategy": "monitor_metric", - "metrics": ["storage_used"], - }, - "microsoft.dbformysql/flexibleservers": { - "category": "database", - "label": "MySQL Flexible Server", - "strategy": "monitor_metric", - "metrics": ["storage_used"], - }, - "microsoft.recoveryservices/vaults": { - "category": "backup", - "label": "Recovery Services Vault", - "strategy": "vault_warning", - }, -} - def filter_data_bearing_resources( resources: list[Any], + registry: dict[str, dict[str, Any]], ) -> list[tuple[Any, dict[str, Any]]]: matched = [] for resource in resources: - entry = EGRESS_RESOURCE_REGISTRY.get(resource.type.strip().lower()) + entry = registry.get(resource.type.strip().lower()) if entry: matched.append((resource, entry)) return matched @@ -295,8 +245,18 @@ def build_egress_inventory( ) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: rows = [] findings = [] - for resource, entry in filter_data_bearing_resources(resources): - collector = _STRATEGY_COLLECTORS[entry["strategy"]] + registry = load_egress_registry(1) + for resource, entry in filter_data_bearing_resources(resources, registry): + collector = _STRATEGY_COLLECTORS.get(entry["strategy"]) + if collector is None: + # Master data can ship a strategy ahead of the engine; skip that one + # resource type rather than aborting the whole egress run. + logger.warning( + "Unknown egress strategy %r for %s; skipping.", + entry["strategy"], + resource.type, + ) + continue try: row = collector(credential, resource_client, resource, entry) except Exception as e: diff --git a/core/utils_report_egress.py b/core/utils_report_egress.py index a8b55a5..5b67e00 100644 --- a/core/utils_report_egress.py +++ b/core/utils_report_egress.py @@ -23,7 +23,12 @@ ) from core.utils_db import load_data -from core.utils_egress import GIB, format_bytes +from core.utils_egress import ( + GIB, + compute_totals, + format_bytes, + read_egress_inventory, +) from core.utils_egress_aws import ARCHIVE_TIERS as AWS_ARCHIVE_TIERS from core.utils_egress_azure import ARCHIVE_TIERS as AZURE_ARCHIVE_TIERS from core.utils_report import ( @@ -55,6 +60,12 @@ "color": "rgba(83, 155, 255, 1)", "in_allocation": True, }, + "file": { + "label": "File Storage", + "badge": "File", + "color": "rgba(245, 158, 11, 1)", + "in_allocation": True, + }, "database": { "label": "Databases", "badge": "Database", @@ -372,9 +383,12 @@ def _load_estimate( with open(json_path, "r", encoding="utf-8") as json_file: payload = json.load(json_file) + # Run metadata stays in the raw artifact; the resources come from the + # assessment database, which is the store of record for the estimate. meta = payload["meta"] - rows = payload["data"]["resources"] - totals = payload["data"]["totals"] + db_path = os.path.join(report_path, "data", "assessment.db") + rows, archive_tiers = read_egress_inventory(db_path) + totals = compute_totals(rows, archive_tiers) pricing = load_pricing(report_path) total_fee, fees_by_id = build_fee_estimate( diff --git a/main.py b/main.py index 28a3faa..790cafd 100644 --- a/main.py +++ b/main.py @@ -794,6 +794,7 @@ def run_assessment( config["cloudServiceProvider"], config["providerDetails"], raw_data_path, + report_path=report_path, name=name, exit_strategy=config["exitStrategy"], assessment_type=config["assessmentType"], diff --git a/tests/test_utils_and_main.py b/tests/test_utils_and_main.py index 4b80a9a..27c5d92 100644 --- a/tests/test_utils_and_main.py +++ b/tests/test_utils_and_main.py @@ -613,7 +613,7 @@ class EgressStageTests(unittest.TestCase): _ESTIMATE_OK = { "success": True, "logs": "", - "json_path": "/tmp/report/raw/egress_estimate.json", + "json_path": "/tmp/report/raw/egress_inventory_raw_data.json", } _REPORT_OK = { "success": True, @@ -689,16 +689,17 @@ def test_egress_invoked_after_report_generation(self): 1, config["providerDetails"], "/tmp/report/raw", + report_path="/tmp/report", name=config["name"], exit_strategy=config["exitStrategy"], assessment_type=config["assessmentType"], ) mock_render.assert_called_once_with( - "/tmp/report", "/tmp/report/raw/egress_estimate.json" + "/tmp/report", "/tmp/report/raw/egress_inventory_raw_data.json" ) mock_pdf.assert_called_once_with( "/tmp/report", - "/tmp/report/raw/egress_estimate.json", + "/tmp/report/raw/egress_inventory_raw_data.json", config["providerDetails"], ) call_names = [name for name, _, _ in manager.mock_calls] @@ -806,6 +807,7 @@ def test_egress_invoked_for_aws_with_provider_code(self): 2, config["providerDetails"], "/tmp/report/raw", + report_path="/tmp/report", name=config["name"], exit_strategy=config["exitStrategy"], assessment_type=config["assessmentType"], diff --git a/tests/test_utils_egress.py b/tests/test_utils_egress.py index d1dfca9..ca157c8 100644 --- a/tests/test_utils_egress.py +++ b/tests/test_utils_egress.py @@ -1,6 +1,7 @@ # tests/test_utils_egress.py import json import os +import sqlite3 import tempfile import unittest from unittest.mock import patch @@ -11,6 +12,8 @@ estimate_egress, format_bytes, new_row, + read_egress_inventory, + write_egress_inventory, ) # Same envelope as the assessment JSON report (see generate_json_report). @@ -66,25 +69,148 @@ def test_archive_tier_set_controls_what_counts_as_archive(self): self.assertEqual(totals["archive_tier_bytes"], 20 * GIB) +# The sample rows all share one type; one catalogue entry covers them. +_SAMPLE_REGISTRY = { + "some/type": { + "resource_type_id": 1, + "category": "object", + "label": "Storage", + "strategy": "whatever", + } +} + + +_EGRESS_TABLES = """ +CREATE TABLE egress_inventory ( + id INTEGER PRIMARY KEY AUTOINCREMENT, resource_type INTEGER NOT NULL, + name TEXT NOT NULL, size_bytes INTEGER, + size_unknown INTEGER NOT NULL DEFAULT 0, flags TEXT, notes TEXT); +CREATE TABLE egress_inventory_tier ( + id INTEGER PRIMARY KEY AUTOINCREMENT, egress_inventory_id INTEGER NOT NULL, + tier TEXT NOT NULL, size_bytes INTEGER NOT NULL, + is_archive INTEGER NOT NULL DEFAULT 0); +""" + + +class EgressInventoryRoundTripTests(unittest.TestCase): + _CATALOGUE = """ +CREATE TABLE resourcetype ( + id INTEGER PRIMARY KEY, csp INTEGER NOT NULL, code TEXT NOT NULL, + name TEXT NOT NULL, icon TEXT NOT NULL, status TEXT NOT NULL); +CREATE TABLE resourcetype_data ( + id INTEGER PRIMARY KEY, resource_type INTEGER NOT NULL, + data_category TEXT NOT NULL, strategy TEXT NOT NULL, params TEXT, + status TEXT NOT NULL); +""" + + def setUp(self): + self._dir = tempfile.TemporaryDirectory() + self.addCleanup(self._dir.cleanup) + self.db_path = os.path.join(self._dir.name, "assessment.db") + conn = sqlite3.connect(self.db_path) + conn.executescript(self._CATALOGUE + _EGRESS_TABLES) + conn.execute( + "INSERT INTO resourcetype VALUES (7, 1, 'some/type', 'Storage', 'i', 't')" + ) + conn.execute( + "INSERT INTO resourcetype_data VALUES (1, 7, 'object', 's', '{}', 't')" + ) + conn.commit() + conn.close() + + def _round_trip(self, rows, archive_tiers): + write_egress_inventory(rows, {"some/type": 7}, archive_tiers, self.db_path) + return read_egress_inventory(self.db_path) + + def test_rows_survive_storage_unchanged(self): + original = _sample_rows() + + stored, _ = self._round_trip(original, {"Archive"}) + + self.assertEqual(len(stored), len(original)) + for before, after in zip(original, stored): + for field in ( + "name", + "size_bytes", + "size_unknown", + "tier_bytes", + "flags", + "notes", + ): + self.assertEqual(after[field], before[field], field) + + def test_label_and_category_come_back_from_the_catalogue(self): + stored, _ = self._round_trip(_sample_rows(), set()) + + self.assertEqual(stored[0]["label"], "Storage") + self.assertEqual(stored[0]["category"], "object") + self.assertEqual(stored[0]["type"], "some/type") + + def test_totals_are_identical_before_and_after_storage(self): + original = _sample_rows() + archive_tiers = {"Archive"} + + stored, stored_tiers = self._round_trip(original, archive_tiers) + + # What the report headlines show must not change with the store. + self.assertEqual( + compute_totals(stored, stored_tiers), + compute_totals(original, archive_tiers), + ) + + def test_is_archive_marks_only_the_archive_tiers(self): + _, stored_tiers = self._round_trip(_sample_rows(), {"Archive"}) + + self.assertEqual(stored_tiers, {"Archive"}) + + def test_row_without_a_catalogue_entry_warns_and_is_skipped(self): + orphan = new_row("/x", "x", "unmapped/type", "Nope", "object") + + with self.assertLogs("core.engine.egress", level="WARNING") as captured: + write_egress_inventory([orphan], {"some/type": 7}, set(), self.db_path) + + rows, _ = read_egress_inventory(self.db_path) + self.assertEqual(rows, []) + self.assertIn("unmapped/type", captured.output[0]) + + class EstimateEgressDispatchTests(unittest.TestCase): def _run(self, cloud_service_provider, provider_details): with tempfile.TemporaryDirectory() as tmp_dir: - result = estimate_egress( - cloud_service_provider, - provider_details, - tmp_dir, - name="Exit Assessment Test", - exit_strategy=3, - assessment_type=1, - ) + db_dir = os.path.join(tmp_dir, "data") + os.makedirs(db_dir) + conn = sqlite3.connect(os.path.join(db_dir, "assessment.db")) + conn.executescript(_EGRESS_TABLES) + conn.commit() + conn.close() + + with patch( + "core.utils_egress.load_egress_registry", + return_value=_SAMPLE_REGISTRY, + ): + result = estimate_egress( + cloud_service_provider, + provider_details, + tmp_dir, + report_path=tmp_dir, + name="Exit Assessment Test", + exit_strategy=3, + assessment_type=1, + ) payload = None if result["success"]: with open(result["json_path"], encoding="utf-8") as json_file: payload = json.load(json_file) self.assertEqual( result["json_path"], - os.path.join(tmp_dir, "egress_estimate.json"), + os.path.join(tmp_dir, "egress_inventory_raw_data.json"), ) + stored = ( + sqlite3.connect(os.path.join(tmp_dir, "data", "assessment.db")) + .execute("SELECT COUNT(*) FROM egress_inventory") + .fetchone()[0] + ) + self.assertEqual(stored, len(payload["data"]["resources"])) return result, payload @patch("core.utils_egress_azure.collect_azure_egress") @@ -108,6 +234,36 @@ def test_azure_dispatch_and_json_schema(self, mock_collect): self.assertEqual(payload["data"]["totals"]["known_size_bytes"], 100 * GIB) self.assertEqual(payload["data"]["totals"]["archive_tier_bytes"], 10 * GIB) + @patch("core.utils_egress_azure.collect_azure_egress") + def test_json_row_shape(self, mock_collect): + rows = _sample_rows() + mock_collect.return_value = (rows, {"Archive"}) + + _, payload = self._run(1, {"any": "details"}) + + resource = payload["data"]["resources"][0] + self.assertEqual( + list(resource), + [ + "id", + "name", + "code", + "category", + "size_bytes", + "size_unknown", + "tier_bytes", + "flags", + "notes", + ], + ) + self.assertEqual(resource["id"], "/sa1") + self.assertEqual(resource["name"], "sa1") + # "type" is published as "code"; the label lives in the catalogue. + self.assertEqual(resource["code"], "some/type") + self.assertNotIn("label", resource) + self.assertNotIn("type", resource) + self.assertEqual(resource["tier_bytes"], {"Hot": 50 * GIB, "Archive": 10 * GIB}) + @patch("core.utils_egress_aws.collect_aws_egress") def test_aws_dispatch_and_json_schema(self, mock_collect): mock_collect.return_value = (_sample_rows(), {"Archive"}) diff --git a/tests/test_utils_egress_aws.py b/tests/test_utils_egress_aws.py index a2768a4..f3a44e7 100644 --- a/tests/test_utils_egress_aws.py +++ b/tests/test_utils_egress_aws.py @@ -6,11 +6,9 @@ from core.utils_egress import GIB from core.utils_egress_aws import ( - EGRESS_RESOURCE_REGISTRY, - _collect_backup_vaults, _collect_dynamodb_tables, - _collect_ebs_snapshots, - _collect_ebs_volumes, + _collect_list_item_size, + _collect_not_sizeable, _collect_rds_instances, _collect_s3_buckets, _list_buckets_in_region, @@ -25,7 +23,110 @@ SNAPSHOT_CODE = "AWS.ec2.describe_snapshots.Snapshots" RDS_CODE = "AWS.rds.describe_db_instances.DBInstances" DYNAMODB_CODE = "AWS.dynamodb.list_tables.TableNames" -BACKUP_CODE = "AWS.backup.list_backup_vaults.BackupVaultList" +# The catalogue row enumerates backup plans; sizing overrides it with vaults. +BACKUP_CODE = "AWS.backup.list_backup_plans.BackupPlansList" + +# Stands in for what load_egress_registry() builds out of resourcetype_data, +# so these tests exercise the engine rather than the shipped master data. +TEST_REGISTRY = { + S3_CODE: { + "category": "object", + "label": "S3 (Simple Storage Service)", + "strategy": "s3_bucket_metrics", + "enumeration": { + "service": "s3", + "operation": "list_buckets", + "result_path": ["Buckets"], + }, + }, + VOLUME_CODE: { + "category": "block", + "label": "Elastic Block Store (EBS)", + "strategy": "ebs_volumes", + "enumeration": { + "service": "ec2", + "operation": "describe_volumes", + "result_path": ["Volumes"], + }, + "sizing": { + "id_field": "VolumeId", + "name_tag": "Name", + "size_field": "Size", + "size_unit": "GiB", + "flags": ["allocated (upper bound)"], + }, + }, + SNAPSHOT_CODE: { + "category": "block", + "label": "EBS Snapshots", + "strategy": "ebs_snapshots", + "enumeration": { + "service": "ec2", + "operation": "describe_snapshots", + "result_path": ["Snapshots"], + "kwargs": {"OwnerIds": ["self"]}, + }, + "sizing": { + "id_field": "SnapshotId", + "size_field": "VolumeSize", + "size_unit": "GiB", + "flags": ["allocated (upper bound)"], + "notes": ["incremental – shares blocks with sibling snapshots"], + }, + }, + RDS_CODE: { + "category": "database", + "label": "RDS (Relational Database Service)", + "strategy": "rds_instances", + "enumeration": { + "service": "rds", + "operation": "describe_db_instances", + "result_path": ["DBInstances"], + }, + }, + DYNAMODB_CODE: { + "category": "database", + "label": "DynamoDB", + "strategy": "dynamodb_tables", + "enumeration": { + "service": "dynamodb", + "operation": "list_tables", + "result_path": ["TableNames"], + }, + }, + BACKUP_CODE: { + "category": "backup", + "label": "Backup", + "strategy": "backup_vaults", + "enumeration": { + "service": "backup", + "operation": "list_backup_vaults", + "result_path": ["BackupVaultList"], + }, + "sizing": { + "id_field": "BackupVaultName", + "arn_field": "BackupVaultArn", + "flags": ["backup vault – not sized"], + "count_field": "NumberOfRecoveryPoints", + "count_note": "{count} recovery points (cannot be exported directly)", + }, + }, +} + +_REGISTRY_PATCHER = None + + +def setUpModule(): + global _REGISTRY_PATCHER + _REGISTRY_PATCHER = patch( + "core.utils_egress_aws.load_egress_registry", + return_value=TEST_REGISTRY, + ) + _REGISTRY_PATCHER.start() + + +def tearDownModule(): + _REGISTRY_PATCHER.stop() def _client_error(code, operation): @@ -114,7 +215,9 @@ def test_falls_back_to_bucket_region_field_when_filter_unsupported(self): }, ] - names = _list_buckets_in_region(s3_client, REGION) + names = _list_buckets_in_region( + s3_client, REGION, TEST_REGISTRY[S3_CODE]["enumeration"] + ) self.assertEqual(names, ["data-eu"]) s3_client.get_bucket_location.assert_not_called() @@ -130,7 +233,9 @@ def test_falls_back_to_get_bucket_location_when_field_missing(self): "LocationConstraint": REGION if Bucket == "data-eu" else None } - names = _list_buckets_in_region(s3_client, REGION) + names = _list_buckets_in_region( + s3_client, REGION, TEST_REGISTRY[S3_CODE]["enumeration"] + ) self.assertEqual(names, ["data-eu"]) @@ -144,7 +249,12 @@ def test_bucket_is_skipped_when_location_lookup_is_denied(self): "AccessDenied", "GetBucketLocation" ) - self.assertEqual(_list_buckets_in_region(s3_client, REGION), []) + self.assertEqual( + _list_buckets_in_region( + s3_client, REGION, TEST_REGISTRY[S3_CODE]["enumeration"] + ), + [], + ) class S3BucketCollectorTests(unittest.TestCase): @@ -201,7 +311,7 @@ def test_storage_type_split_archive_flag_and_region_filter(self): "q2": float(2 * GIB), }, ) - entry = EGRESS_RESOURCE_REGISTRY[S3_CODE] + entry = TEST_REGISTRY[S3_CODE] rows = _collect_s3_buckets(_mock_session(clients), REGION, S3_CODE, entry) @@ -224,7 +334,7 @@ def test_no_datapoints_records_unknown_size(self): metrics=[self._size_metric("fresh-bucket", "StandardStorage")], metric_values={}, ) - entry = EGRESS_RESOURCE_REGISTRY[S3_CODE] + entry = TEST_REGISTRY[S3_CODE] rows = _collect_s3_buckets(_mock_session(clients), REGION, S3_CODE, entry) @@ -241,7 +351,7 @@ def test_replication_configuration_adds_note(self): clients["s3"].get_bucket_replication.return_value = { "ReplicationConfiguration": {"Rules": [{}]} } - entry = EGRESS_RESOURCE_REGISTRY[S3_CODE] + entry = TEST_REGISTRY[S3_CODE] rows = _collect_s3_buckets(_mock_session(clients), REGION, S3_CODE, entry) @@ -265,9 +375,9 @@ def test_volume_size_is_allocated_upper_bound(self): } ], ) - entry = EGRESS_RESOURCE_REGISTRY[VOLUME_CODE] + entry = TEST_REGISTRY[VOLUME_CODE] - rows = _collect_ebs_volumes( + rows = _collect_list_item_size( _mock_session({"ec2": ec2_client}), REGION, VOLUME_CODE, entry ) @@ -281,9 +391,9 @@ def test_snapshots_are_owned_only_and_carry_shared_block_note(self): ec2_client, [{"Snapshots": [{"SnapshotId": "snap-1", "VolumeSize": 50}]}], ) - entry = EGRESS_RESOURCE_REGISTRY[SNAPSHOT_CODE] + entry = TEST_REGISTRY[SNAPSHOT_CODE] - rows = _collect_ebs_snapshots( + rows = _collect_list_item_size( _mock_session({"ec2": ec2_client}), REGION, SNAPSHOT_CODE, entry ) @@ -291,6 +401,43 @@ def test_snapshots_are_owned_only_and_carry_shared_block_note(self): self.assertEqual(rows[0]["size_bytes"], 50 * GIB) self.assertTrue(any("shares blocks" in note for note in rows[0]["notes"])) + def test_owner_filter_comes_from_entry_kwargs_not_the_source(self): + # The master data and the old hardcoded filter agree today, so only a + # changed entry proves the call is actually data-driven. + ec2_client = MagicMock() + paginator = _mock_paginator(ec2_client, [{"Snapshots": []}]) + entry = { + **TEST_REGISTRY[SNAPSHOT_CODE], + "enumeration": { + **TEST_REGISTRY[SNAPSHOT_CODE]["enumeration"], + "kwargs": {"OwnerIds": ["123456"]}, + }, + } + + _collect_list_item_size( + _mock_session({"ec2": ec2_client}), REGION, SNAPSHOT_CODE, entry + ) + + paginator.paginate.assert_called_once_with(OwnerIds=["123456"]) + + def test_entry_without_kwargs_passes_no_extra_arguments(self): + ec2_client = MagicMock() + paginator = _mock_paginator(ec2_client, [{"Snapshots": []}]) + entry = { + **TEST_REGISTRY[SNAPSHOT_CODE], + "enumeration": { + key: value + for key, value in TEST_REGISTRY[SNAPSHOT_CODE]["enumeration"].items() + if key != "kwargs" + }, + } + + _collect_list_item_size( + _mock_session({"ec2": ec2_client}), REGION, SNAPSHOT_CODE, entry + ) + + paginator.paginate.assert_called_once_with() + class RdsCollectorTests(unittest.TestCase): def _clients(self, instances, metric_results): @@ -311,7 +458,7 @@ def test_used_space_computed_from_free_storage_space(self): ], metric_results=[{"Id": "q0", "Values": [float(40 * GIB)]}], ) - entry = EGRESS_RESOURCE_REGISTRY[RDS_CODE] + entry = TEST_REGISTRY[RDS_CODE] rows = _collect_rds_instances(_mock_session(clients), REGION, RDS_CODE, entry) @@ -330,7 +477,7 @@ def test_missing_metric_falls_back_to_allocated_upper_bound(self): ], metric_results=[{"Id": "q0", "Values": []}], ) - entry = EGRESS_RESOURCE_REGISTRY[RDS_CODE] + entry = TEST_REGISTRY[RDS_CODE] rows = _collect_rds_instances(_mock_session(clients), REGION, RDS_CODE, entry) @@ -348,7 +495,7 @@ def test_aurora_instances_are_flagged_not_sized(self): ], metric_results=[], ) - entry = EGRESS_RESOURCE_REGISTRY[RDS_CODE] + entry = TEST_REGISTRY[RDS_CODE] rows = _collect_rds_instances(_mock_session(clients), REGION, RDS_CODE, entry) @@ -371,7 +518,7 @@ def test_table_and_index_sizes_are_summed(self): ], } } - entry = EGRESS_RESOURCE_REGISTRY[DYNAMODB_CODE] + entry = TEST_REGISTRY[DYNAMODB_CODE] rows = _collect_dynamodb_tables( _mock_session({"dynamodb": dynamodb_client}), REGION, DYNAMODB_CODE, entry @@ -383,7 +530,7 @@ def test_empty_table_is_zero_not_unknown(self): dynamodb_client = MagicMock() _mock_paginator(dynamodb_client, [{"TableNames": ["empty"]}]) dynamodb_client.describe_table.return_value = {"Table": {"TableSizeBytes": 0}} - entry = EGRESS_RESOURCE_REGISTRY[DYNAMODB_CODE] + entry = TEST_REGISTRY[DYNAMODB_CODE] rows = _collect_dynamodb_tables( _mock_session({"dynamodb": dynamodb_client}), REGION, DYNAMODB_CODE, entry @@ -406,9 +553,9 @@ def test_vault_is_flagged_not_sized_with_recovery_point_note(self): } ], ) - entry = EGRESS_RESOURCE_REGISTRY[BACKUP_CODE] + entry = TEST_REGISTRY[BACKUP_CODE] - rows = _collect_backup_vaults( + rows = _collect_not_sizeable( _mock_session({"backup": backup_client}), REGION, BACKUP_CODE, entry ) @@ -417,6 +564,74 @@ def test_vault_is_flagged_not_sized_with_recovery_point_note(self): self.assertIn("backup vault – not sized", rows[0]["flags"]) self.assertTrue(any("12 recovery points" in note for note in rows[0]["notes"])) + def test_call_comes_from_entry_enumeration_not_the_catalogue_code(self): + # The catalogue row enumerates backup plans for the inventory scan; + # sizing needs the vaults, so the entry overrides the call. + backup_client = MagicMock() + _mock_paginator(backup_client, [{"BackupVaultList": []}]) + session = _mock_session({"backup": backup_client}) + + _collect_not_sizeable(session, REGION, BACKUP_CODE, TEST_REGISTRY[BACKUP_CODE]) + + self.assertEqual(session.client.call_args.args[0], "backup") + backup_client.get_paginator.assert_called_once_with("list_backup_vaults") + + def test_enumeration_drives_a_different_service_and_operation(self): + other_client = MagicMock() + _mock_paginator(other_client, [{"Vaults": [{"BackupVaultName": "v1"}]}]) + session = _mock_session({"otherservice": other_client}) + entry = { + **TEST_REGISTRY[BACKUP_CODE], + "enumeration": { + "service": "otherservice", + "operation": "list_vaults", + "result_path": ["Vaults"], + }, + } + + rows = _collect_not_sizeable(session, REGION, BACKUP_CODE, entry) + + self.assertEqual(session.client.call_args.args[0], "otherservice") + other_client.get_paginator.assert_called_once_with("list_vaults") + self.assertEqual([row["name"] for row in rows], ["v1"]) + + def test_enumeration_kwargs_are_forwarded(self): + backup_client = MagicMock() + paginator = _mock_paginator(backup_client, [{"BackupVaultList": []}]) + entry = { + **TEST_REGISTRY[BACKUP_CODE], + "enumeration": { + **TEST_REGISTRY[BACKUP_CODE]["enumeration"], + "kwargs": {"ByVaultType": "BACKUP_VAULT"}, + }, + } + + _collect_not_sizeable( + _mock_session({"backup": backup_client}), REGION, BACKUP_CODE, entry + ) + + paginator.paginate.assert_called_once_with(ByVaultType="BACKUP_VAULT") + + def test_nested_result_path_is_walked(self): + backup_client = MagicMock() + _mock_paginator( + backup_client, [{"Outer": {"Inner": [{"BackupVaultName": "nested"}]}}] + ) + entry = { + **TEST_REGISTRY[BACKUP_CODE], + "enumeration": { + "service": "backup", + "operation": "list_backup_vaults", + "result_path": ["Outer", "Inner"], + }, + } + + rows = _collect_not_sizeable( + _mock_session({"backup": backup_client}), REGION, BACKUP_CODE, entry + ) + + self.assertEqual([row["name"] for row in rows], ["nested"]) + class CollectAwsEgressTests(unittest.TestCase): _PROVIDER_DETAILS = { @@ -476,5 +691,207 @@ def test_one_failing_service_does_not_fail_the_stage(self, mock_boto3): self.assertTrue(any(row["id"] == "vol-1" for row in rows)) +class UnknownStrategyTests(unittest.TestCase): + @patch("core.utils_egress_aws.boto3") + def test_unknown_strategy_warns_and_skips_without_raising(self, mock_boto3): + registry = { + "AWS.future.list_things.Things": { + "category": "object", + "label": "Something New", + "strategy": "not_implemented_yet", + }, + VOLUME_CODE: TEST_REGISTRY[VOLUME_CODE], + } + ec2_client = MagicMock() + _mock_paginator(ec2_client, [{"Volumes": [{"VolumeId": "vol-1", "Size": 10}]}]) + mock_boto3.Session.return_value = _mock_session({"ec2": ec2_client}) + + with patch("core.utils_egress_aws.load_egress_registry", return_value=registry): + with self.assertLogs("core.engine.egress.aws", level="WARNING") as captured: + rows, _ = collect_aws_egress( + { + "accessKey": "AKIAIOSFODNN7EXAMPLE", + "secretKey": "secret", + "region": REGION, + } + ) + + # The known strategy still runs; only the unknown one is skipped. + self.assertEqual([row["id"] for row in rows], ["vol-1"]) + self.assertIn("not_implemented_yet", captured.output[0]) + + +class CollectorFailureVisibilityTests(unittest.TestCase): + @patch("core.utils_egress_aws.boto3") + def test_a_dropped_resource_type_is_warned_not_buried_at_debug(self, mock_boto3): + # A master-data row missing its sizing block removes a whole resource + # type from the estimate; that must be visible in run.log. + registry = {VOLUME_CODE: {**TEST_REGISTRY[VOLUME_CODE]}} + del registry[VOLUME_CODE]["sizing"] + ec2_client = MagicMock() + _mock_paginator(ec2_client, [{"Volumes": [{"VolumeId": "vol-1", "Size": 10}]}]) + mock_boto3.Session.return_value = _mock_session({"ec2": ec2_client}) + + with patch("core.utils_egress_aws.load_egress_registry", return_value=registry): + with self.assertLogs("core.engine.egress.aws", level="WARNING") as captured: + rows, _ = collect_aws_egress( + { + "accessKey": "AKIAIOSFODNN7EXAMPLE", + "secretKey": "secret", + "region": REGION, + } + ) + + self.assertEqual(rows, []) + self.assertIn(VOLUME_CODE, captured.output[0]) + self.assertIn("KeyError", captured.output[0]) + self.assertIn("missing from the estimate", captured.output[0]) + + +class ParameterisedSizingTests(unittest.TestCase): + def _entry(self, sizing, result_key="FileSystems"): + return { + "category": "block", + "label": "Elastic File System", + "strategy": "ebs_volumes", + "enumeration": { + "service": "elasticfilesystem", + "operation": "describe_file_systems", + "result_path": [result_key], + }, + "sizing": sizing, + } + + def _collect(self, entry, items, result_key="FileSystems"): + client = MagicMock() + _mock_paginator(client, [{result_key: items}]) + return _collect_list_item_size( + _mock_session({"elasticfilesystem": client}), REGION, "AWS.efs", entry + ) + + def test_nested_size_field_and_byte_unit(self): + entry = self._entry( + { + "id_field": "FileSystemId", + "name_field": "Name", + "size_field": ["SizeInBytes", "Value"], + "size_unit": "B", + "flags": ["metered size"], + } + ) + + rows = self._collect( + entry, + [ + { + "FileSystemId": "fs-1", + "Name": "shared", + "SizeInBytes": {"Value": 4096}, + } + ], + ) + + self.assertEqual(rows[0]["size_bytes"], 4096) + self.assertEqual(rows[0]["name"], "shared") + self.assertEqual(rows[0]["flags"], ["metered size"]) + + def test_missing_nested_size_is_unknown_not_zero(self): + entry = self._entry( + { + "id_field": "FileSystemId", + "size_field": ["SizeInBytes", "Value"], + "size_unit": "B", + } + ) + + rows = self._collect(entry, [{"FileSystemId": "fs-1"}]) + + self.assertIsNone(rows[0]["size_bytes"]) + self.assertTrue(rows[0]["size_unknown"]) + + def test_flags_and_notes_are_withheld_when_size_is_unknown(self): + entry = self._entry( + { + "id_field": "FileSystemId", + "size_field": "Size", + "flags": ["allocated"], + "notes": ["incremental"], + } + ) + + rows = self._collect(entry, [{"FileSystemId": "fs-1"}]) + + self.assertEqual(rows[0]["flags"], []) + self.assertEqual(rows[0]["notes"], []) + + def test_name_tag_falls_back_to_the_identifier(self): + entry = self._entry( + {"id_field": "FileSystemId", "name_tag": "Name", "size_field": "Size"} + ) + + rows = self._collect( + entry, [{"FileSystemId": "fs-1", "Tags": [{"Key": "env", "Value": "prod"}]}] + ) + + self.assertEqual(rows[0]["name"], "fs-1") + + def test_arn_field_falls_back_to_the_identifier_when_absent(self): + client = MagicMock() + _mock_paginator(client, [{"Vaults": [{"VaultName": "v1"}]}]) + entry = { + "category": "backup", + "label": "Some Vault", + "strategy": "backup_vaults", + "enumeration": { + "service": "glacier", + "operation": "list_vaults", + "result_path": ["Vaults"], + }, + "sizing": { + "id_field": "VaultName", + "arn_field": "VaultARN", + "flags": ["not sized"], + "count_field": "NumberOfArchives", + "count_note": "{count} archives held", + }, + } + + rows = _collect_not_sizeable( + _mock_session({"glacier": client}), REGION, "AWS.glacier", entry + ) + + self.assertEqual(rows[0]["id"], "v1") + self.assertEqual(rows[0]["flags"], ["not sized"]) + self.assertEqual(rows[0]["notes"], []) + + def test_count_note_template_is_filled_from_the_item(self): + client = MagicMock() + _mock_paginator( + client, [{"Vaults": [{"VaultName": "v1", "NumberOfArchives": 7}]}] + ) + entry = { + "category": "backup", + "label": "Some Vault", + "strategy": "backup_vaults", + "enumeration": { + "service": "glacier", + "operation": "list_vaults", + "result_path": ["Vaults"], + }, + "sizing": { + "id_field": "VaultName", + "flags": [], + "count_field": "NumberOfArchives", + "count_note": "{count} archives held", + }, + } + + rows = _collect_not_sizeable( + _mock_session({"glacier": client}), REGION, "AWS.glacier", entry + ) + + self.assertEqual(rows[0]["notes"], ["7 archives held"]) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_utils_egress_azure.py b/tests/test_utils_egress_azure.py index 0a9ca76..97c8ff6 100644 --- a/tests/test_utils_egress_azure.py +++ b/tests/test_utils_egress_azure.py @@ -12,6 +12,74 @@ filter_data_bearing_resources, ) +# Stands in for what load_egress_registry() builds out of resourcetype_data, +# so these tests exercise the engine rather than the shipped master data. +TEST_REGISTRY = { + "microsoft.storage/storageaccounts": { + "category": "object", + "label": "Storage Account", + "strategy": "storage_account_metrics", + }, + "microsoft.compute/disks": { + "category": "block", + "label": "Managed Disks", + "strategy": "allocated_size_property", + "api_version": "2024-03-02", + "size_property": "diskSizeGB", + }, + "microsoft.compute/snapshots": { + "category": "block", + "label": "Managed Disk Snapshots", + "strategy": "allocated_size_property", + "api_version": "2024-03-02", + "size_property": "diskSizeGB", + }, + "microsoft.sql/servers/databases": { + "category": "database", + "label": "Azure SQL Database", + "strategy": "monitor_metric", + "metrics": ["storage"], + }, + "microsoft.documentdb/databaseaccounts": { + "category": "database", + "label": "Azure Cosmos DB", + "strategy": "monitor_metric", + "metrics": ["DataUsage", "IndexUsage"], + }, + "microsoft.dbforpostgresql/flexibleservers": { + "category": "database", + "label": "Azure Database for PostgreSQL", + "strategy": "monitor_metric", + "metrics": ["storage_used"], + }, + "microsoft.dbformysql/flexibleservers": { + "category": "database", + "label": "Azure Database for MySQL", + "strategy": "monitor_metric", + "metrics": ["storage_used"], + }, + "microsoft.recoveryservices/vaults": { + "category": "backup", + "label": "Azure Site Recovery", + "strategy": "vault_warning", + }, +} + +_REGISTRY_PATCHER = None + + +def setUpModule(): + global _REGISTRY_PATCHER + _REGISTRY_PATCHER = patch( + "core.utils_egress_azure.load_egress_registry", + return_value=TEST_REGISTRY, + ) + _REGISTRY_PATCHER.start() + + +def tearDownModule(): + _REGISTRY_PATCHER.stop() + def _mock_resource(resource_type, name, resource_id, sku_name=None): resource = MagicMock() @@ -42,7 +110,7 @@ def test_picks_data_bearing_types_out_of_mixed_list(self): _mock_resource("Microsoft.RecoveryServices/vaults", "vault1", "/vault1"), ] - matched = filter_data_bearing_resources(resources) + matched = filter_data_bearing_resources(resources, TEST_REGISTRY) matched_names = [resource.name for resource, _ in matched] self.assertEqual(matched_names, ["sa1", "disk1", "vault1"]) @@ -52,7 +120,7 @@ def test_matching_is_case_insensitive(self): _mock_resource("MICROSOFT.STORAGE/StorageAccounts", "sa1", "/sa1"), ] - matched = filter_data_bearing_resources(resources) + matched = filter_data_bearing_resources(resources, TEST_REGISTRY) self.assertEqual(len(matched), 1) self.assertEqual(matched[0][1]["strategy"], "storage_account_metrics") @@ -62,7 +130,7 @@ def test_returns_empty_for_no_data_bearing_resources(self): _mock_resource("Microsoft.Network/networkInterfaces", "nic1", "/nic1"), ] - self.assertEqual(filter_data_bearing_resources(resources), []) + self.assertEqual(filter_data_bearing_resources(resources, TEST_REGISTRY), []) class FetchMonitorMetricsTests(unittest.TestCase): @@ -325,5 +393,43 @@ def test_returns_rows_and_archive_tiers(self, mock_rmc_cls, mock_fetch): self.assertEqual(archive_tiers, {"Archive"}) +class UnknownStrategyTests(unittest.TestCase): + @patch("core.utils_egress_azure.fetch_monitor_metrics") + def test_unknown_strategy_warns_and_skips_without_raising(self, mock_fetch): + mock_fetch.side_effect = [ + {"UsedCapacity": [{"dimension": None, "value": float(GIB)}]}, + {"BlobCapacity": []}, + ] + registry = { + "microsoft.future/widgets": { + "category": "object", + "label": "Something New", + "strategy": "not_implemented_yet", + }, + "microsoft.storage/storageaccounts": TEST_REGISTRY[ + "microsoft.storage/storageaccounts" + ], + } + resources = [ + _mock_resource("Microsoft.Future/widgets", "widget1", "/w1"), + _mock_resource("Microsoft.Storage/storageAccounts", "sa1", "/sa1"), + ] + + with patch( + "core.utils_egress_azure.load_egress_registry", return_value=registry + ): + with self.assertLogs( + "core.engine.egress.azure", level="WARNING" + ) as captured: + rows, findings = build_egress_inventory( + MagicMock(), MagicMock(), resources + ) + + # The known strategy still runs; only the unknown one is skipped. + self.assertEqual([row["name"] for row in rows], ["sa1"]) + self.assertEqual(rows[0]["size_bytes"], GIB) + self.assertIn("not_implemented_yet", captured.output[0]) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_utils_egress_registry.py b/tests/test_utils_egress_registry.py new file mode 100644 index 0000000..17f7b81 --- /dev/null +++ b/tests/test_utils_egress_registry.py @@ -0,0 +1,281 @@ +# tests/test_utils_egress_registry.py +import os +import sqlite3 +import tempfile +import unittest +from unittest.mock import MagicMock, patch + +from core.utils_db import load_data +from core.utils_egress import estimate_egress, load_egress_registry +from core.utils_egress_azure import filter_data_bearing_resources + +# (id, csp, code, name, icon, status) +RESOURCE_TYPES = [ + (1, 1, "Microsoft.Storage/storageAccounts", "Storage Account", "sa.png", "t"), + # Parent status 'f': stays out of the inventory catalogue but still holds + # data that has to be egressed. + (2, 1, "Microsoft.Compute/disks", "Managed Disks", "disk.png", "f"), + (3, 1, "Microsoft.Network/virtualNetworks", "Virtual Network", "vnet.png", "t"), + (4, 2, "AWS.s3.list_buckets.Buckets", "S3 (Simple Storage Service)", "s3.png", "t"), + (5, 2, "AWS.backup.list_backup_plans.BackupPlansList", "Backup", "bk.png", "t"), +] + +# (id, resource_type, data_category, strategy, params, status) +RESOURCE_TYPE_DATA = [ + (1, 1, "object", "storage_account_metrics", "{}", "t"), + ( + 2, + 2, + "block", + "allocated_size_property", + '{"api_version": "2024-03-02", "size_property": "diskSizeGB"}', + "t", + ), + # Own status 'f': switched off, must never reach the registry. + (3, 3, "network", "should_never_load", "{}", "f"), + (4, 4, "object", "s3_bucket_metrics", "{}", "t"), + ( + 5, + 5, + "backup", + "backup_vaults", + '{"enumeration": {"service": "backup", ' + '"operation": "list_backup_vaults", "result_path": ["BackupVaultList"]}}', + "t", + ), +] + +_SCHEMA = """ +CREATE TABLE resourcetype ( + id INTEGER PRIMARY KEY, + csp INTEGER NOT NULL, + code TEXT NOT NULL, + name TEXT NOT NULL, + icon TEXT NOT NULL, + status TEXT NOT NULL +); +CREATE TABLE resourcetype_data ( + id INTEGER PRIMARY KEY, + resource_type INTEGER NOT NULL, + data_category TEXT NOT NULL, + strategy TEXT NOT NULL, + params TEXT, + status TEXT NOT NULL +); +""" + + +def _make_db(path, *, with_data_table=True, data_rows=RESOURCE_TYPE_DATA): + conn = sqlite3.connect(path) + schema = _SCHEMA + if not with_data_table: + schema = _SCHEMA[: _SCHEMA.index("CREATE TABLE resourcetype_data")] + conn.executescript(schema) + conn.executemany("INSERT INTO resourcetype VALUES (?,?,?,?,?,?)", RESOURCE_TYPES) + if with_data_table: + conn.executemany( + "INSERT INTO resourcetype_data VALUES (?,?,?,?,?,?)", data_rows + ) + conn.commit() + conn.close() + + +class _FixtureDbTests(unittest.TestCase): + WITH_DATA_TABLE = True + DATA_ROWS = RESOURCE_TYPE_DATA + + def setUp(self): + handle, self.db_path = tempfile.mkstemp(suffix=".db") + os.close(handle) + self.addCleanup(os.unlink, self.db_path) + _make_db( + self.db_path, + with_data_table=self.WITH_DATA_TABLE, + data_rows=self.DATA_ROWS, + ) + patcher = patch( + "core.utils_egress.load_data", + side_effect=lambda table: load_data(table, db_path=self.db_path), + ) + self.addCleanup(patcher.stop) + patcher.start() + + +class RegistryConstructionTests(_FixtureDbTests): + def test_azure_registry_joins_and_keys_by_lowercased_code(self): + registry = load_egress_registry(1) + + self.assertEqual( + sorted(registry), + ["microsoft.compute/disks", "microsoft.storage/storageaccounts"], + ) + + def test_aws_registry_keeps_code_verbatim(self): + registry = load_egress_registry(2) + + self.assertEqual( + sorted(registry), + [ + "AWS.backup.list_backup_plans.BackupPlansList", + "AWS.s3.list_buckets.Buckets", + ], + ) + + def test_params_are_merged_into_the_entry(self): + entry = load_egress_registry(1)["microsoft.compute/disks"] + + self.assertEqual(entry["api_version"], "2024-03-02") + self.assertEqual(entry["size_property"], "diskSizeGB") + + def test_label_comes_from_resourcetype_name(self): + registry = load_egress_registry(2) + + self.assertEqual( + registry["AWS.s3.list_buckets.Buckets"]["label"], + "S3 (Simple Storage Service)", + ) + + def test_category_and_strategy_come_from_resourcetype_data(self): + entry = load_egress_registry(1)["microsoft.storage/storageaccounts"] + + self.assertEqual(entry["category"], "object") + self.assertEqual(entry["strategy"], "storage_account_metrics") + + def test_enumeration_params_survive_the_merge(self): + registry = load_egress_registry(2) + entry = registry["AWS.backup.list_backup_plans.BackupPlansList"] + + self.assertEqual( + entry["enumeration"], + { + "service": "backup", + "operation": "list_backup_vaults", + "result_path": ["BackupVaultList"], + }, + ) + + def test_other_csp_rows_are_not_included(self): + self.assertNotIn("AWS.s3.list_buckets.Buckets", load_egress_registry(1)) + self.assertNotIn("microsoft.compute/disks", load_egress_registry(2)) + + +class ParamsCannotShadowCanonicalKeysTests(_FixtureDbTests): + DATA_ROWS = [ + ( + 1, + 1, + "object", + "storage_account_metrics", + '{"label": "hijacked", "category": "hijacked", "strategy": "hijacked"}', + "t", + ) + ] + + def test_canonical_keys_win_over_params(self): + entry = load_egress_registry(1)["microsoft.storage/storageaccounts"] + + self.assertEqual(entry["label"], "Storage Account") + self.assertEqual(entry["category"], "object") + self.assertEqual(entry["strategy"], "storage_account_metrics") + + +class StatusFilteringTests(_FixtureDbTests): + def test_row_with_disabled_parent_resourcetype_is_included(self): + # 513/514/515 ship with resourcetype.status = 'f' on purpose; filtering + # on it would drop the whole block category from the estimate. + registry = load_egress_registry(1) + + self.assertIn("microsoft.compute/disks", registry) + self.assertEqual(registry["microsoft.compute/disks"]["label"], "Managed Disks") + + def test_row_with_disabled_resourcetype_data_is_excluded(self): + registry = load_egress_registry(1) + + self.assertNotIn("microsoft.network/virtualnetworks", registry) + strategies = [entry["strategy"] for entry in registry.values()] + self.assertNotIn("should_never_load", strategies) + + +class AzureKeyMatchingTests(_FixtureDbTests): + def test_lowercased_key_matches_an_arm_resource_type(self): + registry = load_egress_registry(1) + resource = MagicMock() + # ARM returns the provider namespace in its own casing. + resource.type = "Microsoft.Storage/storageAccounts" + resource.name = "sa1" + + matched = filter_data_bearing_resources([resource], registry) + + self.assertEqual(len(matched), 1) + self.assertEqual(matched[0][1]["strategy"], "storage_account_metrics") + self.assertEqual(matched[0][1]["label"], "Storage Account") + + +class MissingTableTests(_FixtureDbTests): + WITH_DATA_TABLE = False + + def test_missing_table_fails_the_stage(self): + # There is no built-in fallback: without master data the estimate would + # be silently wrong, so the egress stage has to fail instead. + with self.assertRaises(sqlite3.Error): + load_egress_registry(1) + + def test_estimate_egress_reports_the_failure_instead_of_raising(self): + with patch( + "core.utils_egress_azure.collect_azure_egress", + side_effect=sqlite3.OperationalError("no such table: resourcetype_data"), + ): + result = estimate_egress( + 1, + {}, + tempfile.gettempdir(), + report_path=tempfile.gettempdir(), + name="acme", + exit_strategy=1, + assessment_type=1, + ) + + self.assertFalse(result["success"]) + self.assertIn("resourcetype_data", result["logs"]) + + +class EmptyRegistryTests(_FixtureDbTests): + DATA_ROWS = [] + + def test_no_matching_rows_raises(self): + with self.assertRaises(ValueError) as ctx: + load_egress_registry(2) + + self.assertIn("CSP 2", str(ctx.exception)) + + +class OtherCspOnlyTests(_FixtureDbTests): + DATA_ROWS = [(1, 1, "object", "storage_account_metrics", "{}", "t")] + + def test_csp_without_enabled_rows_raises_even_when_another_csp_has_them(self): + self.assertIn("microsoft.storage/storageaccounts", load_egress_registry(1)) + with self.assertRaises(ValueError): + load_egress_registry(2) + + +class MalformedParamsTests(_FixtureDbTests): + DATA_ROWS = [ + (1, 1, "object", "storage_account_metrics", "{not json", "t"), + (2, 2, "block", "allocated_size_property", None, "t"), + ] + + def test_bad_params_degrade_to_empty_without_dropping_the_row(self): + registry = load_egress_registry(1) + + self.assertEqual( + sorted(registry), + ["microsoft.compute/disks", "microsoft.storage/storageaccounts"], + ) + self.assertEqual( + registry["microsoft.storage/storageaccounts"]["strategy"], + "storage_account_metrics", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_utils_report_egress.py b/tests/test_utils_report_egress.py index 27f0e1a..cbd4c54 100644 --- a/tests/test_utils_report_egress.py +++ b/tests/test_utils_report_egress.py @@ -1,6 +1,7 @@ # tests/test_utils_report_egress.py import json import os +import pathlib import sqlite3 import tempfile import unittest @@ -11,8 +12,9 @@ from tests.report_fixtures import stage_report_assets -from core.utils_egress import GIB +from core.utils_egress import GIB, write_egress_inventory from core.utils_report_egress import ( + CATEGORIES, _build_allocation, _build_coverage_section, _build_data_landscape_section, @@ -101,6 +103,16 @@ def _totals(rows, unknown_count=0, archive_tier_bytes=0): } +def _archive_tiers(rows): + # Any tier carrying bytes in a fixture that declares archive bytes. + return { + tier + for row in rows + for tier in (row.get("tier_bytes") or {}) + if tier in ("Archive", "Glacier", "Deep Archive") + } + + def _payload(rows, unknown_count=0, archive_tier_bytes=0): return { "meta": { @@ -117,6 +129,93 @@ def _payload(rows, unknown_count=0, archive_tier_bytes=0): } +_EGRESS_SCHEMA = """ +CREATE TABLE resourcetype ( + id INTEGER PRIMARY KEY, csp INTEGER NOT NULL, code TEXT NOT NULL, + name TEXT NOT NULL, icon TEXT NOT NULL, status TEXT NOT NULL); +CREATE TABLE resourcetype_data ( + id INTEGER PRIMARY KEY, resource_type INTEGER NOT NULL, + data_category TEXT NOT NULL, strategy TEXT NOT NULL, params TEXT, + status TEXT NOT NULL); +CREATE TABLE egress_inventory ( + id INTEGER PRIMARY KEY AUTOINCREMENT, resource_type INTEGER NOT NULL, + name TEXT NOT NULL, size_bytes INTEGER, + size_unknown INTEGER NOT NULL DEFAULT 0, flags TEXT, notes TEXT); +CREATE TABLE egress_inventory_tier ( + id INTEGER PRIMARY KEY AUTOINCREMENT, egress_inventory_id INTEGER NOT NULL, + tier TEXT NOT NULL, size_bytes INTEGER NOT NULL, + is_archive INTEGER NOT NULL DEFAULT 0); +""" + + +def _seed_egress_inventory(report_path, rows, archive_tiers=frozenset()): + """Store fixture rows the way the engine does, via the real writer.""" + db_dir = os.path.join(report_path, "data") + os.makedirs(db_dir, exist_ok=True) + db_path = os.path.join(db_dir, "assessment.db") + conn = sqlite3.connect(db_path) + conn.executescript(_EGRESS_SCHEMA) + + # One catalogue row per distinct (type, label, category); the report reads + # label and category back through the FK. + codes, type_ids = {}, {} + for row in rows: + triple = (row["type"], row["label"], row["category"]) + if triple in codes: + continue + rt_id = len(codes) + 1 + code = row["type"] if row["type"] not in type_ids else f"{row['type']}#{rt_id}" + codes[triple] = code + type_ids[code] = rt_id + conn.execute( + "INSERT INTO resourcetype (id, csp, code, name, icon, status) " + "VALUES (?, 1, ?, ?, 'icon.png', 't')", + (rt_id, code, row["label"]), + ) + conn.execute( + "INSERT INTO resourcetype_data (id, resource_type, data_category, " + "strategy, params, status) VALUES (?, ?, ?, 'strategy', '{}', 't')", + (rt_id, rt_id, row["category"]), + ) + conn.commit() + conn.close() + + stored = [ + {**row, "type": codes[(row["type"], row["label"], row["category"])]} + for row in rows + ] + write_egress_inventory(stored, type_ids, set(archive_tiers), db_path) + return db_path + + +class CategoryCoverageTests(unittest.TestCase): + # The CHECK constraint on resourcetype_data.data_category. A category the + # report does not know about renders an unstyled badge and, worse, drops + # out of the allocation chart while still counting towards the totals. + DATA_CATEGORIES = {"object", "block", "file", "database", "backup"} + + def test_every_master_data_category_is_known_to_the_report(self): + self.assertEqual(set(CATEGORIES), self.DATA_CATEGORIES) + + def test_only_backup_is_left_out_of_the_allocation_chart(self): + excluded = {k for k, v in CATEGORIES.items() if not v["in_allocation"]} + + # Backup is enumerated but never sized, so it has nothing to allocate. + self.assertEqual(excluded, {"backup"}) + + def test_allocated_categories_have_a_colour_and_a_label(self): + for key, info in CATEGORIES.items(): + if info["in_allocation"]: + self.assertIsNotNone(info["color"], key) + self.assertIsNotNone(info["label"], key) + + def test_template_styles_every_category_badge(self): + css = pathlib.Path("assets/template/egress.html").read_text(encoding="utf-8") + + for key in CATEGORIES: + self.assertIn(f".category-{key} {{", css, f"no badge rule for {key!r}") + + class LoadPricingTests(unittest.TestCase): @staticmethod def _make_report_db(report_path, rows): @@ -384,9 +483,14 @@ def test_flags_and_notes_become_resource_detail(self): class GenerateEgressHtmlReportTests(unittest.TestCase): def _generate(self, payload): with tempfile.TemporaryDirectory() as tmp_dir: - json_path = os.path.join(tmp_dir, "egress_estimate.json") + json_path = os.path.join(tmp_dir, "egress_inventory_raw_data.json") with open(json_path, "w", encoding="utf-8") as json_file: json.dump(payload, json_file) + _seed_egress_inventory( + tmp_dir, + payload["data"]["resources"], + _archive_tiers(payload["data"]["resources"]), + ) with patch( "core.utils_report_egress.load_data", @@ -804,9 +908,14 @@ class GenerateEgressPdfReportTests(unittest.TestCase): def _generate(self, payload): with tempfile.TemporaryDirectory() as tmp_dir: stage_report_assets(tmp_dir) - json_path = os.path.join(tmp_dir, "egress_estimate.json") + json_path = os.path.join(tmp_dir, "egress_inventory_raw_data.json") with open(json_path, "w", encoding="utf-8") as json_file: json.dump(payload, json_file) + _seed_egress_inventory( + tmp_dir, + payload["data"]["resources"], + _archive_tiers(payload["data"]["resources"]), + ) with patch( "core.utils_report_egress.load_data",