From 6b24c4552fee64932b6f9e5a2f4723c8d9e66027 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bence=20H=C3=A9zs=C5=91?= Date: Sun, 19 Jul 2026 11:41:27 +0200 Subject: [PATCH] add data landscape & egress fee estimation --- README.md | 11 + assets/template/egress.html | 577 ++++++++++++++++++ core/utils_db.py | 1 + core/utils_egress.py | 112 ++++ core/utils_egress_aws.py | 442 ++++++++++++++ core/utils_egress_azure.py | 338 +++++++++++ core/utils_report_egress.py | 940 ++++++++++++++++++++++++++++++ core/utils_report_pdf.py | 9 +- main.py | 98 +++- tests/test_utils_and_main.py | 230 +++++++- tests/test_utils_egress.py | 142 +++++ tests/test_utils_egress_aws.py | 480 +++++++++++++++ tests/test_utils_egress_azure.py | 329 +++++++++++ tests/test_utils_report_egress.py | 861 +++++++++++++++++++++++++++ utils/codes.py | 1 + 15 files changed, 4565 insertions(+), 6 deletions(-) create mode 100644 assets/template/egress.html create mode 100644 core/utils_egress.py create mode 100644 core/utils_egress_aws.py create mode 100644 core/utils_egress_azure.py create mode 100644 core/utils_report_egress.py create mode 100644 tests/test_utils_egress.py create mode 100644 tests/test_utils_egress_aws.py create mode 100644 tests/test_utils_egress_azure.py create mode 100644 tests/test_utils_report_egress.py diff --git a/README.md b/README.md index 9cc393b..816e928 100644 --- a/README.md +++ b/README.md @@ -105,6 +105,17 @@ See the [configuration reference](https://cloudexit.escapecloud.io/config/config Want to see how a regulatory-aligned report looks (DORA / FINMA / UK PRA)? Run with `--dry-run` and send the output `payload.json` to request_report@escapecloud.io — we'll generate a sample you can share with your risk or compliance team. +## Data Landscape & Egress Estimation (alpha) + +Add `--egress` to any assessment to measure how much data lives in the assessed scope and estimate the one-time internet egress fee for moving it out — based on the provider's tiered list prices, with no additional permissions required. + +```bash +python main.py azure --cli --egress +python main.py aws --profile PROFILE --egress +``` + +See the [egress reference](https://cloudexit.escapecloud.io/egress/overview.html) for details. + ## CI/CD cloudexit runs headlessly in CI pipelines via `--non-interactive` and environment variables. A ready-made GitHub Action is available: diff --git a/assets/template/egress.html b/assets/template/egress.html new file mode 100644 index 0000000..483b6c8 --- /dev/null +++ b/assets/template/egress.html @@ -0,0 +1,577 @@ + + + + + + + EscapeCloud Community Edition - Data Landscape & Egress Estimation + + + + + + + + + {% set provider_name = "Microsoft Azure" if cloud_service_provider == 1 else "Amazon Web Services" if cloud_service_provider == 2 else "Unknown Provider" %} + {% set assessment_name = "Basic" if assessment_type == 1 else "Standard" if assessment_type == 2 else "Unknown" %} + +
+
+

+ EscapeCloud + {{ name }} +

+
+ Egress Summary (PDF) +
+
+
+
+ +
+
+ +
+
+
+
+
+
+ + {% if cloud_service_provider == 1 %} + + + + + + + + + + + + + + + + + + + + + + + + {% elif cloud_service_provider == 2 %} + + + + + + + + + + + + + {% else %} + {{ provider_name }} + {% endif %} + +
+
+
Cloud Service Provider
+
{{ provider_name }}
+
+
+
+
+ +
+
+
Total Data
+
{{ total_data_display }}
+
+
+
+
+ +
+
+
Assessment Type
+
{{ assessment_name }}
+
+
+
+
+ +
+
+
Timestamp
+
{{ timestamp }}
+
+
+
+
+
+ +
+
+
+
+
+

Estimated Egress Fee

+
{{ total_fee_display }}
+
+
+ +
+
Estimated Egress Fee
+
+

+ Estimated one-time charge to transfer the identified + in-scope data out of the cloud over the public + internet. Calculated from the detected data volume, + the provider's standard egress price tiers, and + archive retrieval charges where applicable. Excludes + request/transaction fees, cross-region traffic, and + any resources whose size could not be measured. A "+" + means the figure is a lower-bound estimate. +

+
+
+
+
+ {% if unknown_count > 0 %} +
+ + Excludes {{ unknown_count }} resource{% if unknown_count != 1 %}s{% + endif %} with unavailable size. +
+ {% endif %} {% if free_tier %} +
+
+ +
+
+ Included in Free Tier +
+

+ Your scanned footprint ({{ known_data_display }}) falls entirely within the cloud provider's complimentary + monthly internet egress allowance + ({{ free_tier_limit_display }} free tier). +

+
+ {% elif has_fees %} +
+ +
+ {% else %} +
+
+ +

No fee estimate available.

+
+
+ {% endif %} +
+
+
+ +
+
+
+
+
+

Data Allocation

+
{{ total_data_display }}
+
+
+ +
+
Data Allocation
+
+

+ Breakdown of the identified in-scope data volume by + resource category. Based on the sizes detected during + the assessment and intended to show where the data + footprint sits. A "+" indicates that some resources + could not be fully measured. +

+
+
+
+
+ {% if has_allocation %} +
+ +
+ {% else %} +
+
+ +

No data volume available.

+
+
+ {% endif %} +
+
+
+
+
+
+
+
+

Data Landscape

+
+
+
+ + + + + + + + + + + {% if type_groups %} {% for group in type_groups %} + + + + + + + + + + + {% endfor %} {% else %} + + + + {% endif %} + +
#Resource TypeCategorySize
{{ loop.index }} + + {{ group.label | trim }} + {{ group.label }} + + {{ group.category_label }} + {{ group.size_display }}
+
+
+
Estimated Egress Fee
+
{{ group.fee_display }}
+
+
+
Resources
+
+ {% for resource in group.resources %} +
+ {{ resource.name }}{% if resource.fee_display %} + ({{ resource.fee_display }}){% endif %}{% if + resource.detail %} + – {{ resource.detail }}{% endif %} +
+ {% endfor %} +
+
+
+
+ No data-bearing resources were discovered during the + assessment. +
+
+
+
+
+
+ + + + + + {% if has_allocation %} + + {% endif %} + + {% if has_fees %} + + {% endif %} + +
+ + diff --git a/core/utils_db.py b/core/utils_db.py index 462c4ee..1c2d1ed 100644 --- a/core/utils_db.py +++ b/core/utils_db.py @@ -19,6 +19,7 @@ "alternativetechnology", "alternativetechnologyorganization", "risk", + "egresspricing", } diff --git a/core/utils_egress.py b/core/utils_egress.py new file mode 100644 index 0000000..83036d7 --- /dev/null +++ b/core/utils_egress.py @@ -0,0 +1,112 @@ +# core/utils_egress.py +import json +import logging +import os +from typing import Any +from datetime import datetime, timezone + +logger = logging.getLogger("core.engine.egress") + +GIB = 1024**3 + + +def new_row( + resource_id: str, name: str, resource_type: str, label: str, category: str +) -> dict[str, Any]: + return { + "id": resource_id, + "name": name, + "type": resource_type, + "label": label, + "category": category, + "size_bytes": None, + "size_unknown": False, + "tier_bytes": None, + "flags": [], + "notes": [], + } + + +def format_bytes(size_bytes: int | float | None) -> str: + if size_bytes is None: + return "n/a" + value = float(size_bytes) + for unit in ("B", "KiB", "MiB", "GiB", "TiB"): + if value < 1024: + return f"{value:.1f} {unit}" + value /= 1024 + return f"{value:.1f} PiB" + + +def compute_totals( + rows: list[dict[str, Any]], archive_tiers: set[str] +) -> dict[str, Any]: + known_size_bytes = sum( + row["size_bytes"] for row in rows if row["size_bytes"] is not None + ) + archive_tier_bytes = sum( + size + for row in rows + for tier, size in (row["tier_bytes"] or {}).items() + if tier in archive_tiers + ) + unknown_count = sum(1 for row in rows if row["size_unknown"]) + return { + "known_size_bytes": known_size_bytes, + "archive_tier_bytes": archive_tier_bytes, + "resources_discovered": len(rows), + "resources_with_unknown_size": unknown_count, + } + + +def estimate_egress( + cloud_service_provider: int, + provider_details: dict[str, Any], + raw_data_path: str, + *, + name: str, + exit_strategy: int, + assessment_type: int, +) -> dict[str, Any]: + try: + if cloud_service_provider == 1: # Azure + from .utils_egress_azure import collect_azure_egress + + rows, archive_tiers = collect_azure_egress(provider_details) + elif cloud_service_provider == 2: # AWS + from .utils_egress_aws import collect_aws_egress + + rows, archive_tiers = collect_aws_egress(provider_details) + else: + raise ValueError( + f"Unsupported cloud service provider: {cloud_service_provider}" + ) + + json_payload = { + "meta": { + "name": name, + "cloud_service_provider": cloud_service_provider, + "exit_strategy": exit_strategy, + "assessment_type": assessment_type, + "timestamp": datetime.now(timezone.utc).strftime( + "%Y-%m-%d %H:%M:%S UTC" + ), + }, + "data": { + "resources": rows, + "totals": compute_totals(rows, archive_tiers), + }, + } + json_path = os.path.join(raw_data_path, "egress_estimate.json") + with open(json_path, "w", encoding="utf-8") as json_file: + json.dump(json_payload, json_file, indent=4) + + return { + "success": True, + "logs": "Egress estimation completed successfully.", + "json_path": json_path, + } + + except Exception as e: + logger.error(f"Error estimating egress: {str(e)}", exc_info=True) + return {"success": False, "logs": str(e)} diff --git a/core/utils_egress_aws.py b/core/utils_egress_aws.py new file mode 100644 index 0000000..b7d16fb --- /dev/null +++ b/core/utils_egress_aws.py @@ -0,0 +1,442 @@ +# core/utils_egress_aws.py +import boto3 +import logging +from typing import Any +from datetime import datetime, timedelta, timezone +from botocore.exceptions import BotoCoreError, ClientError + +from .utils_egress import GIB, format_bytes, new_row + +logger = logging.getLogger("core.engine.egress.aws") + +METRIC_DATA_BATCH_SIZE = 500 + +METRICS_LOOKBACK_DAYS = 3 + +ARCHIVE_TIERS = {"Archive", "Glacier", "Deep Archive"} + +S3_STORAGE_TYPE_TIERS = { + "StandardStorage": "Standard", + "ReducedRedundancyStorage": "Standard", + "ExpressOneZone": "Standard", + "StandardIAStorage": "Standard-IA", + "StandardIASizeOverhead": "Standard-IA", + "OneZoneIAStorage": "One Zone-IA", + "OneZoneIASizeOverhead": "One Zone-IA", + "IntelligentTieringFAStorage": "Intelligent-Tiering", + "IntelligentTieringIAStorage": "Intelligent-Tiering", + "IntelligentTieringAIAStorage": "Intelligent-Tiering", + "IntelligentTieringAAStorage": "Archive", + "IntelligentTieringDAAStorage": "Deep Archive", + "GlacierInstantRetrievalStorage": "Glacier Instant Retrieval", + "GlacierInstantRetrievalSizeOverhead": "Glacier Instant Retrieval", + "GlacierStorage": "Glacier", + "GlacierStagingStorage": "Glacier", + "GlacierObjectOverhead": "Glacier", + "GlacierS3ObjectOverhead": "Glacier", + "DeepArchiveStorage": "Deep Archive", + "DeepArchiveObjectOverhead": "Deep Archive", + "DeepArchiveS3ObjectOverhead": "Deep Archive", + "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, + metric_specs: list[dict[str, Any]], + *, + lookback_days: int = METRICS_LOOKBACK_DAYS, + period: int = 86400, +) -> dict[str, float | None] | None: + if not metric_specs: + return {} + try: + end = datetime.now(timezone.utc) + start = end - timedelta(days=lookback_days) + results: dict[str, float | None] = {spec["id"]: None for spec in metric_specs} + + for batch_start in range(0, len(metric_specs), METRIC_DATA_BATCH_SIZE): + batch = metric_specs[batch_start : batch_start + METRIC_DATA_BATCH_SIZE] + queries = [ + { + "Id": spec["id"], + "MetricStat": { + "Metric": { + "Namespace": spec["namespace"], + "MetricName": spec["metric_name"], + "Dimensions": spec["dimensions"], + }, + "Period": period, + "Stat": spec.get("stat", "Average"), + }, + "ReturnData": True, + } + for spec in batch + ] + kwargs = { + "MetricDataQueries": queries, + "StartTime": start, + "EndTime": end, + "ScanBy": "TimestampDescending", + } + while True: + response = cloudwatch.get_metric_data(**kwargs) + for series in response.get("MetricDataResults", []): + values = series.get("Values") or [] + if values and results.get(series["Id"]) is None: + results[series["Id"]] = float(values[0]) + next_token = response.get("NextToken") + if not next_token: + break + kwargs["NextToken"] = next_token + return results + except (BotoCoreError, ClientError) as e: + logger.debug("CloudWatch metrics request failed: %s", str(e)) + return None + + +def _paginate(client: Any, operation_name: str, result_key: str, **kwargs) -> list: + items = [] + paginator = client.get_paginator(operation_name) + for page in paginator.paginate(**kwargs): + items.extend(page.get(result_key, [])) + return items + + +def _bucket_region(s3_client: Any, bucket_name: str) -> str: + location = s3_client.get_bucket_location(Bucket=bucket_name).get( + "LocationConstraint" + ) + if not location: + return "us-east-1" + if location == "EU": + return "eu-west-1" + return location + + +def _list_buckets_in_region(s3_client: Any, region: str) -> list[str]: + try: + response = s3_client.list_buckets(BucketRegion=region) + return [bucket["Name"] for bucket in response.get("Buckets", [])] + 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", []): + name = bucket["Name"] + bucket_region = bucket.get("BucketRegion") + if bucket_region is None: + try: + bucket_region = _bucket_region(s3_client, name) + except (BotoCoreError, ClientError) as e: + logger.debug("Skipping bucket %s: %s", name, str(e)) + continue + if bucket_region == region: + bucket_names.append(name) + return bucket_names + + +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) + cloudwatch = session.client("cloudwatch", region_name=region) + + bucket_names = _list_buckets_in_region(s3_client, region) + + rows = [] + for name in bucket_names: + row = new_row( + f"arn:aws:s3:::{name}", name, code, entry["label"], entry["category"] + ) + + try: + metrics = cloudwatch.list_metrics( + Namespace="AWS/S3", + MetricName="BucketSizeBytes", + Dimensions=[{"Name": "BucketName", "Value": name}], + ).get("Metrics", []) + except (BotoCoreError, ClientError) as e: + logger.debug("Listing metrics failed for bucket %s: %s", name, str(e)) + metrics = [] + + specs = [] + spec_tiers = {} + for index, metric in enumerate(metrics): + storage_type = next( + ( + dimension["Value"] + for dimension in metric.get("Dimensions", []) + if dimension["Name"] == "StorageType" + ), + "Unknown", + ) + spec_id = f"q{index}" + specs.append( + { + "id": spec_id, + "namespace": "AWS/S3", + "metric_name": "BucketSizeBytes", + "dimensions": metric.get("Dimensions", []), + } + ) + spec_tiers[spec_id] = S3_STORAGE_TYPE_TIERS.get(storage_type, storage_type) + + values = fetch_latest_metric_values(cloudwatch, specs) + tier_bytes: dict[str, int] = {} + if values: + for spec_id, value in values.items(): + if value is None: + continue + tier = spec_tiers[spec_id] + tier_bytes[tier] = tier_bytes.get(tier, 0) + int(value) + + if tier_bytes: + row["size_bytes"] = sum(tier_bytes.values()) + row["tier_bytes"] = tier_bytes + archive_bytes = sum( + size for tier, size in tier_bytes.items() if tier in ARCHIVE_TIERS + ) + if archive_bytes: + row["flags"].append( + f"Archive-class: {format_bytes(archive_bytes)} (restore required)" + ) + else: + row["size_unknown"] = True + + try: + s3_client.get_bucket_replication(Bucket=name) + row["notes"].append( + "replication configured (replica buckets are counted separately)" + ) + except (BotoCoreError, ClientError): + pass + + rows.append(row) + return rows + + +def _collect_ebs_volumes( + session: Any, region: str, code: str, entry: dict[str, Any] +) -> list[dict[str, Any]]: + ec2_client = session.client("ec2", region_name=region) + 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)") + else: + row["size_unknown"] = True + rows.append(row) + return rows + + +def _collect_ebs_snapshots( + session: Any, region: str, code: str, entry: dict[str, Any] +) -> list[dict[str, Any]]: + ec2_client = session.client("ec2", region_name=region) + 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 + rows.append(row) + return rows + + +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) + cloudwatch = session.client("cloudwatch", region_name=region) + + rows = [] + specs = [] + spec_rows: dict[str, tuple[dict[str, Any], int]] = {} + for index, instance in enumerate( + _paginate(rds_client, "describe_db_instances", "DBInstances") + ): + identifier = instance["DBInstanceIdentifier"] + row = new_row( + instance.get("DBInstanceArn", identifier), + identifier, + code, + entry["label"], + entry["category"], + ) + if (instance.get("Engine") or "").startswith("aurora"): + row["flags"].append("Aurora – cluster-level storage not sized") + rows.append(row) + continue + + allocated_bytes = int(instance.get("AllocatedStorage") or 0) * GIB + spec_id = f"q{index}" + specs.append( + { + "id": spec_id, + "namespace": "AWS/RDS", + "metric_name": "FreeStorageSpace", + "dimensions": [{"Name": "DBInstanceIdentifier", "Value": identifier}], + } + ) + spec_rows[spec_id] = (row, allocated_bytes) + rows.append(row) + + values = fetch_latest_metric_values(cloudwatch, specs, period=300) + for spec_id, (row, allocated_bytes) in spec_rows.items(): + free_bytes = values.get(spec_id) if values else None + if free_bytes is not None and allocated_bytes: + row["size_bytes"] = max(int(allocated_bytes - free_bytes), 0) + row["notes"].append(f"allocated: {format_bytes(allocated_bytes)}") + elif allocated_bytes: + row["size_bytes"] = allocated_bytes + row["flags"].append("allocated (upper bound)") + else: + row["size_unknown"] = True + return rows + + +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) + rows = [] + for table_name in _paginate(dynamodb_client, "list_tables", "TableNames"): + try: + table = dynamodb_client.describe_table(TableName=table_name).get( + "Table", {} + ) + except (BotoCoreError, ClientError) as e: + logger.debug("Describing table %s failed: %s", table_name, str(e)) + row = new_row( + table_name, table_name, code, entry["label"], entry["category"] + ) + row["size_unknown"] = True + rows.append(row) + continue + + row = new_row( + table.get("TableArn", table_name), + table_name, + code, + entry["label"], + entry["category"], + ) + size_bytes = int(table.get("TableSizeBytes") or 0) + size_bytes += sum( + int(index.get("IndexSizeBytes") or 0) + for index in table.get("GlobalSecondaryIndexes", []) + ) + row["size_bytes"] = size_bytes + rows.append(row) + 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) + 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_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, +} + + +def collect_aws_egress( + provider_details: dict[str, Any], +) -> tuple[list[dict[str, Any]], set[str]]: + region = provider_details["region"] + session = boto3.Session( + aws_access_key_id=provider_details["accessKey"], + aws_secret_access_key=provider_details["secretKey"], + aws_session_token=provider_details.get("sessionToken"), + region_name=region, + ) + + rows = [] + for code, entry in EGRESS_RESOURCE_REGISTRY.items(): + collector = _STRATEGY_COLLECTORS[entry["strategy"]] + 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 + ) + + return rows, ARCHIVE_TIERS diff --git a/core/utils_egress_azure.py b/core/utils_egress_azure.py new file mode 100644 index 0000000..6cada1e --- /dev/null +++ b/core/utils_egress_azure.py @@ -0,0 +1,338 @@ +# core/utils_egress_azure.py +import logging +import requests +from typing import Any +from datetime import datetime, timedelta, timezone +from azure.identity import ClientSecretCredential +from azure.mgmt.resource import ResourceManagementClient + +from .utils_egress import GIB, format_bytes, new_row + +logger = logging.getLogger("core.engine.egress.azure") + +VAULT_FINDING = ( + "Recovery Services vault detected: backup recovery points cannot be " + "exported from Azure. An exit means losing restore history or retaining " + "the vault until retention expires." +) + +METRICS_API_VERSION = "2018-01-01" +MANAGEMENT_SCOPE = "https://management.azure.com/.default" +MANAGEMENT_BASE_URL = "https://management.azure.com" + +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], +) -> list[tuple[Any, dict[str, Any]]]: + matched = [] + for resource in resources: + entry = EGRESS_RESOURCE_REGISTRY.get(resource.type.strip().lower()) + if entry: + matched.append((resource, entry)) + return matched + + +def _latest_average(datapoints: list[dict[str, Any]]) -> float | None: + for point in reversed(datapoints): + value = point.get("average") + if value is not None: + return float(value) + return None + + +def fetch_monitor_metrics( + credential: Any, + resource_id: str, + metric_names: list[str], + *, + dimension: str | None = None, + timeout: int = 30, +) -> dict[str, list[dict[str, Any]]] | None: + try: + token = credential.get_token(MANAGEMENT_SCOPE) + end = datetime.now(timezone.utc) + start = end - timedelta(days=2) + params = { + "api-version": METRICS_API_VERSION, + "metricnames": ",".join(metric_names), + "timespan": ( + f"{start.strftime('%Y-%m-%dT%H:%M:%SZ')}/" + f"{end.strftime('%Y-%m-%dT%H:%M:%SZ')}" + ), + "aggregation": "Average", + "interval": "PT1H", + } + if dimension: + params["$filter"] = f"{dimension} eq '*'" + response = requests.get( + f"{MANAGEMENT_BASE_URL}{resource_id}/providers/Microsoft.Insights/metrics", + headers={"Authorization": f"Bearer {token.token}"}, + params=params, + timeout=timeout, + ) + response.raise_for_status() + payload = response.json() + except requests.RequestException as e: + logger.debug("Metrics request failed for %s: %s", resource_id, str(e)) + return None + except Exception as e: + logger.debug("Metrics lookup failed for %s: %s", resource_id, str(e)) + return None + + result: dict[str, list[dict[str, Any]]] = {name: [] for name in metric_names} + for metric in payload.get("value", []): + name = metric.get("name", {}).get("value", "") + series_values = [] + for series in metric.get("timeseries", []): + dimension_value = None + if dimension: + for metadata in series.get("metadatavalues", []): + metadata_name = metadata.get("name", {}).get("value", "") + if metadata_name.lower() == dimension.lower(): + dimension_value = metadata.get("value") + value = _latest_average(series.get("data", [])) + if value is not None: + series_values.append({"dimension": dimension_value, "value": value}) + result[name] = series_values + return result + + +def _metric_total( + metrics: dict[str, list[dict[str, Any]]] | None, name: str +) -> float | None: + if not metrics or not metrics.get(name): + return None + return sum(point["value"] for point in metrics[name]) + + +def _base_row(resource: Any, entry: dict[str, Any]) -> dict[str, Any]: + row = new_row( + resource.id, resource.name, resource.type, entry["label"], entry["category"] + ) + row["findings"] = [] + return row + + +def _unknown_size_finding(resource: Any) -> dict[str, str]: + return { + "severity": "warning", + "resource": resource.name, + "message": ( + f"{resource.name}: size unknown – no metric datapoints in the last " + "2 days (capacity metrics update roughly daily); excluded from totals." + ), + } + + +def _collect_storage_account( + credential: Any, resource_client: Any, resource: Any, entry: dict[str, Any] +) -> dict[str, Any]: + row = _base_row(resource, entry) + + account_metrics = fetch_monitor_metrics(credential, resource.id, ["UsedCapacity"]) + used_capacity = _metric_total(account_metrics, "UsedCapacity") + + blob_metrics = fetch_monitor_metrics( + credential, + f"{resource.id}/blobServices/default", + ["BlobCapacity"], + dimension="Tier", + ) + tier_bytes: dict[str, int] = {} + if blob_metrics: + for point in blob_metrics.get("BlobCapacity", []): + tier = (point["dimension"] or "Unknown").capitalize() + tier_bytes[tier] = tier_bytes.get(tier, 0) + int(point["value"]) + + if used_capacity is not None: + row["size_bytes"] = int(used_capacity) + elif tier_bytes: + row["size_bytes"] = sum(tier_bytes.values()) + row["tier_bytes"] = tier_bytes or None + + archive_bytes = tier_bytes.get("Archive", 0) + if archive_bytes: + row["flags"].append( + f"Archive: {format_bytes(archive_bytes)} (rehydration required)" + ) + + sku_name = getattr(getattr(resource, "sku", None), "name", None) + if sku_name and any(geo in sku_name.upper() for geo in ("GRS", "GZRS")): + row["notes"].append(f"geo-replicated ({sku_name})") + row["findings"].append( + { + "severity": "info", + "resource": resource.name, + "message": ( + f"{resource.name}: SKU {sku_name} is geo-replicated; the " + "geo-secondary copy is not additional data to egress " + "(not double-counted)." + ), + } + ) + + if row["size_bytes"] is None: + row["size_unknown"] = True + row["findings"].append(_unknown_size_finding(resource)) + return row + + +def _collect_allocated_size( + credential: Any, resource_client: Any, resource: Any, entry: dict[str, Any] +) -> dict[str, Any]: + row = _base_row(resource, entry) + row["flags"].append("allocated (upper bound)") + + full_resource = resource_client.resources.get_by_id( + resource.id, entry["api_version"] + ) + size_gb = (full_resource.properties or {}).get(entry["size_property"]) + if size_gb: + row["size_bytes"] = int(size_gb) * GIB + else: + row["size_unknown"] = True + row["findings"].append(_unknown_size_finding(resource)) + return row + + +def _collect_monitor_metric( + credential: Any, resource_client: Any, resource: Any, entry: dict[str, Any] +) -> dict[str, Any]: + row = _base_row(resource, entry) + + metrics = fetch_monitor_metrics(credential, resource.id, entry["metrics"]) + values = [ + total + for name in entry["metrics"] + if (total := _metric_total(metrics, name)) is not None + ] + if values: + row["size_bytes"] = int(sum(values)) + else: + row["size_unknown"] = True + row["findings"].append(_unknown_size_finding(resource)) + return row + + +def _collect_vault( + credential: Any, resource_client: Any, resource: Any, entry: dict[str, Any] +) -> dict[str, Any]: + row = _base_row(resource, entry) + row["flags"].append("backup vault – not sized") + row["findings"].append( + { + "severity": "warning", + "resource": resource.name, + "message": f"{resource.name}: {VAULT_FINDING}", + } + ) + return row + + +_STRATEGY_COLLECTORS = { + "storage_account_metrics": _collect_storage_account, + "allocated_size_property": _collect_allocated_size, + "monitor_metric": _collect_monitor_metric, + "vault_warning": _collect_vault, +} + + +def build_egress_inventory( + credential: Any, resource_client: Any, resources: list[Any] +) -> 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"]] + try: + row = collector(credential, resource_client, resource, entry) + except Exception as e: + logger.debug( + "Egress sizing failed for %s: %s", resource.id, str(e), exc_info=True + ) + row = _base_row(resource, entry) + row["size_unknown"] = True + row["flags"].append("size unavailable") + row["findings"].append( + { + "severity": "warning", + "resource": resource.name, + "message": f"{resource.name}: size lookup failed ({str(e)}).", + } + ) + findings.extend(row.pop("findings")) + rows.append(row) + return rows, findings + + +def collect_azure_egress( + provider_details: dict[str, Any], +) -> tuple[list[dict[str, Any]], set[str]]: + credential = provider_details.get("credential") or ClientSecretCredential( + tenant_id=provider_details["tenantId"], + client_id=provider_details["clientId"], + client_secret=provider_details["clientSecret"], + ) + subscription_id = provider_details["subscriptionId"] + resource_group_name = provider_details["resourceGroupName"] + + resource_client = ResourceManagementClient(credential, subscription_id) + resources = list( + resource_client.resources.list_by_resource_group(resource_group_name) + ) + + rows, _ = build_egress_inventory(credential, resource_client, resources) + return rows, ARCHIVE_TIERS diff --git a/core/utils_report_egress.py b/core/utils_report_egress.py new file mode 100644 index 0000000..1218275 --- /dev/null +++ b/core/utils_report_egress.py @@ -0,0 +1,940 @@ +# core/utils_report_egress.py +import json +import logging +import os +import sqlite3 +from typing import Any +from jinja2 import Environment +from reportlab.lib import colors +from reportlab.lib.colors import HexColor +from reportlab.graphics.charts.piecharts import Pie +from reportlab.graphics.shapes import Drawing, Rect, String +from reportlab.lib.pagesizes import A4 +from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet +from reportlab.lib.units import cm +from reportlab.platypus import ( + Image, + PageBreak, + Paragraph, + SimpleDocTemplate, + Spacer, + Table, + TableStyle, +) + +from core.utils_db import load_data +from core.utils_egress import GIB, format_bytes +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 ( + _build_scope_section, + _build_summary_section, + _default_table_style, +) +from core.utils_report_pdf import draw_header_footer + +PDF_HEADER_TITLE = "EscapeCloud Community Edition - Data & Egress" + +logger = logging.getLogger("core.engine.report_egress") +logger.setLevel(logging.INFO) + +DEFAULT_PRICING_ZONE = "zone1" +UNIT_DIVISORS = {"GB": 10**9, "GiB": 2**30} + +ARCHIVE_TIERS_BY_CSP = {1: AZURE_ARCHIVE_TIERS, 2: AWS_ARCHIVE_TIERS} + +CATEGORIES = { + "object": { + "label": "Object Storage", + "badge": "Object", + "color": "rgba(34, 197, 94, 1)", + "in_allocation": True, + }, + "block": { + "label": "Block (allocated)", + "badge": "Block", + "color": "rgba(83, 155, 255, 1)", + "in_allocation": True, + }, + "database": { + "label": "Databases", + "badge": "Database", + "color": "rgba(168, 85, 247, 1)", + "in_allocation": True, + }, + "backup": { + "label": None, + "badge": "Backup", + "color": None, + "in_allocation": False, + }, +} +EGRESS_FEE_CHART_COLORS = [ + HexColor("#115e59"), + HexColor("#ffae1f"), + HexColor("#539bff"), + HexColor("#a855f7"), + HexColor("#22c55e"), + HexColor("#991b1b"), +] + +PROVIDER_NAMES = {1: "Microsoft Azure", 2: "Amazon Web Services"} +COMPONENT_LABELS = { + "internet_egress": "Internet Egress", + "archive_retrieval": "Archive Retrieval", +} +COVERAGE_NOTES = { + 1: [ + ("Storage Accounts", "Capacity metrics can lag by up to roughly 48 hours"), + ("Managed Disks, Snapshots", "Provisioned size; counted as an upper bound"), + ("SQL Databases", "Used space from database metrics"), + ("Cosmos DB Accounts", "Used space from database metrics"), + ("PostgreSQL / MySQL Flexible Servers", "Used space from database metrics"), + ("Recovery Services Vaults", "Not sized; recovery points cannot be exported"), + ], + 2: [ + ("S3 Buckets", "Metrics can lag by up to roughly 48 hours"), + ( + "EBS Volumes, Snapshots", + "Upper bound; snapshots can over-count shared blocks", + ), + ("RDS Instances", "Aurora is not sized"), + ("DynamoDB Tables", ""), + ("AWS Backup Vaults", "Not sized; recovery points cannot be exported"), + ], +} + +ALPHA_NOTE = ( + "NOTE: This Data Landscape & Egress Estimation feature is currently in " + "alpha version. Estimated values may be incomplete or inaccurate in some " + "environments." +) + +PRICING_SOURCE_LINKS = { + 1: { + "internet_egress": "https://azure.microsoft.com/pricing/details/bandwidth/", + "archive_retrieval": ( + "https://azure.microsoft.com/pricing/details/storage/blobs/" + ), + }, + 2: { + "internet_egress": "https://aws.amazon.com/ec2/pricing/on-demand/", + "archive_retrieval": "https://aws.amazon.com/s3/pricing/", + }, +} + +FALLBACK_ICONS = { + "microsoft.compute/disks": "/icons/azure/compute/10032-icon-service-Disks.png", + "microsoft.compute/snapshots": ( + "/icons/azure/compute/10026-icon-service-Disks-Snapshots.png" + ), + "aws.ec2.describe_volumes.volumes": "/icons/aws/Storage/Elastic-Block-Store.png", + "aws.ec2.describe_snapshots.snapshots": ( + "/icons/aws/Storage/Elastic-Block-Store.png" + ), + "aws.backup.list_backup_vaults.backupvaultlist": "/icons/aws/Storage/Backup.png", +} +DEFAULT_ICON = "/icons/misc/no_image.png" + + +def _unit_divisor(unit: str) -> float: + return UNIT_DIVISORS.get(unit, GIB) + + +def load_pricing(report_path: str) -> list[dict[str, Any]]: + db_path = os.path.join(report_path, "data", "assessment.db") + try: + rows = load_data("egresspricing", db_path=db_path) + except sqlite3.Error: + # Runs created before the dataset shipped the egresspricing table; + # fall back to the master dataset download. + logger.debug("egresspricing missing from %s; using master dataset", db_path) + rows = load_data("egresspricing") + return [row for row in rows if not row.get("valid_to")] + + +def calculate_tiered_cost(total_units: float, tiers: list[dict[str, Any]]) -> float: + cost = 0.0 + for tier in sorted(tiers, key=lambda tier: tier["tier_from"]): + lower = tier["tier_from"] + upper = tier["tier_to"] + if total_units <= lower: + break + covered = total_units if upper is None else min(total_units, upper) + cost += (covered - lower) * tier["price_per_unit"] + return cost + + +def build_fee_estimate( + rows: list[dict[str, Any]], + totals: dict[str, Any], + cloud_service_provider: int, + prices: list[dict[str, Any]], +) -> tuple[float, dict[str, float | None]]: + active = [ + price + for price in prices + if price["csp"] == cloud_service_provider + and price["zone"] == DEFAULT_PRICING_ZONE + ] + internet_tiers = [ + price for price in active if price["component"] == "internet_egress" + ] + if not internet_tiers: + raise ValueError( + "No active internet egress pricing for cloud service provider " + f"{cloud_service_provider} in the egresspricing dataset." + ) + retrieval_prices = [ + price for price in active if price["component"] == "archive_retrieval" + ] + retrieval_rate_per_byte = 0.0 + if retrieval_prices: + retrieval = retrieval_prices[0] + retrieval_rate_per_byte = retrieval["price_per_unit"] / _unit_divisor( + retrieval["unit"] + ) + archive_tiers = ARCHIVE_TIERS_BY_CSP.get(cloud_service_provider, set()) + + total_bytes = totals["known_size_bytes"] + internet_divisor = _unit_divisor(internet_tiers[0]["unit"]) + internet_fee = calculate_tiered_cost(total_bytes / internet_divisor, internet_tiers) + retrieval_fee = totals["archive_tier_bytes"] * retrieval_rate_per_byte + total_fee = internet_fee + retrieval_fee + + fees_by_id: dict[str, float | None] = {} + for row in rows: + if row["size_bytes"] is None: + fees_by_id[row["id"]] = None + continue + share = (row["size_bytes"] / total_bytes) if total_bytes else 0.0 + row_archive_bytes = sum( + size + for tier, size in (row["tier_bytes"] or {}).items() + if tier in archive_tiers + ) + fees_by_id[row["id"]] = ( + internet_fee * share + row_archive_bytes * retrieval_rate_per_byte + ) + return total_fee, fees_by_id + + +def format_fee(fee: float | None) -> str: + if fee is None: + return "n/a" + return f"${fee:,.2f}" + + +def free_tier_limit_display( + prices: list[dict[str, Any]], cloud_service_provider: int +) -> str | None: + tiers = sorted( + ( + price + for price in prices + if price["csp"] == cloud_service_provider + and price["component"] == "internet_egress" + and price["zone"] == DEFAULT_PRICING_ZONE + ), + key=lambda tier: tier["tier_from"], + ) + if tiers and tiers[0]["price_per_unit"] == 0.0 and tiers[0]["tier_to"]: + return f"{tiers[0]['tier_to']:g} {tiers[0]['unit']}" + return None + + +def _summarize_totals( + totals: dict[str, Any], + total_fee: float, + prices: list[dict[str, Any]], + cloud_service_provider: int, +) -> dict[str, Any]: + free_tier_limit = free_tier_limit_display(prices, cloud_service_provider) + free_tier = ( + total_fee == 0 + and totals["known_size_bytes"] > 0 + and free_tier_limit is not None + ) + unknown_count = totals["resources_with_unknown_size"] + total_fee_display = "$0" if free_tier else format_fee(total_fee) + total_data_display = format_bytes(totals["known_size_bytes"]) + if unknown_count > 0: + total_data_display = f"{total_data_display}+" + if not free_tier: + total_fee_display = f"{total_fee_display}+" + return { + "free_tier": free_tier, + "free_tier_limit": free_tier_limit, + "unknown_count": unknown_count, + "total_fee_display": total_fee_display, + "total_data_display": total_data_display, + } + + +def _build_icon_lookup(report_path: str) -> dict[str, str]: + db_path = os.path.join(report_path, "data", "assessment.db") + try: + return { + item["code"].strip().lower(): item["icon"] + for item in load_data("resourcetype", db_path=db_path) + if item.get("icon") + } + except Exception as e: + logger.debug("Resource type icon lookup unavailable: %s", str(e)) + return {} + + +def _resolve_icon(resource_type: str, icon_lookup: dict[str, str]) -> str: + code = resource_type.strip().lower() + candidate = code + while candidate: + if candidate in icon_lookup: + return icon_lookup[candidate] + if "/" not in candidate: + break + candidate = candidate.rsplit("/", 1)[0] + return FALLBACK_ICONS.get(code, DEFAULT_ICON) + + +def _build_allocation(rows: list[dict[str, Any]]) -> tuple[list, list, list]: + allocation_categories = { + key: info for key, info in CATEGORIES.items() if info["in_allocation"] + } + totals_by_category: dict[str, int] = {} + for row in rows: + if row["size_bytes"] is not None: + category = row.get("category") + if category in allocation_categories: + totals_by_category[category] = ( + totals_by_category.get(category, 0) + row["size_bytes"] + ) + + labels, values, colors = [], [], [] + for category, info in allocation_categories.items(): + size_bytes = totals_by_category.get(category, 0) + if size_bytes > 0: + labels.append(info["label"]) + values.append(size_bytes) + colors.append(info["color"]) + return labels, values, colors + + +def _build_type_groups( + rows: list[dict[str, Any]], + fees_by_id: dict[str, float | None], + icon_lookup: dict[str, str], +) -> list[dict[str, Any]]: + groups: dict[str, dict[str, Any]] = {} + for row in rows: + group = groups.setdefault( + row["label"], + { + "label": row["label"], + "icon": _resolve_icon(row["type"], icon_lookup), + "category": row["category"], + "category_label": ( + CATEGORIES.get(row["category"], {}).get("badge") or row["category"] + ), + "size_bytes": None, + "fee": None, + "resources": [], + }, + ) + if row["size_bytes"] is not None: + group["size_bytes"] = (group["size_bytes"] or 0) + row["size_bytes"] + fee = fees_by_id.get(row["id"]) + if fee is not None: + group["fee"] = (group["fee"] or 0.0) + fee + group["resources"].append( + { + "name": row["name"], + "fee_display": format_fee(fee) if fee is not None else None, + "detail": "; ".join(row["flags"] + row["notes"]), + } + ) + + for group in groups.values(): + group["size_display"] = format_bytes(group["size_bytes"]) + group["fee_display"] = format_fee(group["fee"]) + + return sorted( + groups.values(), + key=lambda group: ( + group["fee"] is None, + -(group["fee"] or 0.0), + -(group["size_bytes"] or 0), + ), + ) + + +def _load_estimate( + report_path: str, json_path: str +) -> tuple[dict, list, dict, dict, float, list]: + with open(json_path, "r", encoding="utf-8") as json_file: + payload = json.load(json_file) + + meta = payload["meta"] + rows = payload["data"]["resources"] + totals = payload["data"]["totals"] + + pricing = load_pricing(report_path) + total_fee, fees_by_id = build_fee_estimate( + rows, totals, meta["cloud_service_provider"], pricing + ) + + icon_lookup = _build_icon_lookup(report_path) + type_groups = _build_type_groups(rows, fees_by_id, icon_lookup) + return meta, rows, totals, pricing, total_fee, type_groups + + +def generate_egress_html_report(report_path: str, json_path: str) -> dict[str, Any]: + try: + meta, rows, totals, pricing, total_fee, type_groups = _load_estimate( + report_path, json_path + ) + + summary = _summarize_totals( + totals, total_fee, pricing, meta["cloud_service_provider"] + ) + + allocation_labels, allocation_values, allocation_colors = _build_allocation( + rows + ) + + fee_groups = [group for group in type_groups if (group["fee"] or 0.0) > 0] + fee_labels = [group["label"] for group in fee_groups] + fee_values = [round(group["fee"], 2) for group in fee_groups] + + template_path = os.path.join("assets", "template", "egress.html") + with open(template_path, "r") as file: + template_content = file.read() + + env = Environment(autoescape=True) + template = env.from_string(template_content) + html_content = template.render( + name=meta["name"], + cloud_service_provider=meta["cloud_service_provider"], + assessment_type=meta["assessment_type"], + timestamp=meta["timestamp"], + total_data_display=summary["total_data_display"], + total_fee_display=summary["total_fee_display"], + unknown_count=summary["unknown_count"], + has_allocation=bool(allocation_values), + allocation_labels_json=json.dumps(allocation_labels), + allocation_values_json=json.dumps(allocation_values), + allocation_colors_json=json.dumps(allocation_colors), + has_fees=bool(fee_values), + fee_labels_json=json.dumps(fee_labels), + fee_values_json=json.dumps(fee_values), + free_tier=summary["free_tier"], + free_tier_limit_display=summary["free_tier_limit"], + known_data_display=format_bytes(totals["known_size_bytes"]), + type_groups=type_groups, + ) + + html_path = os.path.join(report_path, "egress.html") + with open(html_path, "w") as report_file: + report_file.write(html_content) + + return { + "success": True, + "logs": "Egress report generated successfully.", + "html_path": html_path, + } + + except Exception as e: + logger.error(f"Error generating egress report: {str(e)}", exc_info=True) + return {"success": False, "logs": str(e)} + + +def _icon_flowable(report_path: str, icon: str) -> Any: + icon_path = os.path.join(report_path, "assets") + icon + if not os.path.exists(icon_path): + icon_path = os.path.join(report_path, "assets") + DEFAULT_ICON + if not os.path.exists(icon_path): + return "" + return Image(icon_path, width=20, height=20) + + +def _fee_breakdown_display(group: dict[str, Any]) -> str: + if group["fee"] is None: + return "n/a (not sized)" + return group["fee_display"] + + +def _egress_fee_chart_groups( + type_groups: list[dict[str, Any]], max_segments: int = 5 +) -> list[tuple[str, float]]: + fee_groups = [ + (group["label"], group["fee"]) + for group in type_groups + if group.get("fee") is not None and group["fee"] > 0 + ] + if len(fee_groups) <= max_segments: + return fee_groups + visible = fee_groups[: max_segments - 1] + other_total = sum(fee for _, fee in fee_groups[max_segments - 1 :]) + if other_total > 0: + visible.append(("Other Components", other_total)) + return visible + + +def _draw_egress_fee_chart(type_groups: list[dict[str, Any]]) -> Drawing | None: + chart_groups = _egress_fee_chart_groups(type_groups) + if not chart_groups: + return None + + total_fee = sum(fee for _, fee in chart_groups) + drawing = Drawing(360, 170) + + pie = Pie() + pie.x = 35 + pie.y = 20 + pie.width = 125 + pie.height = 125 + pie.data = [fee for _, fee in chart_groups] + pie.innerRadiusFraction = 0.55 + for index, _ in enumerate(chart_groups): + color = EGRESS_FEE_CHART_COLORS[index % len(EGRESS_FEE_CHART_COLORS)] + pie.slices[index].fillColor = color + pie.slices[index].strokeColor = colors.white + pie.slices[index].strokeWidth = 1 + drawing.add(pie) + + legend_x = 190 + legend_y = 130 + for index, (label, fee) in enumerate(chart_groups): + y = legend_y - index * 22 + color = EGRESS_FEE_CHART_COLORS[index % len(EGRESS_FEE_CHART_COLORS)] + share = (fee / total_fee * 100) if total_fee else 0 + drawing.add(Rect(legend_x, y, 8, 8, fillColor=color, strokeColor=color)) + drawing.add( + String( + legend_x + 14, + y - 1, + f"{label}: {share:.1f}%", + fontName="Helvetica", + fontSize=8, + fillColor=HexColor("#112726"), + ) + ) + return drawing + + +def _build_coverage_section(cloud_service_provider, styles, content_style): + rows = COVERAGE_NOTES.get(cloud_service_provider) + if not rows: + return [] + + content = [] + content.append( + Paragraph( + "The estimation covers the following data-related resource types " + "within the assessed scope:", + content_style, + ) + ) + table_data = [["Name", "Notes"]] + [[name, notes] for name, notes in rows] + coverage_table = Table(table_data, colWidths=[7 * cm, 8.5 * cm]) + coverage_table.setStyle(_default_table_style()) + content.append(coverage_table) + content.append(Spacer(1, 6)) + note_style = ParagraphStyle( + "CoverageNote", + fontSize=8, + leading=10, + textColor=HexColor("#6c757d"), + ) + content.append(Paragraph(ALPHA_NOTE, note_style)) + content.append(PageBreak()) + return content + + +def _build_estimated_costs_section( + total_fee, totals, free_tier, free_tier_limit, type_groups, styles, content_style +): + content = [] + content.append(Spacer(1, 12)) + content.append(Paragraph("Estimated Costs", styles["Heading2"])) + + unknown_count = totals["resources_with_unknown_size"] + if free_tier: + fee_display = "$0" + intro = ( + f"The scanned footprint ({format_bytes(totals['known_size_bytes'])}) " + "falls entirely within the cloud provider's complimentary monthly " + f"internet egress allowance ({free_tier_limit} free tier)." + ) + else: + fee_display = format_fee(total_fee) + if unknown_count > 0: + fee_display = f"{fee_display}+" + intro = ( + "Estimated one-time internet egress fee for transferring the data " + "identified on the previous page out of the cloud provider." + ) + content.append(Paragraph(intro, content_style)) + + card_title_style = ParagraphStyle( + "EgressCostCardTitle", + parent=content_style, + fontName="Helvetica-Bold", + fontSize=8, + leading=10, + textColor=HexColor("#115e59"), + ) + card_amount_style = ParagraphStyle( + "EgressCostCardAmount", + parent=content_style, + fontName="Helvetica-Bold", + fontSize=20, + leading=24, + textColor=HexColor("#112726"), + ) + warning_style = ParagraphStyle( + "EgressCostWarning", + parent=content_style, + fontSize=8, + leading=10, + textColor=HexColor("#664d03"), + ) + breakdown_header_style = ParagraphStyle( + "EgressCostBreakdownHeader", + parent=card_title_style, + textColor=colors.white, + ) + total_card_data = [ + [ + Paragraph("TOTAL ESTIMATED NETWORK TRANSIT FEE", card_title_style), + ], + [Paragraph(fee_display, card_amount_style)], + ] + if unknown_count > 0: + total_card_data.append( + [ + Paragraph( + f"Excludes {unknown_count} " + f"resource{'s' if unknown_count != 1 else ''} with " + "unavailable size calculations.", + warning_style, + ) + ] + ) + total_card = Table(total_card_data, colWidths=[15.5 * cm]) + total_card.setStyle( + TableStyle( + [ + ("BACKGROUND", (0, 0), (-1, -1), HexColor("#f8fafc")), + ("BOX", (0, 0), (-1, -1), 1, HexColor("#d7e3e1")), + ("LEFTPADDING", (0, 0), (-1, -1), 12), + ("RIGHTPADDING", (0, 0), (-1, -1), 12), + ("TOPPADDING", (0, 0), (-1, -1), 8), + ("BOTTOMPADDING", (0, 0), (-1, -1), 8), + ] + ) + ) + content.append(total_card) + content.append(Spacer(1, 12)) + + chart = _draw_egress_fee_chart(type_groups) + if chart: + chart_table = Table([[chart]], colWidths=[15.5 * cm]) + chart_table.setStyle( + TableStyle( + [ + ("BOX", (0, 0), (-1, -1), 1, HexColor("#d7e3e1")), + ("ALIGN", (0, 0), (-1, -1), "CENTER"), + ("TOPPADDING", (0, 0), (-1, -1), 10), + ("BOTTOMPADDING", (0, 0), (-1, -1), 10), + ] + ) + ) + content.append(chart_table) + content.append(Spacer(1, 12)) + + breakdown_data = [ + [ + Paragraph("Resource Type", breakdown_header_style), + Paragraph("Estimated Cost (List Rate)", breakdown_header_style), + ] + ] + for group in type_groups: + breakdown_data.append([group["label"], _fee_breakdown_display(group)]) + breakdown_data.append( + [ + Paragraph("Estimated Total Egress Fee", breakdown_header_style), + Paragraph(fee_display, breakdown_header_style), + ] + ) + breakdown_table = Table(breakdown_data, colWidths=[10 * cm, 5.5 * cm]) + breakdown_table.setStyle( + TableStyle( + [ + ("BACKGROUND", (0, 0), (-1, 0), HexColor("#115e59")), + ("TEXTCOLOR", (0, 0), (-1, 0), colors.white), + ("BACKGROUND", (0, -1), (-1, -1), HexColor("#115e59")), + ("TEXTCOLOR", (0, -1), (-1, -1), colors.white), + ("BOX", (0, 0), (-1, -1), 1, HexColor("#112726")), + ("INNERGRID", (0, 0), (-1, -1), 0.5, HexColor("#d7e3e1")), + ("ALIGN", (1, 1), (1, -1), "RIGHT"), + ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), + ("LEFTPADDING", (0, 0), (-1, -1), 8), + ("RIGHTPADDING", (0, 0), (-1, -1), 8), + ("TOPPADDING", (0, 0), (-1, -1), 7), + ("BOTTOMPADDING", (0, 0), (-1, -1), 7), + ] + ) + ) + content.append(Paragraph("Cost Breakdown", styles["Heading2"])) + content.append(breakdown_table) + content.append(PageBreak()) + return content + + +def _build_data_landscape_section( + type_groups, + total_size_display, + report_path, + styles, + content_style, +): + content = [] + content.append(Spacer(1, 12)) + content.append(Paragraph("Data Landscape", styles["Heading1"])) + content.append( + Paragraph( + "The Data Landscape summarizes the data-bearing cloud resources " + "identified within the defined scope, with the amount of data per " + "resource type:", + content_style, + ) + ) + content.append(Spacer(1, 12)) + + table_data = [["#", "Resource Type", "", "Size"]] + for index, group in enumerate(type_groups, start=1): + table_data.append( + [ + str(index), + group["label"], + _icon_flowable(report_path, group["icon"]), + group["size_display"], + ] + ) + table_data.append(["Total", "", "", total_size_display]) + + landscape_table = Table(table_data, colWidths=[1 * cm, 9 * cm, 1.5 * cm, 4 * cm]) + landscape_table.setStyle( + TableStyle( + [ + ("BACKGROUND", (0, 0), (-1, 0), HexColor("#115e59")), + ("TEXTCOLOR", (0, 0), (-1, 0), colors.white), + ("BACKGROUND", (0, -1), (-1, -1), HexColor("#115e59")), + ("TEXTCOLOR", (0, -1), (-1, -1), colors.white), + ("BOX", (0, 0), (-1, -1), 1, HexColor("#112726")), + ("BOTTOMPADDING", (0, 0), (-1, 0), 12), + ("TOPPADDING", (0, 0), (-1, 0), 12), + ("ALIGN", (0, 1), (1, -2), "LEFT"), + ("ALIGN", (2, 1), (2, -2), "CENTER"), + ("ALIGN", (3, 0), (-1, -1), "CENTER"), + ("VALIGN", (0, 1), (-1, -1), "MIDDLE"), + ] + ) + ) + content.append(landscape_table) + content.append(PageBreak()) + return content + + +def _tier_display(row: dict[str, Any]) -> str: + unit = row["unit"] + if row["tier_from"] == 0 and row["tier_to"] is None: + return "all volumes" + if row["tier_from"] == 0 and row["price_per_unit"] == 0: + return f"first {row['tier_to']:,.0f} {unit} / month" + if row["tier_to"] is None: + return f"over {row['tier_from']:,.0f} {unit}" + return f"{row['tier_from']:,.0f} {unit} – {row['tier_to']:,.0f} {unit}" + + +def _price_display(row: dict[str, Any]) -> str: + if row["price_per_unit"] == 0: + return "Free" + return f"${row['price_per_unit']:g} / {row['unit']}" + + +def _build_pricing_basis_section(prices, cloud_service_provider, styles, content_style): + rows = sorted( + ( + price + for price in prices + if price["csp"] == cloud_service_provider + and price["zone"] == DEFAULT_PRICING_ZONE + and price["component"] in COMPONENT_LABELS + ), + key=lambda price: ( + list(COMPONENT_LABELS).index(price["component"]), + price["tier_from"], + ), + ) + if not rows: + return [] + + content = [] + content.append(Spacer(1, 12)) + content.append(Paragraph("Appendix – Egress Prices", styles["Heading1"])) + + provider_name = PROVIDER_NAMES.get(cloud_service_provider, "Unknown Provider") + content.append( + Paragraph( + "Our calculations and estimations are based on the following " + f"{provider_name} list prices (Zone 1 – US/Europe, public " + "internet routing):", + content_style, + ) + ) + content.append(Spacer(1, 12)) + + show_valid_from = len({row["valid_from"] for row in rows}) > 1 + header = ["#", "Component", "Tier", "Price"] + if show_valid_from: + header.append("Valid from") + table_data = [header] + for index, row in enumerate(rows, start=1): + table_row = [ + str(index), + COMPONENT_LABELS[row["component"]], + _tier_display(row), + _price_display(row), + ] + if show_valid_from: + table_row.append(row["valid_from"]) + table_data.append(table_row) + + col_widths = ( + [1 * cm, 4 * cm, 5.5 * cm, 2.5 * cm, 2.5 * cm] + if show_valid_from + else [1 * cm, 4.5 * cm, 6 * cm, 4 * cm] + ) + pricing_table = Table(table_data, colWidths=col_widths) + pricing_table.setStyle( + TableStyle( + [ + ("BACKGROUND", (0, 0), (-1, 0), HexColor("#115e59")), + ("TEXTCOLOR", (0, 0), (-1, 0), colors.white), + ("BOX", (0, 0), (-1, -1), 1, HexColor("#112726")), + ("BOTTOMPADDING", (0, 0), (-1, 0), 12), + ("TOPPADDING", (0, 0), (-1, 0), 12), + ("ALIGN", (0, 1), (2, -1), "LEFT"), + ("ALIGN", (3, 0), (-1, -1), "CENTER"), + ("VALIGN", (0, 1), (-1, -1), "MIDDLE"), + ] + ) + ) + content.append(pricing_table) + content.append(Spacer(1, 12)) + + footnote_style = ParagraphStyle( + "PricingFootnote", + fontSize=8, + leading=10, + textColor=HexColor("#6c757d"), + spaceAfter=4, + ) + if not any(row["component"] == "archive_retrieval" for row in rows): + content.append( + Paragraph( + "Archive retrieval pricing was not available for this provider " + "at assessment time; rehydration fees are not included in the " + "estimate.", + footnote_style, + ) + ) + links = PRICING_SOURCE_LINKS.get(cloud_service_provider, {}) + components_shown = {row["component"] for row in rows} + source_lines = [ + (COMPONENT_LABELS[component], links[component]) + for component in COMPONENT_LABELS + if component in components_shown and component in links + ] + if len(source_lines) == 1: + content.append(Paragraph(f"Source: {source_lines[0][1]}", footnote_style)) + else: + for label, link in source_lines: + content.append(Paragraph(f"Source ({label}): {link}", footnote_style)) + content.append( + Paragraph( + "List prices at the time of the assessment; excludes " + "request/transaction costs and cross-region traffic.", + footnote_style, + ) + ) + return content + + +def generate_egress_pdf_report( + report_path: str, json_path: str, provider_details: dict[str, Any] +) -> dict[str, Any]: + try: + meta, rows, totals, pricing, total_fee, type_groups = _load_estimate( + report_path, json_path + ) + + summary = _summarize_totals( + totals, total_fee, pricing, meta["cloud_service_provider"] + ) + + pdf_path = os.path.join(report_path, "egress.pdf") + + def header_footer(canvas, doc): + draw_header_footer(report_path, canvas, doc, title=PDF_HEADER_TITLE) + + doc = SimpleDocTemplate( + pdf_path, pagesize=A4, title="EscapeCloud_-_Data_and_Egress_Report" + ) + styles = getSampleStyleSheet() + content_style = ParagraphStyle( + "ContentStyle", fontSize=10, leading=12, spaceAfter=10 + ) + styles["Heading1"].leading = 1.5 * styles["Heading1"].fontSize + styles["Heading1"].textColor = HexColor("#112726") + styles["Heading2"].leading = 1.5 * styles["Heading2"].fontSize + styles["Heading2"].textColor = HexColor("#112726") + + content = [] + content += _build_summary_section(meta, styles, content_style) + content += _build_scope_section(meta, provider_details, styles, content_style) + content += _build_coverage_section( + meta["cloud_service_provider"], styles, content_style + ) + content += _build_data_landscape_section( + type_groups, + summary["total_data_display"], + report_path, + styles, + content_style, + ) + content += _build_estimated_costs_section( + total_fee, + totals, + summary["free_tier"], + summary["free_tier_limit"], + type_groups, + styles, + content_style, + ) + content += _build_pricing_basis_section( + pricing, meta["cloud_service_provider"], styles, content_style + ) + + doc.build(content, onFirstPage=header_footer, onLaterPages=header_footer) + + return { + "success": True, + "logs": "Egress PDF report generated successfully.", + "pdf_path": pdf_path, + } + + except Exception as e: + logger.error(f"Error generating egress PDF report: {str(e)}", exc_info=True) + return {"success": False, "logs": str(e)} diff --git a/core/utils_report_pdf.py b/core/utils_report_pdf.py index 273c43a..2e4f365 100644 --- a/core/utils_report_pdf.py +++ b/core/utils_report_pdf.py @@ -107,14 +107,19 @@ def transform_alt_tech_for_pdf( return alt_tech -def draw_header_footer(report_path: str, canvas, doc) -> None: +def draw_header_footer( + report_path: str, + canvas, + doc, + title: str = "EscapeCloud Community Edition - Report", +) -> None: # Save the state of the canvas to not affect the drawing canvas.saveState() width, height = A4 # Include the date in the format mm-dd-yyyy current_date = datetime.now().strftime("%m-%d-%Y") - left_text_content1 = "EscapeCloud Community Edition - Report" + left_text_content1 = title left_text_content2 = f"Date: {current_date}" # Define the header content with Paragraphs diff --git a/main.py b/main.py index 01d3449..595ef40 100644 --- a/main.py +++ b/main.py @@ -21,6 +21,11 @@ sync_assessment, generate_report, ) +from core.utils_egress import estimate_egress +from core.utils_report_egress import ( + generate_egress_html_report, + generate_egress_pdf_report, +) from core.utils_sync import write_assessment_payload from utils.azure import ( select_subscription, @@ -253,7 +258,7 @@ def handle_aws(args): return # Run the AWS assessment pipeline - run_assessment(config, "aws", dry_run=args.dry_run) + run_assessment(config, "aws", dry_run=args.dry_run, egress=args.egress) def handle_azure(args): @@ -489,10 +494,10 @@ def handle_azure(args): # Run the Azure assessment pipeline # logger.info("Starting Azure assessment pipeline.") - run_assessment(config, "azure", dry_run=args.dry_run) + run_assessment(config, "azure", dry_run=args.dry_run, egress=args.egress) -def run_assessment(config, provider_name, *, dry_run=False): +def run_assessment(config, provider_name, *, dry_run=False, egress=False): # Record the assessment start time to propagate across stages started_at = int(time.time()) @@ -733,6 +738,69 @@ def run_assessment(config, provider_name, *, dry_run=False): ) sys.exit(codes.REPORT) + # Stage 7: Egress Estimation (opt-in via --egress). + # A failure here must not discard the completed assessment. + egress_failed = False + egress_json_path = None + egress_html_path = None + egress_pdf_path = None + if egress: + console.print("-------------------------------------------") + console.print("Stage #7 – Egress Estimation", style="bold") + + with console.status("Estimating egress data...", spinner="dots"): + egress_result = estimate_egress( + config["cloudServiceProvider"], + config["providerDetails"], + raw_data_path, + name=name, + exit_strategy=config["exitStrategy"], + assessment_type=config["assessmentType"], + ) + + if egress_result["success"]: + print_step("Estimating egress data...", status="ok") + egress_json_path = egress_result.get("json_path") + + with console.status("Generating egress report...", spinner="dots"): + egress_report_result = generate_egress_html_report( + report_path, egress_json_path + ) + + if egress_report_result["success"]: + print_step("Generating egress report...", status="ok") + egress_html_path = egress_report_result.get("html_path") + else: + print_step( + "Generating egress report...", + status="error", + logs=egress_report_result["logs"], + ) + egress_failed = True + + with console.status("Generating egress PDF...", spinner="dots"): + egress_pdf_result = generate_egress_pdf_report( + report_path, egress_json_path, config["providerDetails"] + ) + + if egress_pdf_result["success"]: + print_step("Generating egress PDF...", status="ok") + egress_pdf_path = egress_pdf_result.get("pdf_path") + else: + print_step( + "Generating egress PDF...", + status="error", + logs=egress_pdf_result["logs"], + ) + egress_failed = True + else: + print_step( + "Estimating egress data...", + status="error", + logs=egress_result["logs"], + ) + egress_failed = True + # Output the report path after the separator console.print("-------------------------------------------") console.print( @@ -758,8 +826,15 @@ def run_assessment(config, provider_name, *, dry_run=False): json_report_path = report_status.get("reports", {}).get("JSON") if json_report_path: console.print(f"JSON Report: {json_report_path}", style="cyan") + if egress_html_path: + console.print(f"Egress Report: {egress_html_path}", style="cyan") + if egress_pdf_path: + console.print(f"Egress PDF: {egress_pdf_path}", style="cyan") console.print("-------------------------------------------") + if egress_failed: + sys.exit(codes.EGRESS) + except Exception as e: console.print(f"[red]Unexpected error: {e}[/red]") sys.exit(codes.UNEXPECTED) @@ -780,6 +855,8 @@ def parse_arguments(): " python3 main.py azure --name 'DMS System' # Use a pre-defined assessment name\n" " python3 main.py aws --config config.json --dry-run # Local report + payload.json, no remote sync\n" " python3 main.py azure --config config.json --dry-run\n" + " python3 main.py aws --config config.json --egress # Estimate egress data volume\n" + " python3 main.py azure --config config.json --egress\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -815,6 +892,13 @@ def parse_arguments(): "remote sync." ), ) + aws_parser.add_argument( + "--egress", + action="store_true", + help=( + "Estimate how much data lives in the region and " "would need to move out." + ), + ) # Subparser for Azure azure_parser = subparsers.add_parser("azure", help="Perform an Azure assessment.") @@ -843,6 +927,14 @@ def parse_arguments(): "remote sync." ), ) + azure_parser.add_argument( + "--egress", + action="store_true", + help=( + "Estimate how much data lives in the resource group and " + "would need to move out." + ), + ) return parser.parse_args() diff --git a/tests/test_utils_and_main.py b/tests/test_utils_and_main.py index 7a0e0f9..dacc0c7 100644 --- a/tests/test_utils_and_main.py +++ b/tests/test_utils_and_main.py @@ -369,7 +369,7 @@ def test_dry_run_flag_passed_from_handle_aws(self): ): main.handle_aws(_ni_aws_args(dry_run=True)) - mock_run.assert_called_once_with(ANY, "aws", dry_run=True) + mock_run.assert_called_once_with(ANY, "aws", dry_run=True, egress=False) def _ni_aws_args(**kwargs): @@ -380,6 +380,7 @@ def _ni_aws_args(**kwargs): name=None, non_interactive=True, dry_run=False, + egress=False, ) defaults.update(kwargs) return Namespace(**defaults) @@ -393,6 +394,7 @@ def _ni_azure_args(**kwargs): name=None, non_interactive=True, dry_run=False, + egress=False, ) defaults.update(kwargs) return Namespace(**defaults) @@ -556,6 +558,232 @@ def test_missing_resource_group_exits_config(self): self.assertEqual(ctx.exception.code, codes.CONFIG) +class EgressStageTests(unittest.TestCase): + _ESTIMATE_OK = { + "success": True, + "logs": "", + "json_path": "/tmp/report/raw/egress_estimate.json", + } + _REPORT_OK = { + "success": True, + "logs": "", + "html_path": "/tmp/report/egress.html", + } + _PDF_OK = { + "success": True, + "logs": "", + "pdf_path": "/tmp/report/egress.pdf", + } + + @staticmethod + def _azure_config(): + return { + "name": "Azure Egress Test", + "cloudServiceProvider": 1, + "exitStrategy": 1, + "assessmentType": 1, + "providerDetails": { + "credential": MagicMock(), + "tenantId": "tenant-id", + "subscriptionId": "sub-id", + "resourceGroupName": "my-rg", + }, + } + + def test_egress_module_not_invoked_without_flag(self): + patches = _base_patches() + for p in patches: + p.start() + mock_estimate = patch("main.estimate_egress").start() + mock_render = patch("main.generate_egress_html_report").start() + mock_pdf = patch("main.generate_egress_pdf_report").start() + try: + main.run_assessment(self._azure_config(), "azure") + finally: + patch.stopall() + + mock_estimate.assert_not_called() + mock_render.assert_not_called() + mock_pdf.assert_not_called() + + def test_egress_invoked_after_report_generation(self): + manager = MagicMock() + patches = _base_patches() + for p in patches: + p.start() + mock_report = patch( + "main.generate_report", + return_value={"success": True, "reports": {}}, + ).start() + mock_estimate = patch( + "main.estimate_egress", return_value=self._ESTIMATE_OK + ).start() + mock_render = patch( + "main.generate_egress_html_report", return_value=self._REPORT_OK + ).start() + mock_pdf = patch( + "main.generate_egress_pdf_report", return_value=self._PDF_OK + ).start() + manager.attach_mock(mock_report, "generate_report") + manager.attach_mock(mock_estimate, "estimate_egress") + manager.attach_mock(mock_render, "generate_egress_html_report") + manager.attach_mock(mock_pdf, "generate_egress_pdf_report") + config = self._azure_config() + try: + main.run_assessment(config, "azure", egress=True) + finally: + patch.stopall() + + mock_estimate.assert_called_once_with( + 1, + config["providerDetails"], + "/tmp/report/raw", + 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" + ) + mock_pdf.assert_called_once_with( + "/tmp/report", + "/tmp/report/raw/egress_estimate.json", + config["providerDetails"], + ) + call_names = [name for name, _, _ in manager.mock_calls] + self.assertLess( + call_names.index("generate_report"), call_names.index("estimate_egress") + ) + self.assertLess( + call_names.index("estimate_egress"), + call_names.index("generate_egress_html_report"), + ) + self.assertLess( + call_names.index("generate_egress_html_report"), + call_names.index("generate_egress_pdf_report"), + ) + + def test_egress_failure_still_prints_outputs_and_exits_egress(self): + patches = _base_patches() + for p in patches: + p.start() + patch( + "main.estimate_egress", + return_value={"success": False, "logs": "metrics api error"}, + ).start() + mock_render = patch("main.generate_egress_html_report").start() + mock_pdf = patch("main.generate_egress_pdf_report").start() + mock_print = patch("main.console.print").start() + try: + with self.assertRaises(SystemExit) as ctx: + main.run_assessment(self._azure_config(), "azure", egress=True) + finally: + patch.stopall() + + self.assertEqual(ctx.exception.code, codes.EGRESS) + mock_render.assert_not_called() + mock_pdf.assert_not_called() + printed = [str(call) for call in mock_print.call_args_list] + self.assertTrue(any("Outputs:" in line for line in printed)) + + def test_egress_report_failure_still_prints_outputs_and_exits_egress(self): + patches = _base_patches() + for p in patches: + p.start() + patch("main.estimate_egress", return_value=self._ESTIMATE_OK).start() + patch( + "main.generate_egress_html_report", + return_value={"success": False, "logs": "template error"}, + ).start() + patch("main.generate_egress_pdf_report", return_value=self._PDF_OK).start() + mock_print = patch("main.console.print").start() + try: + with self.assertRaises(SystemExit) as ctx: + main.run_assessment(self._azure_config(), "azure", egress=True) + finally: + patch.stopall() + + self.assertEqual(ctx.exception.code, codes.EGRESS) + printed = [str(call) for call in mock_print.call_args_list] + self.assertTrue(any("Outputs:" in line for line in printed)) + # The JSON path is internal input for the reports; it is never listed + # in Outputs. The failed HTML report is not listed, but the PDF ran + # independently and succeeded, so it is. + self.assertFalse(any("Egress Estimate:" in line for line in printed)) + self.assertFalse(any("Egress Report:" in line for line in printed)) + self.assertTrue(any("Egress PDF:" in line for line in printed)) + + def test_egress_pdf_failure_still_prints_outputs_and_exits_egress(self): + patches = _base_patches() + for p in patches: + p.start() + patch("main.estimate_egress", return_value=self._ESTIMATE_OK).start() + patch("main.generate_egress_html_report", return_value=self._REPORT_OK).start() + patch( + "main.generate_egress_pdf_report", + return_value={"success": False, "logs": "reportlab error"}, + ).start() + mock_print = patch("main.console.print").start() + try: + with self.assertRaises(SystemExit) as ctx: + main.run_assessment(self._azure_config(), "azure", egress=True) + finally: + patch.stopall() + + self.assertEqual(ctx.exception.code, codes.EGRESS) + printed = [str(call) for call in mock_print.call_args_list] + self.assertTrue(any("Outputs:" in line for line in printed)) + self.assertTrue(any("Egress Report:" in line for line in printed)) + self.assertFalse(any("Egress PDF:" in line for line in printed)) + + def test_egress_invoked_for_aws_with_provider_code(self): + patches = _base_patches() + for p in patches: + p.start() + mock_estimate = patch( + "main.estimate_egress", return_value=self._ESTIMATE_OK + ).start() + patch("main.generate_egress_html_report", return_value=self._REPORT_OK).start() + patch("main.generate_egress_pdf_report", return_value=self._PDF_OK).start() + config = VALID_CONFIG.copy() + try: + main.run_assessment(config, "aws", egress=True) + finally: + patch.stopall() + + mock_estimate.assert_called_once_with( + 2, + config["providerDetails"], + "/tmp/report/raw", + name=config["name"], + exit_strategy=config["exitStrategy"], + assessment_type=config["assessmentType"], + ) + + def test_handle_azure_passes_egress_flag(self): + mock_credential = MagicMock() + with ( + patch.dict(os.environ, NonInteractiveAzureTests._BASE_ENV, clear=False), + patch("main.ClientSecretCredential", return_value=mock_credential), + patch("main.run_assessment") as mock_run, + patch("main.console.print"), + ): + main.handle_azure(_ni_azure_args(egress=True)) + + mock_run.assert_called_once_with(ANY, "azure", dry_run=False, egress=True) + + def test_handle_aws_passes_egress_flag(self): + with ( + patch.dict(os.environ, NonInteractiveAWSTests._BASE_ENV, clear=False), + patch("main.validate_region"), + patch("main.run_assessment") as mock_run, + patch("main.console.print"), + ): + main.handle_aws(_ni_aws_args(egress=True)) + + mock_run.assert_called_once_with(ANY, "aws", dry_run=False, egress=True) + + class ResolveModeEnvVarTests(unittest.TestCase): def test_host_and_key_env_override_config(self): fake_config = types.SimpleNamespace(HOST="config-host.io", KEY="config-key") diff --git a/tests/test_utils_egress.py b/tests/test_utils_egress.py new file mode 100644 index 0000000..d1dfca9 --- /dev/null +++ b/tests/test_utils_egress.py @@ -0,0 +1,142 @@ +# tests/test_utils_egress.py +import json +import os +import tempfile +import unittest +from unittest.mock import patch + +from core.utils_egress import ( + GIB, + compute_totals, + estimate_egress, + format_bytes, + new_row, +) + +# Same envelope as the assessment JSON report (see generate_json_report). +EXPECTED_META_KEYS = [ + "assessment_type", + "cloud_service_provider", + "exit_strategy", + "name", + "timestamp", +] +EXPECTED_DATA_KEYS = ["resources", "totals"] + + +def _sample_rows(): + storage = new_row("/sa1", "sa1", "some/type", "Storage", "object") + storage["size_bytes"] = 60 * GIB + storage["tier_bytes"] = {"Hot": 50 * GIB, "Archive": 10 * GIB} + disk = new_row("/disk1", "disk1", "some/type", "Disk", "block") + disk["size_bytes"] = 40 * GIB + unknown = new_row("/db1", "db1", "some/type", "Database", "database") + unknown["size_unknown"] = True + return [storage, disk, unknown] + + +class FormatBytesTests(unittest.TestCase): + def test_formats_binary_units_and_none(self): + self.assertEqual(format_bytes(None), "n/a") + self.assertEqual(format_bytes(512), "512.0 B") + self.assertEqual(format_bytes(GIB), "1.0 GiB") + self.assertEqual(format_bytes(1536 * GIB), "1.5 TiB") + + +class ComputeTotalsTests(unittest.TestCase): + def test_sums_known_sizes_archive_tiers_and_unknowns(self): + totals = compute_totals(_sample_rows(), archive_tiers={"Archive"}) + + self.assertEqual(totals["known_size_bytes"], 100 * GIB) + self.assertEqual(totals["archive_tier_bytes"], 10 * GIB) + self.assertEqual(totals["resources_discovered"], 3) + self.assertEqual(totals["resources_with_unknown_size"], 1) + + def test_archive_tier_set_controls_what_counts_as_archive(self): + rows = [new_row("/b1", "b1", "some/type", "Bucket", "object")] + rows[0]["size_bytes"] = 30 * GIB + rows[0]["tier_bytes"] = { + "Standard": 10 * GIB, + "Glacier": 15 * GIB, + "Deep Archive": 5 * GIB, + } + + totals = compute_totals(rows, archive_tiers={"Glacier", "Deep Archive"}) + + self.assertEqual(totals["archive_tier_bytes"], 20 * GIB) + + +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, + ) + 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"), + ) + return result, payload + + @patch("core.utils_egress_azure.collect_azure_egress") + def test_azure_dispatch_and_json_schema(self, mock_collect): + mock_collect.return_value = (_sample_rows(), {"Archive"}) + + result, payload = self._run(1, {"any": "details"}) + + self.assertTrue(result["success"]) + mock_collect.assert_called_once_with({"any": "details"}) + # Same meta/data envelope as the assessment JSON report; cost + # scenarios and findings belong to the Platform offering and must + # not leak into the JSON output. + self.assertEqual(sorted(payload.keys()), ["data", "meta"]) + self.assertEqual(sorted(payload["meta"].keys()), EXPECTED_META_KEYS) + self.assertEqual(sorted(payload["data"].keys()), EXPECTED_DATA_KEYS) + self.assertEqual(payload["meta"]["name"], "Exit Assessment Test") + self.assertEqual(payload["meta"]["cloud_service_provider"], 1) + self.assertEqual(payload["meta"]["exit_strategy"], 3) + self.assertEqual(payload["meta"]["assessment_type"], 1) + self.assertEqual(payload["data"]["totals"]["known_size_bytes"], 100 * GIB) + self.assertEqual(payload["data"]["totals"]["archive_tier_bytes"], 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"}) + + result, payload = self._run(2, {"region": "eu-central-1"}) + + self.assertTrue(result["success"]) + mock_collect.assert_called_once_with({"region": "eu-central-1"}) + self.assertEqual(sorted(payload.keys()), ["data", "meta"]) + self.assertEqual(sorted(payload["meta"].keys()), EXPECTED_META_KEYS) + self.assertEqual(sorted(payload["data"].keys()), EXPECTED_DATA_KEYS) + self.assertEqual(payload["meta"]["cloud_service_provider"], 2) + + def test_unsupported_provider_fails_without_raising(self): + result, payload = self._run(3, {}) + + self.assertFalse(result["success"]) + self.assertIn("Unsupported cloud service provider", result["logs"]) + self.assertIsNone(payload) + + @patch("core.utils_egress_azure.collect_azure_egress") + def test_collection_error_returns_failure(self, mock_collect): + mock_collect.side_effect = RuntimeError("enumeration failed") + + result, payload = self._run(1, {}) + + self.assertFalse(result["success"]) + self.assertEqual(result["logs"], "enumeration failed") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_utils_egress_aws.py b/tests/test_utils_egress_aws.py new file mode 100644 index 0000000..a2768a4 --- /dev/null +++ b/tests/test_utils_egress_aws.py @@ -0,0 +1,480 @@ +# tests/test_utils_egress_aws.py +import unittest +from unittest.mock import MagicMock, patch + +from botocore.exceptions import ClientError + +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_rds_instances, + _collect_s3_buckets, + _list_buckets_in_region, + collect_aws_egress, + fetch_latest_metric_values, +) + +REGION = "eu-central-1" + +S3_CODE = "AWS.s3.list_buckets.Buckets" +VOLUME_CODE = "AWS.ec2.describe_volumes.Volumes" +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" + + +def _client_error(code, operation): + return ClientError({"Error": {"Code": code, "Message": code}}, operation) + + +def _mock_session(clients): + session = MagicMock() + session.client.side_effect = lambda service, **kwargs: clients[service] + return session + + +def _mock_paginator(client, pages): + paginator = MagicMock() + paginator.paginate.return_value = pages + client.get_paginator.return_value = paginator + return paginator + + +class FetchLatestMetricValuesTests(unittest.TestCase): + def test_returns_latest_value_per_query(self): + cloudwatch = MagicMock() + cloudwatch.get_metric_data.return_value = { + "MetricDataResults": [ + {"Id": "q0", "Values": [123.0, 50.0]}, + {"Id": "q1", "Values": []}, + ] + } + specs = [ + {"id": "q0", "namespace": "AWS/S3", "metric_name": "m", "dimensions": []}, + {"id": "q1", "namespace": "AWS/S3", "metric_name": "m", "dimensions": []}, + ] + + result = fetch_latest_metric_values(cloudwatch, specs) + + # ScanBy=TimestampDescending: the first value is the latest datapoint. + self.assertEqual(result, {"q0": 123.0, "q1": None}) + kwargs = cloudwatch.get_metric_data.call_args.kwargs + self.assertEqual(kwargs["ScanBy"], "TimestampDescending") + + def test_merges_values_across_pages(self): + cloudwatch = MagicMock() + cloudwatch.get_metric_data.side_effect = [ + { + "MetricDataResults": [{"Id": "q0", "Values": []}], + "NextToken": "page-2", + }, + {"MetricDataResults": [{"Id": "q0", "Values": [42.0]}]}, + ] + specs = [ + {"id": "q0", "namespace": "AWS/S3", "metric_name": "m", "dimensions": []}, + ] + + result = fetch_latest_metric_values(cloudwatch, specs) + + self.assertEqual(result, {"q0": 42.0}) + + def test_empty_specs_return_empty_dict_without_api_call(self): + cloudwatch = MagicMock() + + self.assertEqual(fetch_latest_metric_values(cloudwatch, []), {}) + cloudwatch.get_metric_data.assert_not_called() + + def test_returns_none_on_client_error(self): + cloudwatch = MagicMock() + cloudwatch.get_metric_data.side_effect = _client_error( + "AccessDenied", "GetMetricData" + ) + specs = [ + {"id": "q0", "namespace": "AWS/S3", "metric_name": "m", "dimensions": []}, + ] + + self.assertIsNone(fetch_latest_metric_values(cloudwatch, specs)) + + +class ListBucketsInRegionTests(unittest.TestCase): + def test_falls_back_to_bucket_region_field_when_filter_unsupported(self): + s3_client = MagicMock() + s3_client.list_buckets.side_effect = [ + _client_error("InvalidRequest", "ListBuckets"), + { + "Buckets": [ + {"Name": "data-eu", "BucketRegion": REGION}, + {"Name": "data-us", "BucketRegion": "us-east-1"}, + ] + }, + ] + + names = _list_buckets_in_region(s3_client, REGION) + + self.assertEqual(names, ["data-eu"]) + s3_client.get_bucket_location.assert_not_called() + + def test_falls_back_to_get_bucket_location_when_field_missing(self): + s3_client = MagicMock() + s3_client.list_buckets.side_effect = [ + _client_error("InvalidRequest", "ListBuckets"), + {"Buckets": [{"Name": "data-eu"}, {"Name": "legacy-us"}]}, + ] + s3_client.get_bucket_location.side_effect = lambda Bucket: { + # us-east-1 buckets report None via the legacy API. + "LocationConstraint": REGION if Bucket == "data-eu" else None + } + + names = _list_buckets_in_region(s3_client, REGION) + + self.assertEqual(names, ["data-eu"]) + + def test_bucket_is_skipped_when_location_lookup_is_denied(self): + s3_client = MagicMock() + s3_client.list_buckets.side_effect = [ + _client_error("InvalidRequest", "ListBuckets"), + {"Buckets": [{"Name": "no-access"}]}, + ] + s3_client.get_bucket_location.side_effect = _client_error( + "AccessDenied", "GetBucketLocation" + ) + + self.assertEqual(_list_buckets_in_region(s3_client, REGION), []) + + +class S3BucketCollectorTests(unittest.TestCase): + def _clients(self, bucket_locations, metrics, metric_values): + s3_client = MagicMock() + + def list_buckets(**kwargs): + names = bucket_locations + if "BucketRegion" in kwargs: + names = [ + name + for name, location in bucket_locations.items() + if (location or "us-east-1") == kwargs["BucketRegion"] + ] + return {"Buckets": [{"Name": name} for name in names]} + + s3_client.list_buckets.side_effect = list_buckets + s3_client.get_bucket_location.side_effect = lambda Bucket: { + "LocationConstraint": bucket_locations[Bucket] + } + s3_client.get_bucket_replication.side_effect = _client_error( + "ReplicationConfigurationNotFoundError", "GetBucketReplication" + ) + cloudwatch = MagicMock() + cloudwatch.list_metrics.return_value = {"Metrics": metrics} + cloudwatch.get_metric_data.return_value = { + "MetricDataResults": [ + {"Id": query_id, "Values": [value]} + for query_id, value in metric_values.items() + ] + } + return {"s3": s3_client, "cloudwatch": cloudwatch} + + @staticmethod + def _size_metric(bucket, storage_type): + return { + "Dimensions": [ + {"Name": "BucketName", "Value": bucket}, + {"Name": "StorageType", "Value": storage_type}, + ] + } + + def test_storage_type_split_archive_flag_and_region_filter(self): + clients = self._clients( + bucket_locations={"data-eu": REGION, "data-us": None}, + metrics=[ + self._size_metric("data-eu", "StandardStorage"), + self._size_metric("data-eu", "GlacierStorage"), + self._size_metric("data-eu", "DeepArchiveStorage"), + ], + metric_values={ + "q0": float(50 * GIB), + "q1": float(8 * GIB), + "q2": float(2 * GIB), + }, + ) + entry = EGRESS_RESOURCE_REGISTRY[S3_CODE] + + rows = _collect_s3_buckets(_mock_session(clients), REGION, S3_CODE, entry) + + # The us-east-1 bucket is outside the assessment region, and the + # filter must not rely on GetBucketLocation (not in ViewOnlyAccess). + self.assertEqual([row["name"] for row in rows], ["data-eu"]) + clients["s3"].list_buckets.assert_called_once_with(BucketRegion=REGION) + clients["s3"].get_bucket_location.assert_not_called() + row = rows[0] + self.assertEqual(row["size_bytes"], 60 * GIB) + self.assertEqual( + row["tier_bytes"], + {"Standard": 50 * GIB, "Glacier": 8 * GIB, "Deep Archive": 2 * GIB}, + ) + self.assertTrue(any("restore required" in flag for flag in row["flags"])) + + def test_no_datapoints_records_unknown_size(self): + clients = self._clients( + bucket_locations={"fresh-bucket": REGION}, + metrics=[self._size_metric("fresh-bucket", "StandardStorage")], + metric_values={}, + ) + entry = EGRESS_RESOURCE_REGISTRY[S3_CODE] + + rows = _collect_s3_buckets(_mock_session(clients), REGION, S3_CODE, entry) + + self.assertIsNone(rows[0]["size_bytes"]) + self.assertTrue(rows[0]["size_unknown"]) + + def test_replication_configuration_adds_note(self): + clients = self._clients( + bucket_locations={"replicated": REGION}, + metrics=[self._size_metric("replicated", "StandardStorage")], + metric_values={"q0": float(GIB)}, + ) + clients["s3"].get_bucket_replication.side_effect = None + clients["s3"].get_bucket_replication.return_value = { + "ReplicationConfiguration": {"Rules": [{}]} + } + entry = EGRESS_RESOURCE_REGISTRY[S3_CODE] + + rows = _collect_s3_buckets(_mock_session(clients), REGION, S3_CODE, entry) + + self.assertTrue(any("replication" in note for note in rows[0]["notes"])) + + +class EbsCollectorTests(unittest.TestCase): + def test_volume_size_is_allocated_upper_bound(self): + ec2_client = MagicMock() + _mock_paginator( + ec2_client, + [ + { + "Volumes": [ + { + "VolumeId": "vol-1", + "Size": 100, + "Tags": [{"Key": "Name", "Value": "data-disk"}], + } + ] + } + ], + ) + entry = EGRESS_RESOURCE_REGISTRY[VOLUME_CODE] + + rows = _collect_ebs_volumes( + _mock_session({"ec2": ec2_client}), REGION, VOLUME_CODE, entry + ) + + self.assertEqual(rows[0]["name"], "data-disk") + self.assertEqual(rows[0]["size_bytes"], 100 * GIB) + self.assertIn("allocated (upper bound)", rows[0]["flags"]) + + def test_snapshots_are_owned_only_and_carry_shared_block_note(self): + ec2_client = MagicMock() + paginator = _mock_paginator( + ec2_client, + [{"Snapshots": [{"SnapshotId": "snap-1", "VolumeSize": 50}]}], + ) + entry = EGRESS_RESOURCE_REGISTRY[SNAPSHOT_CODE] + + rows = _collect_ebs_snapshots( + _mock_session({"ec2": ec2_client}), REGION, SNAPSHOT_CODE, entry + ) + + paginator.paginate.assert_called_once_with(OwnerIds=["self"]) + self.assertEqual(rows[0]["size_bytes"], 50 * GIB) + self.assertTrue(any("shares blocks" in note for note in rows[0]["notes"])) + + +class RdsCollectorTests(unittest.TestCase): + def _clients(self, instances, metric_results): + rds_client = MagicMock() + _mock_paginator(rds_client, [{"DBInstances": instances}]) + cloudwatch = MagicMock() + cloudwatch.get_metric_data.return_value = {"MetricDataResults": metric_results} + return {"rds": rds_client, "cloudwatch": cloudwatch} + + def test_used_space_computed_from_free_storage_space(self): + clients = self._clients( + instances=[ + { + "DBInstanceIdentifier": "appdb", + "Engine": "postgres", + "AllocatedStorage": 100, + } + ], + metric_results=[{"Id": "q0", "Values": [float(40 * GIB)]}], + ) + entry = EGRESS_RESOURCE_REGISTRY[RDS_CODE] + + rows = _collect_rds_instances(_mock_session(clients), REGION, RDS_CODE, entry) + + self.assertEqual(rows[0]["size_bytes"], 60 * GIB) + self.assertTrue(any("allocated" in note for note in rows[0]["notes"])) + self.assertNotIn("allocated (upper bound)", rows[0]["flags"]) + + def test_missing_metric_falls_back_to_allocated_upper_bound(self): + clients = self._clients( + instances=[ + { + "DBInstanceIdentifier": "appdb", + "Engine": "mysql", + "AllocatedStorage": 100, + } + ], + metric_results=[{"Id": "q0", "Values": []}], + ) + entry = EGRESS_RESOURCE_REGISTRY[RDS_CODE] + + rows = _collect_rds_instances(_mock_session(clients), REGION, RDS_CODE, entry) + + self.assertEqual(rows[0]["size_bytes"], 100 * GIB) + self.assertIn("allocated (upper bound)", rows[0]["flags"]) + + def test_aurora_instances_are_flagged_not_sized(self): + clients = self._clients( + instances=[ + { + "DBInstanceIdentifier": "aurora-1", + "Engine": "aurora-postgresql", + "AllocatedStorage": 1, + } + ], + metric_results=[], + ) + entry = EGRESS_RESOURCE_REGISTRY[RDS_CODE] + + rows = _collect_rds_instances(_mock_session(clients), REGION, RDS_CODE, entry) + + self.assertIsNone(rows[0]["size_bytes"]) + self.assertTrue(any("Aurora" in flag for flag in rows[0]["flags"])) + clients["cloudwatch"].get_metric_data.assert_not_called() + + +class DynamoDbCollectorTests(unittest.TestCase): + def test_table_and_index_sizes_are_summed(self): + dynamodb_client = MagicMock() + _mock_paginator(dynamodb_client, [{"TableNames": ["orders"]}]) + dynamodb_client.describe_table.return_value = { + "Table": { + "TableArn": "arn:aws:dynamodb:eu-central-1:1:table/orders", + "TableSizeBytes": 1000, + "GlobalSecondaryIndexes": [ + {"IndexSizeBytes": 300}, + {"IndexSizeBytes": 200}, + ], + } + } + entry = EGRESS_RESOURCE_REGISTRY[DYNAMODB_CODE] + + rows = _collect_dynamodb_tables( + _mock_session({"dynamodb": dynamodb_client}), REGION, DYNAMODB_CODE, entry + ) + + self.assertEqual(rows[0]["size_bytes"], 1500) + + 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] + + rows = _collect_dynamodb_tables( + _mock_session({"dynamodb": dynamodb_client}), REGION, DYNAMODB_CODE, entry + ) + + self.assertEqual(rows[0]["size_bytes"], 0) + self.assertFalse(rows[0]["size_unknown"]) + + +class BackupVaultCollectorTests(unittest.TestCase): + def test_vault_is_flagged_not_sized_with_recovery_point_note(self): + backup_client = MagicMock() + _mock_paginator( + backup_client, + [ + { + "BackupVaultList": [ + {"BackupVaultName": "prod-vault", "NumberOfRecoveryPoints": 12} + ] + } + ], + ) + entry = EGRESS_RESOURCE_REGISTRY[BACKUP_CODE] + + rows = _collect_backup_vaults( + _mock_session({"backup": backup_client}), REGION, BACKUP_CODE, entry + ) + + self.assertIsNone(rows[0]["size_bytes"]) + self.assertFalse(rows[0]["size_unknown"]) + self.assertIn("backup vault – not sized", rows[0]["flags"]) + self.assertTrue(any("12 recovery points" in note for note in rows[0]["notes"])) + + +class CollectAwsEgressTests(unittest.TestCase): + _PROVIDER_DETAILS = { + "accessKey": "AKIAIOSFODNN7EXAMPLE", + "secretKey": "secret", + "region": REGION, + } + + def _empty_clients(self): + s3_client = MagicMock() + s3_client.list_buckets.return_value = {"Buckets": []} + clients = {"s3": s3_client, "cloudwatch": MagicMock()} + for service, result_key in ( + ("ec2", "Volumes"), + ("rds", "DBInstances"), + ("dynamodb", "TableNames"), + ("backup", "BackupVaultList"), + ): + client = MagicMock() + _mock_paginator(client, [{result_key: []}]) + clients[service] = client + return clients + + @patch("core.utils_egress_aws.boto3") + def test_returns_rows_and_archive_tiers(self, mock_boto3): + clients = self._empty_clients() + mock_boto3.Session.return_value = _mock_session(clients) + + rows, archive_tiers = collect_aws_egress(self._PROVIDER_DETAILS) + + self.assertEqual(rows, []) + self.assertEqual(archive_tiers, {"Archive", "Glacier", "Deep Archive"}) + mock_boto3.Session.assert_called_once_with( + aws_access_key_id="AKIAIOSFODNN7EXAMPLE", + aws_secret_access_key="secret", + aws_session_token=None, + region_name=REGION, + ) + + @patch("core.utils_egress_aws.boto3") + def test_one_failing_service_does_not_fail_the_stage(self, mock_boto3): + clients = self._empty_clients() + clients["s3"].list_buckets.side_effect = _client_error( + "AccessDenied", "ListBuckets" + ) + volume_paginator = MagicMock() + volume_paginator.paginate.return_value = [ + {"Volumes": [{"VolumeId": "vol-1", "Size": 10}]} + ] + clients["ec2"].get_paginator.return_value = volume_paginator + mock_boto3.Session.return_value = _mock_session(clients) + + rows, _ = collect_aws_egress(self._PROVIDER_DETAILS) + + # Volumes and snapshots share the ec2 paginator mock here; the point + # is that the S3 failure is contained and other rows still arrive. + self.assertTrue(any(row["id"] == "vol-1" for row in rows)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_utils_egress_azure.py b/tests/test_utils_egress_azure.py new file mode 100644 index 0000000..0a9ca76 --- /dev/null +++ b/tests/test_utils_egress_azure.py @@ -0,0 +1,329 @@ +# tests/test_utils_egress_azure.py +import unittest +from unittest.mock import MagicMock, patch + +import requests + +from core.utils_egress import GIB, format_bytes +from core.utils_egress_azure import ( + build_egress_inventory, + collect_azure_egress, + fetch_monitor_metrics, + filter_data_bearing_resources, +) + + +def _mock_resource(resource_type, name, resource_id, sku_name=None): + resource = MagicMock() + resource.type = resource_type + resource.name = name + resource.id = resource_id + if sku_name: + resource.sku = MagicMock() + resource.sku.name = sku_name + else: + resource.sku = None + return resource + + +def _mock_credential(): + credential = MagicMock() + credential.get_token.return_value = MagicMock(token="fake-token") + return credential + + +class RegistryFilteringTests(unittest.TestCase): + def test_picks_data_bearing_types_out_of_mixed_list(self): + resources = [ + _mock_resource("Microsoft.Storage/storageAccounts", "sa1", "/sa1"), + _mock_resource("Microsoft.Network/virtualNetworks", "vnet1", "/vnet1"), + _mock_resource("Microsoft.Compute/disks", "disk1", "/disk1"), + _mock_resource("Microsoft.Compute/virtualMachines", "vm1", "/vm1"), + _mock_resource("Microsoft.RecoveryServices/vaults", "vault1", "/vault1"), + ] + + matched = filter_data_bearing_resources(resources) + matched_names = [resource.name for resource, _ in matched] + + self.assertEqual(matched_names, ["sa1", "disk1", "vault1"]) + + def test_matching_is_case_insensitive(self): + resources = [ + _mock_resource("MICROSOFT.STORAGE/StorageAccounts", "sa1", "/sa1"), + ] + + matched = filter_data_bearing_resources(resources) + + self.assertEqual(len(matched), 1) + self.assertEqual(matched[0][1]["strategy"], "storage_account_metrics") + + def test_returns_empty_for_no_data_bearing_resources(self): + resources = [ + _mock_resource("Microsoft.Network/networkInterfaces", "nic1", "/nic1"), + ] + + self.assertEqual(filter_data_bearing_resources(resources), []) + + +class FetchMonitorMetricsTests(unittest.TestCase): + @patch("core.utils_egress_azure.requests.get") + def test_parses_normal_response_using_latest_datapoint(self, mock_get): + mock_response = MagicMock() + mock_response.json.return_value = { + "value": [ + { + "name": {"value": "UsedCapacity"}, + "timeseries": [ + { + "metadatavalues": [], + "data": [ + {"average": 50.0}, + {"average": 123.0}, + {"average": None}, + ], + } + ], + } + ] + } + mock_get.return_value = mock_response + + result = fetch_monitor_metrics(_mock_credential(), "/sa1", ["UsedCapacity"]) + + self.assertEqual( + result, {"UsedCapacity": [{"dimension": None, "value": 123.0}]} + ) + + @patch("core.utils_egress_azure.requests.get") + def test_parses_dimension_split_response(self, mock_get): + mock_response = MagicMock() + mock_response.json.return_value = { + "value": [ + { + "name": {"value": "BlobCapacity"}, + "timeseries": [ + { + "metadatavalues": [ + {"name": {"value": "tier"}, "value": "Hot"} + ], + "data": [{"average": 10.0}], + }, + { + "metadatavalues": [ + {"name": {"value": "tier"}, "value": "Archive"} + ], + "data": [{"average": 20.0}], + }, + ], + } + ] + } + mock_get.return_value = mock_response + + result = fetch_monitor_metrics( + _mock_credential(), + "/sa1/blobServices/default", + ["BlobCapacity"], + dimension="Tier", + ) + + self.assertEqual( + result, + { + "BlobCapacity": [ + {"dimension": "Hot", "value": 10.0}, + {"dimension": "Archive", "value": 20.0}, + ] + }, + ) + params = mock_get.call_args.kwargs["params"] + self.assertEqual(params["$filter"], "Tier eq '*'") + + @patch("core.utils_egress_azure.requests.get") + def test_handles_empty_series_without_treating_as_zero(self, mock_get): + mock_response = MagicMock() + mock_response.json.return_value = { + "value": [{"name": {"value": "storage"}, "timeseries": []}] + } + mock_get.return_value = mock_response + + result = fetch_monitor_metrics(_mock_credential(), "/db1", ["storage"]) + + self.assertEqual(result, {"storage": []}) + + @patch("core.utils_egress_azure.requests.get") + def test_handles_missing_datapoints_without_treating_as_zero(self, mock_get): + mock_response = MagicMock() + mock_response.json.return_value = { + "value": [ + { + "name": {"value": "storage"}, + "timeseries": [{"metadatavalues": [], "data": [{"average": None}]}], + } + ] + } + mock_get.return_value = mock_response + + result = fetch_monitor_metrics(_mock_credential(), "/db1", ["storage"]) + + self.assertEqual(result, {"storage": []}) + + @patch("core.utils_egress_azure.requests.get") + def test_returns_none_on_http_error(self, mock_get): + mock_response = MagicMock() + mock_response.raise_for_status.side_effect = requests.HTTPError("403") + mock_get.return_value = mock_response + + result = fetch_monitor_metrics(_mock_credential(), "/sa1", ["UsedCapacity"]) + + self.assertIsNone(result) + + @patch("core.utils_egress_azure.requests.get") + def test_returns_none_on_connection_error(self, mock_get): + mock_get.side_effect = requests.ConnectionError("network down") + + result = fetch_monitor_metrics(_mock_credential(), "/sa1", ["UsedCapacity"]) + + self.assertIsNone(result) + + +class StorageAccountTierSplitTests(unittest.TestCase): + @patch("core.utils_egress_azure.fetch_monitor_metrics") + def test_hot_cool_archive_split_and_archive_flag(self, mock_fetch): + mock_fetch.side_effect = [ + {"UsedCapacity": [{"dimension": None, "value": float(60 * GIB)}]}, + { + "BlobCapacity": [ + {"dimension": "Hot", "value": float(30 * GIB)}, + {"dimension": "Cool", "value": float(20 * GIB)}, + {"dimension": "Archive", "value": float(10 * GIB)}, + ] + }, + ] + resource = _mock_resource( + "Microsoft.Storage/storageAccounts", "sa1", "/sa1", sku_name="Standard_LRS" + ) + + rows, findings = build_egress_inventory(MagicMock(), MagicMock(), [resource]) + + self.assertEqual(len(rows), 1) + row = rows[0] + self.assertEqual(row["size_bytes"], 60 * GIB) + self.assertEqual( + row["tier_bytes"], + {"Hot": 30 * GIB, "Cool": 20 * GIB, "Archive": 10 * GIB}, + ) + self.assertTrue(any("Archive" in flag for flag in row["flags"])) + self.assertEqual(findings, []) + + @patch("core.utils_egress_azure.fetch_monitor_metrics") + def test_geo_replicated_sku_adds_informational_finding(self, mock_fetch): + mock_fetch.side_effect = [ + {"UsedCapacity": [{"dimension": None, "value": float(GIB)}]}, + {"BlobCapacity": []}, + ] + resource = _mock_resource( + "Microsoft.Storage/storageAccounts", + "sa1", + "/sa1", + sku_name="Standard_RAGRS", + ) + + rows, findings = build_egress_inventory(MagicMock(), MagicMock(), [resource]) + + self.assertTrue(any("geo-replicated" in note for note in rows[0]["notes"])) + info_findings = [f for f in findings if f["severity"] == "info"] + self.assertEqual(len(info_findings), 1) + self.assertIn("not additional data to egress", info_findings[0]["message"]) + + @patch("core.utils_egress_azure.fetch_monitor_metrics", return_value=None) + def test_no_datapoints_records_unknown_size(self, mock_fetch): + resource = _mock_resource("Microsoft.Storage/storageAccounts", "sa1", "/sa1") + + rows, findings = build_egress_inventory(MagicMock(), MagicMock(), [resource]) + + self.assertIsNone(rows[0]["size_bytes"]) + self.assertTrue(rows[0]["size_unknown"]) + self.assertEqual(format_bytes(rows[0]["size_bytes"]), "n/a") + self.assertTrue(any("size unknown" in f["message"] for f in findings)) + + +class AllocatedSizeTests(unittest.TestCase): + def test_disk_size_read_from_property_and_flagged_as_upper_bound(self): + resource = _mock_resource("Microsoft.Compute/disks", "disk1", "/disk1") + resource_client = MagicMock() + full_resource = MagicMock() + full_resource.properties = {"diskSizeGB": 128} + resource_client.resources.get_by_id.return_value = full_resource + + rows, findings = build_egress_inventory( + MagicMock(), resource_client, [resource] + ) + + self.assertEqual(rows[0]["size_bytes"], 128 * GIB) + self.assertIn("allocated (upper bound)", rows[0]["flags"]) + self.assertEqual(findings, []) + + def test_collector_error_is_contained_as_finding(self): + resource = _mock_resource("Microsoft.Compute/disks", "disk1", "/disk1") + resource_client = MagicMock() + resource_client.resources.get_by_id.side_effect = RuntimeError("api error") + + rows, findings = build_egress_inventory( + MagicMock(), resource_client, [resource] + ) + + self.assertIsNone(rows[0]["size_bytes"]) + self.assertIn("size unavailable", rows[0]["flags"]) + self.assertTrue(any("size lookup failed" in f["message"] for f in findings)) + + +class VaultDetectionTests(unittest.TestCase): + @patch("core.utils_egress_azure.fetch_monitor_metrics") + def test_vault_produces_warning_finding_and_no_size(self, mock_fetch): + resource = _mock_resource("Microsoft.RecoveryServices/vaults", "vault1", "/v1") + resource_client = MagicMock() + + rows, findings = build_egress_inventory( + MagicMock(), resource_client, [resource] + ) + + self.assertIsNone(rows[0]["size_bytes"]) + self.assertFalse(rows[0]["size_unknown"]) + warnings = [f for f in findings if f["severity"] == "warning"] + self.assertEqual(len(warnings), 1) + self.assertIn("cannot be exported from Azure", warnings[0]["message"]) + mock_fetch.assert_not_called() + resource_client.resources.get_by_id.assert_not_called() + + +class CollectAzureEgressTests(unittest.TestCase): + @patch("core.utils_egress_azure.fetch_monitor_metrics") + @patch("core.utils_egress_azure.ResourceManagementClient") + def test_returns_rows_and_archive_tiers(self, mock_rmc_cls, mock_fetch): + mock_fetch.side_effect = [ + {"UsedCapacity": [{"dimension": None, "value": float(10 * GIB)}]}, + {"BlobCapacity": []}, + ] + resource = _mock_resource("Microsoft.Storage/storageAccounts", "sa1", "/sa1") + mock_client = MagicMock() + mock_client.resources.list_by_resource_group.return_value = [resource] + mock_rmc_cls.return_value = mock_client + + rows, archive_tiers = collect_azure_egress( + { + "credential": _mock_credential(), + "subscriptionId": "sub-id", + "resourceGroupName": "rg", + } + ) + + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["name"], "sa1") + self.assertEqual(rows[0]["size_bytes"], 10 * GIB) + self.assertNotIn("findings", rows[0]) + self.assertEqual(archive_tiers, {"Archive"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_utils_report_egress.py b/tests/test_utils_report_egress.py new file mode 100644 index 0000000..27f0e1a --- /dev/null +++ b/tests/test_utils_report_egress.py @@ -0,0 +1,861 @@ +# tests/test_utils_report_egress.py +import json +import os +import sqlite3 +import tempfile +import unittest +from unittest.mock import patch + +from reportlab.graphics.shapes import Drawing +from reportlab.platypus import PageBreak, Paragraph, Table + +from tests.report_fixtures import stage_report_assets + +from core.utils_egress import GIB +from core.utils_report_egress import ( + _build_allocation, + _build_coverage_section, + _build_data_landscape_section, + _build_estimated_costs_section, + _build_pricing_basis_section, + _build_type_groups, + _resolve_icon, + build_fee_estimate, + calculate_tiered_cost, + format_fee, + free_tier_limit_display, + generate_egress_html_report, + generate_egress_pdf_report, + load_pricing, +) + + +def _price_row( + csp, component, tier_from, tier_to, price, unit="GiB", zone="zone1", valid_to=None +): + return { + "id": 0, + "csp": csp, + "component": component, + "zone": zone, + "tier_from": tier_from, + "tier_to": tier_to, + "price_per_unit": price, + "currency": "USD", + "unit": unit, + "valid_from": "2026-07-01", + "valid_to": valid_to, + } + + +AZURE_INTERNET_TIERS = [ + _price_row(1, "internet_egress", 0, 100, 0.0), + _price_row(1, "internet_egress", 100, 10240, 0.087), + _price_row(1, "internet_egress", 10240, 51200, 0.083), + _price_row(1, "internet_egress", 51200, 153600, 0.07), + _price_row(1, "internet_egress", 153600, None, 0.05), +] + +SIMPLE_PRICING = [ + _price_row(1, "internet_egress", 0, 100, 0.0), + _price_row(1, "internet_egress", 100, None, 0.1), + _price_row(1, "archive_retrieval", 0, None, 0.02), +] + +RENDER_PRICING = AZURE_INTERNET_TIERS + [ + _price_row(1, "archive_retrieval", 0, None, 0.02), +] + + +def _fake_load_data(table_name, db_path=None): + if table_name == "egresspricing": + return list(RENDER_PRICING) + return [] + + +def _row(name, label, category, size_bytes, resource_type="some/type", **overrides): + row = { + "id": f"/{name}", + "name": name, + "type": resource_type, + "label": label, + "category": category, + "size_bytes": size_bytes, + "size_unknown": False, + "tier_bytes": None, + "flags": [], + "notes": [], + } + row.update(overrides) + return row + + +def _totals(rows, unknown_count=0, archive_tier_bytes=0): + return { + "known_size_bytes": sum( + row["size_bytes"] for row in rows if row["size_bytes"] is not None + ), + "archive_tier_bytes": archive_tier_bytes, + "resources_discovered": len(rows), + "resources_with_unknown_size": unknown_count, + } + + +def _payload(rows, unknown_count=0, archive_tier_bytes=0): + return { + "meta": { + "name": "Exit Assessment Test", + "cloud_service_provider": 1, + "exit_strategy": 3, + "assessment_type": 1, + "timestamp": "2026-07-12 18:45:08 UTC", + }, + "data": { + "resources": rows, + "totals": _totals(rows, unknown_count, archive_tier_bytes), + }, + } + + +class LoadPricingTests(unittest.TestCase): + @staticmethod + def _make_report_db(report_path, rows): + db_dir = os.path.join(report_path, "data") + os.makedirs(db_dir, exist_ok=True) + conn = sqlite3.connect(os.path.join(db_dir, "assessment.db")) + conn.execute( + "CREATE TABLE egresspricing (id INTEGER PRIMARY KEY, csp INTEGER, " + "component TEXT, zone TEXT, tier_from REAL, tier_to REAL, " + "price_per_unit REAL, currency TEXT, unit TEXT, valid_from TEXT, " + "valid_to TEXT)" + ) + conn.executemany( + "INSERT INTO egresspricing (csp, component, zone, tier_from, tier_to, " + "price_per_unit, currency, unit, valid_from, valid_to) " + "VALUES (:csp, :component, :zone, :tier_from, :tier_to, " + ":price_per_unit, :currency, :unit, :valid_from, :valid_to)", + [{key: row[key] for key in row if key != "id"} for row in rows], + ) + conn.commit() + conn.close() + + def test_loads_only_active_rows_from_run_database(self): + rows = SIMPLE_PRICING + [ + _price_row(1, "internet_egress", 100, None, 0.2, valid_to="2026-06-30"), + ] + with tempfile.TemporaryDirectory() as tmp_dir: + self._make_report_db(tmp_dir, rows) + loaded = load_pricing(tmp_dir) + + self.assertEqual(len(loaded), len(SIMPLE_PRICING)) + self.assertTrue(all(not row["valid_to"] for row in loaded)) + + def test_falls_back_to_master_dataset_when_table_missing(self): + with patch( + "core.utils_report_egress.load_data", + side_effect=[sqlite3.OperationalError("no such table"), SIMPLE_PRICING], + ) as mock_load: + loaded = load_pricing("/tmp/report") + + self.assertEqual(len(loaded), len(SIMPLE_PRICING)) + self.assertEqual(mock_load.call_count, 2) + # The fallback call uses the master dataset default path. + self.assertEqual(mock_load.call_args_list[1].kwargs, {}) + + @unittest.skipUnless( + os.path.exists("datasets/data.db"), "master dataset not downloaded" + ) + def test_shipped_dataset_has_contiguous_zone1_ladders(self): + from itertools import groupby + + from core.utils_db import load_data as db_load + + rows = [ + row + for row in db_load("egresspricing", db_path="datasets/data.db") + if not row["valid_to"] + ] + rows.sort(key=lambda r: (r["csp"], r["component"], r["zone"], r["tier_from"])) + seen = set() + for key, group in groupby( + rows, key=lambda r: (r["csp"], r["component"], r["zone"]) + ): + group = list(group) + seen.add(key) + self.assertEqual(group[0]["tier_from"], 0, key) + for lower, upper in zip(group, group[1:]): + self.assertEqual(lower["tier_to"], upper["tier_from"], key) + self.assertIsNone(group[-1]["tier_to"], key) + for csp in (1, 2): + self.assertIn((csp, "internet_egress", "zone1"), seen) + + +class TieredCostTests(unittest.TestCase): + def test_free_tier_under_100_gib(self): + self.assertEqual(calculate_tiered_cost(50, AZURE_INTERNET_TIERS), 0.0) + self.assertEqual(calculate_tiered_cost(100, AZURE_INTERNET_TIERS), 0.0) + + def test_multi_tib_crosses_tiers(self): + # 20 TiB: 100 GiB free, 10,140 GiB @ 0.087, 10,240 GiB @ 0.083 + expected = 10140 * 0.087 + 10240 * 0.083 + self.assertAlmostEqual( + calculate_tiered_cost(20 * 1024, AZURE_INTERNET_TIERS), expected + ) + + def test_unbounded_top_tier(self): + expected = 10140 * 0.087 + 40960 * 0.083 + 102400 * 0.07 + 100 * 0.05 + self.assertAlmostEqual( + calculate_tiered_cost(153700, AZURE_INTERNET_TIERS), expected + ) + + +class BuildFeeEstimateTests(unittest.TestCase): + def test_prorates_internet_fee_and_adds_retrieval_per_resource(self): + rows = [ + _row( + "sa1", + "Storage Account", + "object", + 600 * GIB, + tier_bytes={"Hot": 400 * GIB, "Archive": 200 * GIB}, + ), + _row("disk1", "Managed Disk", "block", 400 * GIB), + _row("db1", "SQL Database", "database", None, size_unknown=True), + ] + totals = _totals(rows, unknown_count=1, archive_tier_bytes=200 * GIB) + + total_fee, fees_by_id = build_fee_estimate(rows, totals, 1, SIMPLE_PRICING) + + # internet: (1000 - 100) * 0.1 = 90; retrieval: 200 * 0.02 = 4 + self.assertAlmostEqual(total_fee, 94.0) + self.assertAlmostEqual(fees_by_id["/sa1"], 90 * 0.6 + 4.0) + self.assertAlmostEqual(fees_by_id["/disk1"], 90 * 0.4) + self.assertIsNone(fees_by_id["/db1"]) + + def test_zero_bytes_produce_zero_fee(self): + rows = [_row("vault1", "Backup Vault", "backup", None)] + totals = _totals(rows) + + total_fee, fees_by_id = build_fee_estimate(rows, totals, 1, SIMPLE_PRICING) + + self.assertEqual(total_fee, 0.0) + self.assertIsNone(fees_by_id["/vault1"]) + + def test_format_fee(self): + self.assertEqual(format_fee(None), "n/a") + self.assertEqual(format_fee(0), "$0.00") + self.assertEqual(format_fee(1732.1), "$1,732.10") + + def test_free_tier_limit_derived_from_pricing(self): + self.assertEqual(free_tier_limit_display(SIMPLE_PRICING, 1), "100 GiB") + self.assertIsNone(free_tier_limit_display(SIMPLE_PRICING, 2)) + + +class PricingUnitAndZoneTests(unittest.TestCase): + def test_units_follow_the_pricing_rows(self): + prices = [ + _price_row(1, "internet_egress", 0, 100, 0.0, unit="GB"), + _price_row(1, "internet_egress", 100, None, 0.1, unit="GB"), + ] + rows = [_row("sa1", "Storage Account", "object", 1000 * 10**9)] + + total_fee, _ = build_fee_estimate(rows, _totals(rows), 1, prices) + + # Decimal-GB ladder: (1000 GB - 100 GB free) * 0.1 + self.assertAlmostEqual(total_fee, 90.0) + + def test_other_zones_are_ignored(self): + prices = SIMPLE_PRICING + [ + _price_row(1, "internet_egress", 100, None, 9.9, zone="zone3"), + ] + rows = [_row("sa1", "Storage Account", "object", 200 * GIB)] + + total_fee, _ = build_fee_estimate(rows, _totals(rows), 1, prices) + + self.assertAlmostEqual(total_fee, 10.0) + + def test_missing_provider_pricing_raises(self): + rows = [_row("sa1", "Storage Account", "object", 200 * GIB)] + + with self.assertRaises(ValueError): + build_fee_estimate(rows, _totals(rows), 2, SIMPLE_PRICING) + + +class ResolveIconTests(unittest.TestCase): + def test_exact_match_from_resourcetype_table(self): + lookup = {"microsoft.storage/storageaccounts": "/icons/azure/storage/sa.png"} + + icon = _resolve_icon("Microsoft.Storage/storageAccounts", lookup) + + self.assertEqual(icon, "/icons/azure/storage/sa.png") + + def test_trims_path_segments_until_match(self): + lookup = {"microsoft.sql": "/icons/azure/databases/sql.png"} + + icon = _resolve_icon("Microsoft.Sql/servers/databases", lookup) + + self.assertEqual(icon, "/icons/azure/databases/sql.png") + + def test_known_gap_uses_hardcoded_fallback(self): + icon = _resolve_icon("AWS.ec2.describe_volumes.Volumes", {}) + + self.assertEqual(icon, "/icons/aws/Storage/Elastic-Block-Store.png") + + def test_unknown_type_uses_default_icon(self): + icon = _resolve_icon("Vendor.Unknown/things", {}) + + self.assertEqual(icon, "/icons/misc/no_image.png") + + +class AllocationSeriesTests(unittest.TestCase): + def test_groups_known_bytes_by_category_in_fixed_order(self): + rows = [ + _row("disk1", "Managed Disk", "block", 40 * GIB), + _row("sa1", "Storage Account", "object", 60 * GIB), + _row("db1", "SQL Database", "database", 10 * GIB), + _row("vault1", "Recovery Services Vault", "backup", None), + _row("db2", "SQL Database", "database", None, size_unknown=True), + ] + + labels, values, colors = _build_allocation(rows) + + self.assertEqual( + labels, + ["Object Storage", "Block (allocated)", "Databases"], + ) + self.assertEqual(values, [60 * GIB, 40 * GIB, 10 * GIB]) + self.assertEqual(len(colors), 3) + + def test_zero_categories_are_omitted(self): + rows = [_row("sa1", "Storage Account", "object", 5 * GIB)] + + labels, values, _ = _build_allocation(rows) + + self.assertEqual(labels, ["Object Storage"]) + self.assertEqual(values, [5 * GIB]) + + +class TypeGroupTests(unittest.TestCase): + def test_groups_by_label_and_sorts_by_fee(self): + rows = [ + _row("disk1", "Managed Disk", "block", 100 * GIB), + _row("disk2", "Managed Disk", "block", 50 * GIB), + _row("sa1", "Storage Account", "object", 400 * GIB), + _row("vault1", "Recovery Services Vault", "backup", None), + ] + fees_by_id = {"/disk1": 10.0, "/disk2": 5.0, "/sa1": 40.0, "/vault1": None} + + groups = _build_type_groups(rows, fees_by_id, {}) + + self.assertEqual( + [group["label"] for group in groups], + ["Storage Account", "Managed Disk", "Recovery Services Vault"], + ) + disk_group = groups[1] + self.assertEqual(disk_group["size_display"], "150.0 GiB") + self.assertEqual(disk_group["fee_display"], "$15.00") + self.assertEqual(len(disk_group["resources"]), 2) + self.assertEqual(disk_group["resources"][0]["fee_display"], "$10.00") + vault_group = groups[2] + self.assertEqual(vault_group["size_display"], "n/a") + self.assertEqual(vault_group["fee_display"], "n/a") + self.assertIsNone(vault_group["resources"][0]["fee_display"]) + + def test_flags_and_notes_become_resource_detail(self): + rows = [ + _row( + "disk1", + "Managed Disk", + "block", + 100 * GIB, + flags=["allocated (upper bound)"], + notes=["shared with vm-1"], + ), + ] + + groups = _build_type_groups(rows, {"/disk1": 1.0}, {}) + + self.assertEqual( + groups[0]["resources"][0]["detail"], + "allocated (upper bound); shared with vm-1", + ) + + +class GenerateEgressHtmlReportTests(unittest.TestCase): + def _generate(self, payload): + with tempfile.TemporaryDirectory() as tmp_dir: + json_path = os.path.join(tmp_dir, "egress_estimate.json") + with open(json_path, "w", encoding="utf-8") as json_file: + json.dump(payload, json_file) + + with patch( + "core.utils_report_egress.load_data", + side_effect=_fake_load_data, + ): + result = generate_egress_html_report(tmp_dir, json_path) + + html = None + if result["success"]: + self.assertEqual( + result["html_path"], os.path.join(tmp_dir, "egress.html") + ) + with open(result["html_path"], encoding="utf-8") as html_file: + html = html_file.read() + return result, html + + def test_renders_report_with_fees_table_and_charts(self): + rows = [ + _row( + "proddata", + "Storage Account", + "object", + 2048 * GIB, + resource_type="Microsoft.Storage/storageAccounts", + ), + _row( + "vm-osdisk", + "Managed Disk", + "block", + 512 * GIB, + resource_type="Microsoft.Compute/disks", + flags=["allocated (upper bound)"], + ), + ] + result, html = self._generate(_payload(rows)) + + self.assertTrue(result["success"], result["logs"]) + self.assertIn("Exit Assessment Test", html) + normalized_html = " ".join(html.split()) + self.assertIn( + "The Data Landscape & Egress Estimation feature is currently in", + html, + ) + self.assertIn("mailto:beta@escapecloud.io", html) + # (2560 - 100) GiB * 0.087 = $214.02 total internet egress fee + self.assertIn("$214.02", html) + self.assertIn("Estimated Egress Fee", html) + self.assertIn("Estimated one-time charge to transfer the identified", html) + self.assertIn( + 'any resources whose size could not be measured. A "+" means', + normalized_html, + ) + self.assertIn("lower-bound estimate.", html) + self.assertIn("feesChart", html) + self.assertIn("allocationChart", html) + self.assertIn("proddata", html) + self.assertIn("category-object", html) + self.assertIn("allocated (upper bound)", html) + self.assertNotIn("Estimates only", html) + self.assertNotIn("At least", html) + + def test_unknown_sizes_add_plus_suffix_to_headlines(self): + rows = [ + _row("sa1", "Storage Account", "object", 2048 * GIB), + _row("db1", "SQL Database", "database", None, size_unknown=True), + ] + result, html = self._generate(_payload(rows, unknown_count=1)) + + self.assertTrue(result["success"], result["logs"]) + # 2048 GiB known: (2048 - 100) * 0.087 = $169.48, "+" marks the lower bound + self.assertIn("2.0 TiB+", html) + self.assertIn("$169.48+", html) + self.assertIn("Excludes 1 resource", html) + self.assertIn("n/a", html) + + def test_free_tier_environment_shows_included_in_free_tier_card(self): + rows = [_row("sa1", "Storage Account", "object", 5 * GIB)] + result, html = self._generate(_payload(rows)) + + self.assertTrue(result["success"], result["logs"]) + self.assertIn("Included in Free Tier", html) + # The fee headline stays visible above the card, as a clean "$0". + self.assertIn("Estimated Egress Fee", html) + self.assertIn('
$0
', html) + self.assertIn("5.0 GiB", html) + self.assertIn("100 GiB free tier", html) + self.assertNotIn("No fee estimate available.", html) + self.assertNotIn("feesChart", html) + + def test_empty_inventory_renders_empty_states(self): + result, html = self._generate(_payload([])) + + self.assertTrue(result["success"], result["logs"]) + self.assertIn("$0.00", html) + # Zero known bytes is the empty state, not the free-tier success card. + self.assertIn("No fee estimate available.", html) + self.assertNotIn("Included in Free Tier", html) + self.assertIn("No data volume available.", html) + self.assertIn( + "No data-bearing resources were discovered during the", + html, + ) + self.assertNotIn("allocationChart", html) + + def test_missing_json_returns_failure(self): + with tempfile.TemporaryDirectory() as tmp_dir: + result = generate_egress_html_report( + tmp_dir, os.path.join(tmp_dir, "missing.json") + ) + + self.assertFalse(result["success"]) + + +def _make_pdf_styles(): + from reportlab.lib.colors import HexColor + from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet + + styles = getSampleStyleSheet() + content_style = ParagraphStyle( + "ContentStyle", fontSize=10, leading=12, spaceAfter=10 + ) + styles["Heading1"].textColor = HexColor("#112726") + styles["Heading2"].textColor = HexColor("#112726") + return styles, content_style + + +class BuildEstimatedCostsSectionTests(unittest.TestCase): + def setUp(self): + self.styles, self.content_style = _make_pdf_styles() + self.type_groups = [ + { + "label": "Storage Account", + "fee": 180.0, + "fee_display": "$180.00", + }, + { + "label": "Managed Disk", + "fee": 34.02, + "fee_display": "$34.02", + }, + { + "label": "Backup Vault", + "fee": None, + "fee_display": "n/a", + }, + ] + + def _totals(self, known, unknown_count=0): + return { + "known_size_bytes": known, + "archive_tier_bytes": 0, + "resources_discovered": 1, + "resources_with_unknown_size": unknown_count, + } + + def _tables(self, content): + tables = [f for f in content if isinstance(f, Table)] + self.assertGreaterEqual(len(tables), 2) + return tables + + def test_normal_fee_shell(self): + content = _build_estimated_costs_section( + 214.02, + self._totals(2560 * GIB), + False, + "100 GiB", + self.type_groups, + self.styles, + self.content_style, + ) + + tables = self._tables(content) + self.assertEqual(tables[0]._cellvalues[1][0].text, "$214.02") + self.assertIsInstance(tables[1]._cellvalues[0][0], Drawing) + breakdown = tables[-1]._cellvalues + self.assertEqual(breakdown[1], ["Storage Account", "$180.00"]) + self.assertEqual(breakdown[2], ["Managed Disk", "$34.02"]) + self.assertEqual(breakdown[3], ["Backup Vault", "n/a (not sized)"]) + self.assertEqual(breakdown[-1][0].text, "Estimated Total Egress Fee") + self.assertEqual(breakdown[-1][1].text, "$214.02") + paragraphs = [f.text for f in content if isinstance(f, Paragraph)] + self.assertTrue(any("Estimated Costs" in text for text in paragraphs)) + self.assertTrue(any("Cost Breakdown" in text for text in paragraphs)) + + def test_unknowns_add_plus_suffix_to_fee(self): + content = _build_estimated_costs_section( + 100.0, + self._totals(2000 * GIB, unknown_count=2), + False, + "100 GiB", + self.type_groups, + self.styles, + self.content_style, + ) + + tables = self._tables(content) + self.assertEqual(tables[0]._cellvalues[1][0].text, "$100.00+") + self.assertIn("Excludes 2 resources", tables[0]._cellvalues[2][0].text) + paragraphs = [f.text for f in content if isinstance(f, Paragraph)] + self.assertTrue(any("Estimated Costs" in text for text in paragraphs)) + + def test_free_tier_variant(self): + content = _build_estimated_costs_section( + 0.0, + self._totals(5 * GIB), + True, + "100 GiB", + [{"label": "Storage Account", "fee": 0.0, "fee_display": "$0.00"}], + self.styles, + self.content_style, + ) + + tables = self._tables(content) + self.assertEqual(tables[0]._cellvalues[1][0].text, "$0") + self.assertEqual(len(tables), 2) + paragraphs = [f.text for f in content if isinstance(f, Paragraph)] + self.assertTrue(any("100 GiB free tier" in text for text in paragraphs)) + + +class BuildDataLandscapeSectionTests(unittest.TestCase): + def test_table_rows_and_total(self): + styles, content_style = _make_pdf_styles() + type_groups = [ + { + "label": "S3 Bucket", + "icon": "/icons/aws/Storage/Simple-Storage-Service.png", + "fee_display": "$200.00", + "size_display": "20.0 MiB", + }, + { + "label": "Backup Vault", + "icon": "/icons/aws/Storage/Backup.png", + "fee_display": "n/a", + "size_display": "n/a", + }, + ] + + with tempfile.TemporaryDirectory() as tmp_dir: + content = _build_data_landscape_section( + type_groups, + "20.0 MiB+", + tmp_dir, + styles, + content_style, + ) + + tables = [f for f in content if isinstance(f, Table)] + self.assertEqual(len(tables), 1) + cells = tables[0]._cellvalues + # Costs column deliberately absent — dollars live on the Estimated + # Costs page so each type is not shown twice in the report. + self.assertEqual(cells[0], ["#", "Resource Type", "", "Size"]) + self.assertEqual(cells[1][1], "S3 Bucket") + self.assertEqual(cells[1][3], "20.0 MiB") + self.assertEqual(cells[2][3], "n/a") + self.assertEqual(cells[-1][0], "Total") + self.assertEqual(cells[-1][3], "20.0 MiB+") + for row in cells: + for cell in row: + self.assertNotIn("$", str(cell)) + + +class BuildCoverageSectionTests(unittest.TestCase): + def setUp(self): + self.styles, self.content_style = _make_pdf_styles() + + def _cells(self, content): + tables = [f for f in content if isinstance(f, Table)] + self.assertEqual(len(tables), 1) + return tables[0]._cellvalues + + def test_aws_coverage_rows_and_alpha_note(self): + content = _build_coverage_section(2, self.styles, self.content_style) + + cells = self._cells(content) + self.assertEqual(cells[0], ["Name", "Notes"]) + names = [row[0] for row in cells[1:]] + self.assertIn("S3 Buckets", names) + self.assertIn("AWS Backup Vaults", names) + texts = [f.text for f in content if isinstance(f, Paragraph)] + self.assertTrue(any(text.startswith("NOTE:") for text in texts)) + self.assertTrue(any("alpha version" in text for text in texts)) + # Estimated Costs starts on a fresh page after the coverage block. + self.assertIsInstance(content[-1], PageBreak) + + def test_azure_coverage_rows(self): + content = _build_coverage_section(1, self.styles, self.content_style) + + names = [row[0] for row in self._cells(content)[1:]] + self.assertIn("Storage Accounts", names) + self.assertIn("Recovery Services Vaults", names) + + def test_unknown_provider_renders_nothing(self): + self.assertEqual( + _build_coverage_section(9, self.styles, self.content_style), [] + ) + + +class BuildPricingBasisSectionTests(unittest.TestCase): + def setUp(self): + self.styles, self.content_style = _make_pdf_styles() + + def _texts(self, content): + return [f.text for f in content if isinstance(f, Paragraph)] + + def _table(self, content): + tables = [f for f in content if isinstance(f, Table)] + self.assertEqual(len(tables), 1) + return tables[0]._cellvalues + + def test_renders_used_price_rows_with_free_tier(self): + content = _build_pricing_basis_section( + RENDER_PRICING, 1, self.styles, self.content_style + ) + + texts = self._texts(content) + self.assertTrue(any("Appendix – Egress Prices" in text for text in texts)) + self.assertTrue( + any( + "Microsoft Azure list prices (Zone 1 – US/Europe, public " + "internet routing):" in text + for text in texts + ) + ) + self.assertFalse(any("effective as of" in text for text in texts)) + cells = self._table(content) + self.assertEqual(cells[0], ["#", "Component", "Tier", "Price"]) + self.assertEqual(cells[1][1], "Internet Egress") + self.assertEqual(cells[1][2], "first 100 GiB / month") + self.assertEqual(cells[1][3], "Free") + self.assertEqual(cells[2][3], "$0.087 / GiB") + self.assertEqual(cells[-1][1], "Archive Retrieval") + self.assertEqual(cells[-1][2], "all volumes") + self.assertEqual(cells[-1][3], "$0.02 / GiB") + self.assertFalse(any("rehydration fees" in text for text in texts)) + self.assertTrue( + any( + "Source (Internet Egress): " + "https://azure.microsoft.com/pricing/details/bandwidth/" in text + for text in texts + ) + ) + self.assertTrue( + any( + "Source (Archive Retrieval): " + "https://azure.microsoft.com/pricing/details/storage/blobs/" in text + for text in texts + ) + ) + self.assertTrue( + any("List prices at the time of the assessment" in text for text in texts) + ) + + def test_missing_retrieval_rows_add_note(self): + content = _build_pricing_basis_section( + AZURE_INTERNET_TIERS, 1, self.styles, self.content_style + ) + + texts = self._texts(content) + self.assertTrue( + any("rehydration fees are not included" in text for text in texts) + ) + self.assertTrue( + any( + "Source: https://azure.microsoft.com/pricing/details/bandwidth/" in text + for text in texts + ) + ) + self.assertFalse(any("storage/blobs" in text for text in texts)) + + def test_unused_zone_component_and_provider_rows_are_excluded(self): + prices = RENDER_PRICING + [ + _price_row(1, "internet_egress", 0, None, 9.9, zone="zone3"), + _price_row(1, "internet_egress_transit_isp", 0, None, 9.9), + _price_row(2, "internet_egress", 0, None, 9.9), + ] + + content = _build_pricing_basis_section( + prices, 1, self.styles, self.content_style + ) + + cells = self._table(content) + self.assertEqual(len(cells) - 1, len(RENDER_PRICING)) + + def test_mixed_valid_from_adds_column_without_effective_date_claim(self): + retrieval = dict( + _price_row(1, "archive_retrieval", 0, None, 0.02), + valid_from="2026-07-14", + ) + prices = AZURE_INTERNET_TIERS + [retrieval] + + content = _build_pricing_basis_section( + prices, 1, self.styles, self.content_style + ) + + texts = self._texts(content) + self.assertFalse(any("effective as of" in text for text in texts)) + cells = self._table(content) + self.assertEqual(cells[0][-1], "Valid from") + self.assertEqual(cells[-1][-1], "2026-07-14") + + def test_empty_pricing_renders_nothing(self): + self.assertEqual( + _build_pricing_basis_section([], 1, self.styles, self.content_style), [] + ) + + +class GenerateEgressPdfReportTests(unittest.TestCase): + _PROVIDER_DETAILS = { + "tenantId": "tenant-id", + "subscriptionId": "sub-id", + "resourceGroupName": "my-rg", + } + + 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") + with open(json_path, "w", encoding="utf-8") as json_file: + json.dump(payload, json_file) + + with patch( + "core.utils_report_egress.load_data", + side_effect=_fake_load_data, + ): + result = generate_egress_pdf_report( + tmp_dir, json_path, self._PROVIDER_DETAILS + ) + + pdf_size = 0 + if result["success"]: + self.assertEqual( + result["pdf_path"], os.path.join(tmp_dir, "egress.pdf") + ) + pdf_size = os.path.getsize(result["pdf_path"]) + return result, pdf_size + + def test_generates_pdf_document(self): + rows = [ + _row("proddata", "Storage Account", "object", 2048 * GIB), + _row( + "vm-osdisk", + "Managed Disk", + "block", + 512 * GIB, + flags=["allocated (upper bound)"], + ), + _row("vault1", "Recovery Services Vault", "backup", None), + ] + result, pdf_size = self._generate(_payload(rows)) + + self.assertTrue(result["success"], result["logs"]) + self.assertGreater(pdf_size, 0) + + def test_free_tier_footprint_generates_pdf(self): + rows = [_row("sa1", "Storage Account", "object", 5 * GIB)] + result, pdf_size = self._generate(_payload(rows)) + + self.assertTrue(result["success"], result["logs"]) + self.assertGreater(pdf_size, 0) + + def test_missing_json_returns_failure(self): + with tempfile.TemporaryDirectory() as tmp_dir: + result = generate_egress_pdf_report( + tmp_dir, os.path.join(tmp_dir, "missing.json"), {} + ) + + self.assertFalse(result["success"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/utils/codes.py b/utils/codes.py index 0c402c9..c6572f0 100644 --- a/utils/codes.py +++ b/utils/codes.py @@ -9,3 +9,4 @@ COST_INVENTORY = 6 RISK_ASSESSMENT = 7 REPORT = 8 +EGRESS = 9