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_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/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_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__":
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",