-
+
+
@@ -339,7 +381,7 @@
+
+
+
+
+
{% endif %}
+ {% if summary.costs_egress_note %}
+
-
-
-
- - The Data Landscape & Egress Estimation feature is currently in - alpha. Estimated values may be incomplete or inaccurate in some - environments. If you notice bugs or unexpected results, please - let us know at - beta@escapecloud.io -
-Assessment Type
{{ assessment_name }}
+
+
+
+ Document ID
+{{ document_id }}
+
+
+
+
+
+
+
+
+ {% endif %}
+
+ {{ introduction.heading }}
+{{ introduction.purpose_heading }}
+{{ introduction.purpose }}
+{{ introduction.scope_heading }}
+{{ introduction.scope }}
+{{ introduction.limitations_heading }}
+ {% for paragraph in introduction.limitations %} +{{ paragraph }}
+ {% endfor %} + {% if coverage %} +{{ coverage.heading }}
+{{ coverage.intro }}
+| {{ header }} | + {% endfor %} +|
|---|---|
| {{ name }} | +{{ note }} | +
-
+ Data Landscape
+{{ landscape.heading }}
+
{{ landscape.lead }}
+
+ {% for label, value, attention in landscape.stats %}
+
+
+
+ {% endfor %}
+ {{ label }}
+ {{ value }}
+ | # | Resource Type | Category | -Size | +{{ landscape.headers[2] }} | @@ -426,6 +468,124 @@
|---|
+ {% if landscape.not_sized %}
+ {{ landscape.not_sized }}
{% endif %} +{{ landscape.caption }}
+{{ landscape.source_note }}
+
+
+
+ {% if pricing %}
+
+
+
+
+
+
+ {% endif %}
+
+ {{ costs.heading }}
+{{ costs.lead }}
+
+ {% for label, value, attention in costs.stats %}
+
+ {% if costs.rows %}
+
+
+ {% endfor %}
+ {{ label }}
+ {{ value }}
+ | {{ costs.headers[0] }} | +{{ costs.headers[1] }} | +{{ costs.headers[2] }} | +{{ costs.headers[3] }} | +
|---|---|---|---|
| {{ row.label }} | +{{ row.size }} | +{{ row.share_text }} | +{{ row.fee }} | +
| {{ costs.total[0] }} | +{{ costs.total[1] }} | ++ | {{ costs.total[2] }} | +
{{ costs.caption }}
+
+
+ {% endif %}
+
+
+
+
+
+
+
+
+ {{ pricing.heading }}
+{{ pricing.intro }}
+| {{ header }} | + {% endfor %} +
|---|
| {{ cell }} | + {% endfor %} +
+ {% if pricing.units %}
+ {{ pricing.units }}
{% endif %} + {% if pricing.no_archive %}{{ pricing.no_archive }}
{% endif %} + {% for template, url in pricing.sources %} +{{ template.split("{link}")[0] }}{{ url }}{{ template.split("{link}")[1] }}
+ {% endfor %} +{{ pricing.note }}
+{{ pricing.glossary_heading }}
+-
+ {% for term, definition in pricing.glossary %}
+
- {{ term }} +
- {{ definition }} + {% endfor %} +
+
@@ -474,14 +634,10 @@
+
+
{{ next_steps.heading }}
+ {% for paragraph in next_steps.paragraphs %} ++ {% for part in paragraph %}{% if part is string %}{{ part }}{% else %}{{ part[0] }}{% endif %}{% endfor %} +
+ {% endfor %}{{ block.heading }}
if (value === null || value === undefined) { return "n/a"; } - const units = ["B", "KiB", "MiB", "GiB", "TiB"]; - let scaled = value; - let unitIndex = 0; - while (scaled >= 1024 && unitIndex < units.length - 1) { - scaled /= 1024; - unitIndex += 1; - } - return scaled.toFixed(1) + " " + units[unitIndex]; + return (value / 1024 ** 3).toLocaleString(undefined, { + minimumFractionDigits: 1, + maximumFractionDigits: 1, + }) + " GiB"; } diff --git a/cloudexit/assets/template/index.html b/cloudexit/assets/template/index.html index 2c787c1..0087909 100644 --- a/cloudexit/assets/template/index.html +++ b/cloudexit/assets/template/index.html @@ -294,6 +294,9 @@Costs Overview
{{ summary.costs_egress_note }}
+ {% endif %} diff --git a/cloudexit/cli/app.py b/cloudexit/cli/app.py index 8667119..69e218c 100644 --- a/cloudexit/cli/app.py +++ b/cloudexit/cli/app.py @@ -29,10 +29,7 @@ create_subscription_client, ) from cloudexit.egress.estimate import estimate_egress -from cloudexit.report.egress import ( - generate_egress_html_report, - generate_egress_pdf_report, -) +from cloudexit.report.egress import generate_egress_html_report from cloudexit.platform.client import ApiError, check_key from cloudexit.platform.run_output import build_run_json, exit_code_of from cloudexit.platform.status import host_catalogue_lines, run_status @@ -992,7 +989,6 @@ def run_assessment( egress_collected = False egress_json_path = None egress_html_path = None - egress_pdf_path = None if egress: console.print("Stage #5 – Egress Estimation", style="bold") @@ -1014,7 +1010,10 @@ def run_assessment( with console.status("Generating egress report...", spinner="dots"): egress_report_result = generate_egress_html_report( - report_path, egress_json_path + report_path, + egress_json_path, + config["providerDetails"], + sync_id, ) if egress_report_result["success"]: @@ -1027,22 +1026,6 @@ def run_assessment( 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...", @@ -1170,6 +1153,7 @@ def run_assessment( raw_data_path, upload=report_info(upload_result), sync_id=sync_id, + egress_json_path=egress_json_path, ) # Handle the result @@ -1181,6 +1165,20 @@ def run_assessment( ) sys.exit(codes.REPORT) + # The egress PDF is written with the report, after the upload, so it can + # show the run and the platform link. Its failure is an egress failure, + # not a report failure. + egress_pdf_path = report_status.get("reports", {}).get("EGRESS_PDF") + if egress_pdf_path: + print_step("Generating egress PDF...", status="ok") + elif egress_json_path: + print_step( + "Generating egress PDF...", + status="error", + logs=report_status.get("egress_pdf_logs", "no egress PDF was written"), + ) + egress_failed = True + # Output the report path after the separator if not (dry_run or non_interactive): console.print("-------------------------------------------") diff --git a/cloudexit/engine.py b/cloudexit/engine.py index 2c3efa6..f5da2a8 100644 --- a/cloudexit/engine.py +++ b/cloudexit/engine.py @@ -22,7 +22,12 @@ from cloudexit.providers.azure.sdk import azure_import from cloudexit.catalogue.db import connect, load_data from cloudexit.tfstate import build_tfstate_resource_inventory -from cloudexit.report.context import build_context, collect_provenance +from cloudexit.report.context import ( + EGRESS_DOCUMENT_CODE, + build_context, + collect_provenance, +) +from cloudexit.report.egress_report import generate_egress_report from cloudexit.report.executive_summary import generate_executive_summary from cloudexit.report.html_sections import html_sections from cloudexit.report.writers import ( @@ -432,6 +437,7 @@ def generate_report( raw_data_path: str, upload: dict[str, str] | None = None, sync_id: str | None = None, + egress_json_path: str | None = None, ) -> dict[str, Any]: try: db_path = os.path.join(report_path, "data", "assessment.db") @@ -493,6 +499,8 @@ def generate_report( provenance=provenance, report_path=report_path, ) + if egress_json_path: + context.egress_document_id = context.document_id_for(EGRESS_DOCUMENT_CODE) # Generate HTML report reports["HTML"] = generate_html_report( @@ -530,7 +538,20 @@ def generate_report( provenance=provenance.as_dict(), ) - return {"success": True, "reports": reports} + result = {"success": True, "reports": reports} + + # The Egress & Data Landscape is written last, from the same context: a + # problem with it never costs the Executive Summary + if egress_json_path: + try: + reports["EGRESS_PDF"] = generate_egress_report( + context, egress_json_path + ) + except Exception as e: + logger.error("Error generating the egress PDF: %s", e, exc_info=True) + result["egress_pdf_logs"] = str(e) + + return result except Exception as e: return {"success": False, "logs": f"Error generating report: {str(e)}"} diff --git a/cloudexit/report/context.py b/cloudexit/report/context.py index e56e52e..9f24c96 100644 --- a/cloudexit/report/context.py +++ b/cloudexit/report/context.py @@ -14,6 +14,16 @@ STRATEGY_CODES = {1: "repatriation", 2: "hybrid_cloud", 3: "alternate_cloud"} COST_MONTHS = 6 DOCUMENT_CODE = "ES" +EGRESS_DOCUMENT_CODE = "EDL" + + +def short_run_id(run_id): + # The end of a UUIDv7 is random; its start is the time. + return run_id.replace("-", "")[-8:] if run_id else "" + + +def format_document_id(code, run_id): + return f"{code}-{short_run_id(run_id) or 'LOCAL'}-EN" @dataclass(frozen=True) @@ -31,8 +41,7 @@ class Provenance: @property def run_short_id(self): - # The end of a UUIDv7 is random; its start is the time. - return self.run_id.replace("-", "")[-8:] if self.run_id else "" + return short_run_id(self.run_id) @property def collected_date(self): @@ -146,10 +155,16 @@ class ReportContext: alternatives: list report_path: str appearance: Appearance = DEFAULT_APPEARANCE + # Set when the run has an egress estimate, so the Executive Summary can name + # the document that carries it. + egress_document_id: str | None = None + + def document_id_for(self, code): + return format_document_id(code, self.provenance.run_id) @property def document_id(self): - return f"{DOCUMENT_CODE}-{self.provenance.run_short_id or 'LOCAL'}-EN" + return self.document_id_for(DOCUMENT_CODE) @property def total_resources(self): diff --git a/cloudexit/report/cover.py b/cloudexit/report/cover.py index 4af84e6..6e9c0a6 100644 --- a/cloudexit/report/cover.py +++ b/cloudexit/report/cover.py @@ -88,10 +88,10 @@ def _paragraph(value, style): class RecordCover: - def __init__(self, title, context, b): + def __init__(self, title, context, b, document_id=None): self.title = title self.project_name = context.name - self.document_id = context.document_id + self.document_id = document_id or context.document_id self.appearance = context.appearance self.report_path = context.report_path self.provenance = context.provenance diff --git a/cloudexit/report/egress.py b/cloudexit/report/egress.py index 36bd1ca..d4ca4a0 100644 --- a/cloudexit/report/egress.py +++ b/cloudexit/report/egress.py @@ -3,49 +3,37 @@ import logging import os import sqlite3 +from collections import Counter from datetime import datetime from typing import Any -from xml.sax.saxutils import escape 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 cloudexit.report.helpers import initials -from cloudexit.report.placeholder import IconPlaceholder -from cloudexit.resources import asset_path, find_asset +from cloudexit.report.helpers import format_gib, gib_number, initials, share_label +from cloudexit.resources import asset_path from cloudexit.catalogue.db import load_data from cloudexit.egress.model import ( GIB, compute_totals, - format_bytes, read_egress_inventory, ) from cloudexit.providers.aws.egress import ARCHIVE_TIERS as AWS_ARCHIVE_TIERS from cloudexit.providers.azure.egress import ARCHIVE_TIERS as AZURE_ARCHIVE_TIERS -from cloudexit.report.context import scope_facts -from cloudexit.report.notices import notices_html +from cloudexit.report.context import ( + DOCUMENT_CODE, + EGRESS_DOCUMENT_CODE, + PROVIDER_NAMES, + format_document_id, + scope_facts, +) +from cloudexit.report.notices import notices_html, text_parts +from cloudexit.report.sections import BOOK_DEMO_URL, PRODUCT_SITES from cloudexit.report.wording import BLOCKS -PDF_HEADER_TITLE = "EscapeCloud Community Edition - Data & Egress" - logger = logging.getLogger(__name__) DEFAULT_PRICING_ZONE = "zone1" UNIT_DIVISORS = {"GB": 10**9, "GiB": 2**30} +COMPONENTS = ("internet_egress", "archive_retrieval") ARCHIVE_TIERS_BY_CSP = {1: AZURE_ARCHIVE_TIERS, 2: AWS_ARCHIVE_TIERS} @@ -81,46 +69,6 @@ "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: { @@ -135,6 +83,8 @@ }, } +COVERAGE_NOTES_KEYS = {1: "eg_coverage_notes_azure", 2: "eg_coverage_notes_aws"} + FALLBACK_ICONS = { "microsoft.compute/disks": "/icons/azure/compute/10032-icon-service-Disks.png", "microsoft.compute/snapshots": ( @@ -148,187 +98,6 @@ } -def default_table_style() -> TableStyle: - """Shared header style used by summary and scope tables.""" - return TableStyle( - [ - ("BACKGROUND", (0, 0), (-1, 0), HexColor("#115e59")), - ("TEXTCOLOR", (0, 0), (-1, 0), colors.white), - ("GRID", (0, 0), (-1, -1), 1, HexColor("#000000")), - ("ALIGN", (0, 0), (-1, -1), "LEFT"), - ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), - ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"), - ("FONTSIZE", (0, 0), (-1, 0), 11), - ("BOTTOMPADDING", (0, 0), (-1, 0), 12), - ("TOPPADDING", (0, 0), (-1, 0), 12), - ] - ) - - -def build_summary_section(metadata, styles, content_style): - """Page 1: Summary table.""" - content = [] - content.append(Spacer(1, 12)) - content.append(Paragraph("Summary", styles["Heading1"])) - content.append(Paragraph("Quick overview of the assessment:", content_style)) - - cloud_service_provider_map = { - "1": "Microsoft Azure", - "2": "Amazon Web Services", - "3": "Alibaba Cloud", - "4": "Google Cloud", - } - exit_strategy_map = { - "1": "Repatriation to On-Premises", - "2": "Hybrid Cloud Adoption", - "3": "Migration to Alternate Cloud", - } - type_map = {"1": "Basic", "2": "Standard"} - - summary_data = [ - ["Name", "Value"], - [ - "Cloud Service Provider", - cloud_service_provider_map.get( - str(metadata["cloud_service_provider"]), "Unknown" - ), - ], - [ - "Exit Strategy", - exit_strategy_map.get(str(metadata["exit_strategy"]), "Unknown"), - ], - ["Assessment Type", type_map.get(str(metadata["assessment_type"]), "Unknown")], - ["TimeStamp", metadata["timestamp"]], - ] - - summary_table = Table(summary_data, colWidths=[4 * cm, 11.5 * cm]) - summary_table.setStyle(default_table_style()) - content.append(summary_table) - content.append(Spacer(1, 12)) - - upload = metadata.get("upload") - if upload: - if upload.get("status") == "uploaded": - note = "Uploaded to the platform." - if upload.get("url"): - url = escape(upload["url"]) - note += f' {url}' - else: - note = f"Not uploaded: {escape(upload.get('reason', 'unknown'))}" - content.append(Paragraph(note, content_style)) - return content - - -# Cap on how many subscriptions / resource groups / regions a scope row lists -# before it summarises the remainder. The full list stays in the manifest. -MAX_SCOPE_VALUES = 3 - - -def _shorten(value: str, head: int = 4, tail: int = 4) -> str: - if not isinstance(value, str): - return "N/A" - if len(value) <= head + tail + 1: - return value - return f"{value[:head]}…{value[-tail:]}" - - -def _join_scope_values(values) -> str: - if not values: - return "Not determinable from state" - shown = list(values)[:MAX_SCOPE_VALUES] - remaining = len(values) - len(shown) - text = ", ".join(shown) - return f"{text}, +{remaining} more" if remaining > 0 else text - - -def _build_tfstate_scope_rows(metadata, scope): - rows = [] - - file_name = scope.get("file") or "N/A" - sha256 = scope.get("sha256") - if sha256: - file_name = f"{file_name} (SHA-256 {_shorten(sha256)})" - rows.append(["File", file_name]) - - lineage = scope.get("lineage") - serial = scope.get("serial") - if lineage and serial is not None: - rows.append(["State", f"{lineage} (Serial: {serial})"]) - elif lineage: - rows.append(["State", lineage]) - elif serial is not None: - rows.append(["State", f"Serial: {serial}"]) - - if metadata["cloud_service_provider"] == 1: # Azure - rows.append(["Subscription", _join_scope_values(scope.get("subscriptions"))]) - rows.append( - ["Resource Group(s)", _join_scope_values(scope.get("resource_groups"))] - ) - else: # AWS - rows.append(["Region(s)", _join_scope_values(scope.get("locations"))]) - - return rows - - -def _live_scope_rows(cloud_service_provider, provider_details): - # The same facts as the Assessment Scope on page 2 of the Executive - # Summary, and nothing that identifies the credentials in use - b = BLOCKS - scope = scope_facts(provider_details, cloud_service_provider, None) - values = dict(scope.rows) - if scope.kind == "azure": - return [ - [b["ev_tenant_id"], values.get("tenant") or b["ev_dash"]], - [b["ev_subscription_id"], values.get("subscription") or b["ev_dash"]], - [b["ev_resource_group"], values.get("resource_group") or b["ev_dash"]], - ] - rows = [] - if values.get("account"): - rows.append([b["ev_account_id"], values["account"]]) - rows.append([b["ev_region"], values.get("region") or b["ev_dash"]]) - return rows - - -def build_scope_section( - metadata, provider_details, styles, content_style, tfstate_scope=None -): - """Page 1: Scope of Assessment table.""" - content = [] - content.append(Paragraph("Scope of Assessment", styles["Heading2"])) - - tfstate_path = provider_details.get("tfstatePath") - if tfstate_path: - content.append( - Paragraph( - "Assessed from Terraform state. Resources not managed by " - "Terraform are not included.", - content_style, - ) - ) - else: - content.append(Paragraph("Defined scope of assessment:", content_style)) - - scope_data = [["Name", "Value"]] - - if tfstate_path: - # Fall back to the little we know from the config when the manifest is - # unavailable, rather than rendering a table of N/A. - scope = tfstate_scope or {"file": os.path.basename(tfstate_path)} - scope_data.extend(_build_tfstate_scope_rows(metadata, scope)) - elif metadata["cloud_service_provider"] in (1, 2): - scope_data.extend( - _live_scope_rows(metadata["cloud_service_provider"], provider_details) - ) - else: - scope_data.append(["N/A", "N/A"]) - - scope_table = Table(scope_data, colWidths=[4 * cm, 11.5 * cm]) - scope_table.setStyle(default_table_style()) - content.append(scope_table) - content.append(Spacer(1, 12)) - return content - - def _unit_divisor(unit: str) -> float: return UNIT_DIVISORS.get(unit, GIB) @@ -417,9 +186,9 @@ def format_fee(fee: float | None) -> str: return f"${fee:,.2f}" -def free_tier_limit_display( +def _free_tier( prices: list[dict[str, Any]], cloud_service_provider: int -) -> str | None: +) -> dict[str, Any] | None: tiers = sorted( ( price @@ -431,10 +200,24 @@ def free_tier_limit_display( 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 tiers[0] return None +def free_tier_limit_display( + prices: list[dict[str, Any]], cloud_service_provider: int +) -> str | None: + tier = _free_tier(prices, cloud_service_provider) + return f"{tier['tier_to']:g} {tier['unit']}" if tier else None + + +def free_allowance_bytes( + prices: list[dict[str, Any]], cloud_service_provider: int +) -> float | None: + tier = _free_tier(prices, cloud_service_provider) + return tier["tier_to"] * _unit_divisor(tier["unit"]) if tier else None + + def _summarize_totals( totals: dict[str, Any], total_fee: float, @@ -448,8 +231,8 @@ def _summarize_totals( 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"]) + total_fee_display = format_fee(total_fee) + total_data_display = format_gib(totals["known_size_bytes"]) if unknown_count > 0: total_data_display = f"{total_data_display}+" if not free_tier: @@ -547,7 +330,7 @@ def _build_type_groups( ) for group in groups.values(): - group["size_display"] = format_bytes(group["size_bytes"]) + group["size_display"] = format_gib(group["size_bytes"]) group["fee_display"] = format_fee(group["fee"]) return sorted( @@ -583,6 +366,23 @@ def _load_estimate( return meta, rows, totals, pricing, total_fee, type_groups +def load_egress_estimate(report_path: str, json_path: str) -> dict[str, Any]: + meta, rows, totals, pricing, total_fee, type_groups = _load_estimate( + report_path, json_path + ) + return { + "meta": meta, + "rows": rows, + "totals": totals, + "pricing": pricing, + "total_fee": total_fee, + "type_groups": type_groups, + "summary": _summarize_totals( + totals, total_fee, pricing, meta["cloud_service_provider"] + ), + } + + def _generated_at(timestamp): # The run's own time, as the estimate stored it; now when it cannot be read try: @@ -591,652 +391,452 @@ def _generated_at(timestamp): return datetime.now() -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 - ) +def _share(part, whole): + return part / whole * 100 if whole else 0 - summary = _summarize_totals( - totals, total_fee, pricing, meta["cloud_service_provider"] - ) - allocation_labels, allocation_values, allocation_colors = _build_allocation( - rows - ) +def _plus(text, estimate): + # Resources without a size are left out, so the figure is a lower bound + return f"{text}+" if estimate["summary"]["unknown_count"] > 0 else text - 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_content = asset_path("template", "egress.html").read_text( - encoding="utf-8" - ) - - 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, - notices=notices_html( - meta["cloud_service_provider"], _generated_at(meta["timestamp"]) - ), - ) - - html_path = os.path.join(report_path, "egress.html") - with open(html_path, "w", encoding="utf-8") 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 | None, label: str) -> Any: - icon_path = find_asset(report_path, icon) if icon else None - if icon_path is None: - return IconPlaceholder(label, 20) - return Image(str(icon_path), width=20, height=20) +def introduction_facts(scope, b=BLOCKS): + return { + "heading": b["eg_heading_introduction"], + "purpose_heading": b["eg_heading_purpose"], + "purpose": b["eg_purpose_block"], + "scope_heading": b["eg_heading_scope"], + "scope": scope_sentence(scope, b), + "limitations_heading": b["eg_heading_limitations"], + "limitations": [ + b["eg_limitations_block"], + b["eg_limitations_scope"], + b["eg_limitations_data"], + ], + } -def _fee_breakdown_display(group: dict[str, Any]) -> str: - if group["fee"] is None: - return "n/a (not sized)" - return group["fee_display"] +def scope_sentence(scope, b=BLOCKS): + if scope.kind == "azure": + return b["eg_scope_azure"].format( + resource_group=scope.resource_group or b["ev_dash"] + ) + if scope.kind == "aws": + return b["eg_scope_aws"].format(region=scope.region or b["ev_dash"]) + return b["eg_scope_plain"] -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: +def coverage_facts(cloud_service_provider, b=BLOCKS): + key = COVERAGE_NOTES_KEYS.get(cloud_service_provider) + if not key: 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 + return { + "heading": b["eg_heading_coverage"], + "intro": b["eg_coverage_intro"], + "headers": (b["eg_th_resource_type"], b["eg_th_measured"]), + "rows": [(name, note or b["ev_dash"]) for name, note in b[key]], + } -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, - ) +def landscape_facts(estimate, b=BLOCKS): + rows = estimate["rows"] + summary = estimate["summary"] + total = estimate["totals"]["known_size_bytes"] + resources = len(rows) + unsized = [row for row in rows if row["size_bytes"] is None] + sized = resources - len(unsized) + counts = Counter(row["label"] for row in rows) + groups = sorted( + (group for group in estimate["type_groups"] if group["size_bytes"] is not None), + key=lambda group: -group["size_bytes"], ) - 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 - + top = groups[0] if groups else None -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)." - ) + if not resources: + lead = b["eg_no_data_notice"] + elif not sized: + key = "eg_land_none_sized_one" if resources == 1 else "eg_land_none_sized" + lead = b[key].format(resources=resources) 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, + sentences = [ + b["eg_land_lead_one" if resources == 1 else "eg_land_lead"].format( + total=format_gib(total), sized=sized, resources=resources + ) + ] + if len(groups) > 1 and total > 0 and top["size_bytes"]: + sentences.append( + b["eg_land_top"].format( + type=top["label"], + size=format_gib(top["size_bytes"]), + share=share_label(_share(top["size_bytes"], total)), ) - ] - ) - 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)) + if len(unsized) == 1: + sentences.append(b["eg_land_unsized_one"]) + elif unsized: + sentences.append(b["eg_land_unsized_many"].format(count=len(unsized))) + lead = " ".join(sentences) - 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), - ] + unsized_counts = Counter(row["label"] for row in unsized) + not_sized = [ + b["eg_land_not_sized_one" if count == 1 else "eg_land_not_sized_many"].format( + type=label, count=count ) - ) - 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, + for label, count in sorted( + unsized_counts.items(), key=lambda item: (-item[1], item[0]) ) + ] + return { + "heading": b["eg_heading_landscape"], + "empty": not resources, + "lead": lead, + "stats": [ + (b["eg_st_volume"], summary["total_data_display"], False), + ( + b["eg_st_sized"], + b["eg_st_sized_value"].format(sized=sized, resources=resources), + False, + ), + (b["eg_st_not_sized"], str(len(unsized)), bool(unsized)), + (b["eg_st_largest"], top["label"] if top else b["ev_dash"], False), + ], + "headers": ( + b["eg_th_resource_type"], + b["eg_th_resources"], + b["eg_th_size_gib"], + b["eg_th_share"], + ), + "rows": [ + { + "label": group["label"], + "icon": group["icon"], + "initials": group["initials"], + "resources": counts[group["label"]], + "size": gib_number(group["size_bytes"]), + "share": _share(group["size_bytes"], total), + "share_text": share_label(_share(group["size_bytes"], total)), + } + for group in groups + ], + "total": (b["eg_total"], resources, _plus(gib_number(total), estimate)), + "not_sized": ( + b["eg_land_not_sized"].format(types=", ".join(not_sized)) + if not_sized + else None + ), + "caption": b["eg_land_caption"], + "source_note": b["eg_land_source_note"], + } + + +def _allowance_text(prices, cloud_service_provider): + # The provider's own figure, plus the GiB value when it is published in GB + tier = _free_tier(prices, cloud_service_provider) + if tier is None: + return None + limit = f"{tier['tier_to']:g} {tier['unit']}" + if tier["unit"] == "GiB": + return limit + return ( + f"{limit} ({format_gib(free_allowance_bytes(prices, cloud_service_provider))})" ) - 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["label"]), - 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( +def cost_facts(estimate, b=BLOCKS): + rows = estimate["rows"] + summary = estimate["summary"] + csp = estimate["meta"]["cloud_service_provider"] + total = estimate["totals"]["known_size_bytes"] + total_fee = estimate["total_fee"] + resources = len(rows) + unsized = sum(1 for row in rows if row["size_bytes"] is None) + sized = resources - unsized + allowance = free_allowance_bytes(estimate["pricing"], csp) + allowance_text = _allowance_text(estimate["pricing"], csp) + groups = [ + group for group in estimate["type_groups"] if group["size_bytes"] is not None + ] + billable = total if allowance is None else max(0, total - allowance) + + if not resources: + sentences = [b["eg_no_data_notice"]] + elif not sized: + sentences = [b["eg_cost_lead_none"]] + elif summary["free_tier"]: + sentences = [ + b["eg_cost_lead_free"].format(fee=format_fee(total_fee)), + b["eg_cost_headroom"].format( + used=format_gib(total), allowance=allowance_text + ), + ] + else: + sentences = [ + b["eg_cost_lead_paid"].format( + total=format_gib(total), fee=format_fee(total_fee) + ) + ] + if allowance is not None: + sentences.append( + b["eg_cost_lead_allowance"].format( + allowance=allowance_text, billable=format_gib(billable) + ) + ) + top = max(groups, key=lambda group: group["fee"] or 0) + if total_fee > 0 and top["fee"] and len(groups) > 1: + sentences.append( + b["eg_cost_lead_top"].format( + type=top["label"], + share=share_label(_share(top["fee"], total_fee)), + ) + ) + if sized and unsized == 1: + sentences.append(b["eg_cost_unsized_one"]) + elif sized and unsized: + sentences.append(b["eg_cost_unsized_many"].format(count=unsized)) + + if summary["free_tier"]: + volume_stat = (b["eg_st_allowance"], summary["free_tier_limit"], False) + else: + volume_stat = (b["eg_st_billable"], format_gib(billable), False) + return { + "heading": b["eg_heading_costs"], + "empty": not resources, + "lead": " ".join(sentences), + "stats": [ + (b["eg_st_fee"], summary["total_fee_display"], False), + (b["eg_st_volume"], summary["total_data_display"], False), + volume_stat, + (b["eg_st_not_sized"], str(unsized), bool(unsized)), + ], + "headers": ( + b["eg_th_resource_type"], + b["eg_th_size_gib"], + b["eg_th_share"], + b["eg_th_fee"], + ), + # A fee of zero has no split to show + "rows": ( [ - ("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"), + { + "label": group["label"], + "icon": group["icon"], + "initials": group["initials"], + "size": gib_number(group["size_bytes"]), + "share": _share(group["size_bytes"], total), + "share_text": share_label(_share(group["size_bytes"], total)), + "fee": format_fee(group["fee"]), + } + for group in groups ] - ) - ) - content.append(landscape_table) - content.append(PageBreak()) - return content + if total_fee > 0 + else [] + ), + "total": ( + b["eg_total"], + _plus(gib_number(total), estimate), + summary["total_fee_display"], + ), + "caption": b["eg_cost_caption"], + } -def _tier_display(row: dict[str, Any]) -> str: +def _tier_display(row, b): unit = row["unit"] if row["tier_from"] == 0 and row["tier_to"] is None: - return "all volumes" + return b["eg_tier_all_volumes"] if row["tier_from"] == 0 and row["price_per_unit"] == 0: - return f"first {row['tier_to']:,.0f} {unit} / month" + return b["eg_tier_first"].format(to=f"{row['tier_to']:,.0f}", unit=unit) 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}" + return b["eg_tier_over"].format(from_=f"{row['tier_from']:,.0f}", unit=unit) + return b["eg_tier_range"].format( + from_=f"{row['tier_from']:,.0f}", to=f"{row['tier_to']:,.0f}", unit=unit + ) -def _price_display(row: dict[str, Any]) -> str: +def _price_display(row, b): if row["price_per_unit"] == 0: - return "Free" - return f"${row['price_per_unit']:g} / {row['unit']}" + return b["eg_price_free"] + return f"{row['price_per_unit']:g} / {row['unit']}" -def _build_pricing_basis_section(prices, cloud_service_provider, styles, content_style): +def pricing_facts(estimate, b=BLOCKS): + csp = estimate["meta"]["cloud_service_provider"] rows = sorted( ( price - for price in prices - if price["csp"] == cloud_service_provider + for price in estimate["pricing"] + if price["csp"] == csp and price["zone"] == DEFAULT_PRICING_ZONE - and price["component"] in COMPONENT_LABELS - ), - key=lambda price: ( - list(COMPONENT_LABELS).index(price["component"]), - price["tier_from"], + and price["component"] in COMPONENTS ), + key=lambda price: (COMPONENTS.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)) + return None - show_valid_from = len({row["valid_from"] for row in rows}) > 1 - header = ["#", "Component", "Tier", "Price"] + provider = PROVIDER_NAMES.get(csp) or b["eg_provider_fallback"] + valid_dates = {row["valid_from"] for row in rows} + show_valid_from = len(valid_dates) > 1 + if show_valid_from: + intro = b["eg_pricing_intro_mixed"].format(provider=provider) + else: + intro = b["eg_pricing_intro"].format( + provider=provider, valid=next(iter(valid_dates)) + ) + headers = [b["eg_th_component"], b["eg_th_tier"], b["eg_th_price"]] if show_valid_from: - header.append("Valid from") - table_data = [header] - for index, row in enumerate(rows, start=1): + headers.append(b["eg_th_valid_from"]) + + table = [] + for row in rows: table_row = [ - str(index), - COMPONENT_LABELS[row["component"]], - _tier_display(row), - _price_display(row), + b[f"eg_component_{row['component']}"], + _tier_display(row, b), + _price_display(row, b), ] if show_valid_from: table_row.append(row["valid_from"]) - table_data.append(table_row) + table.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"), - ] - ) + unit = next( + (row["unit"] for row in rows if row["component"] == "internet_egress"), None ) - 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 + units = {"GB": "eg_pricing_units", "GiB": "eg_pricing_units_gib"}.get(unit) + shown = {row["component"] for row in rows} + links = PRICING_SOURCE_LINKS.get(csp, {}) + sources = [ + (component, links[component]) + for component in COMPONENTS + if component in shown and component in links ] - if len(source_lines) == 1: - content.append(Paragraph(f"Source: {source_lines[0][1]}", footnote_style)) + if len(sources) == 1: + source_lines = [(b["eg_pricing_source"], sources[0][1])] 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 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 = title - left_text_content2 = f"Date: {current_date}" - - # Define the header content with Paragraphs - header_style = ParagraphStyle( - "HeaderStyle", fontSize=10, textColor=HexColor("#9cafae") - ) - header_data = [ - [Paragraph(left_text_content1, header_style), "", ""], - [Paragraph(left_text_content2, header_style), "", ""], - ] + source_lines = [ + ( + b["eg_pricing_source_component"].replace( + "{component}", b[f"eg_component_{component}"] + ), + link, + ) + for component, link in sources + ] + return { + "heading": b["eg_heading_pricing"], + "intro": intro, + "headers": headers, + "rows": table, + "units": b[units].format(provider=provider) if units else None, + "no_archive": ( + None if "archive_retrieval" in shown else b["eg_pricing_no_archive"] + ), + # (template with a {link} field, url) + "sources": source_lines, + "note": b["eg_pricing_note"], + "glossary_heading": b["eg_heading_glossary"], + "glossary": [ + (entry["term"], entry["definition"]) + for entry in sorted( + b["eg_glossary"], key=lambda entry: entry["term"].lower() + ) + ], + } - # Create the header table - table = Table( - header_data, colWidths=[width - 188 - doc.rightMargin - doc.leftMargin, 10, 150] - ) - # Define the style for the table - table.setStyle( - TableStyle( - [ - ("SPAN", (1, 0), (1, 1)), # Merge Column 2 in both rows - ("SPAN", (2, 0), (2, 1)), # Merge Column 3 in both rows - ( - "ALIGN", - (0, 0), - (0, 0), - "LEFT", - ), # Align left_text_content1 to the left - ( - "ALIGN", - (0, 1), - (0, 1), - "LEFT", - ), # Align left_text_content2 to the left - ("ALIGN", (2, 0), (2, 1), "RIGHT"), # Align logo to the right - ("VALIGN", (0, 0), (0, 1), "TOP"), # Vertically align to the top - ("VALIGN", (2, 0), (2, 1), "MIDDLE"), # Vertically align to the middle - # ('GRID', (0, 0), (-1, -1), 0.5, colors.red), # Temporary borders for visualization - ] +def next_steps_facts( + executive_id, uploaded=False, platform_url=None, assessment_id=None, b=BLOCKS +): + # Paragraphs as text and (text, url) links, like the notices + sites = dict(PRODUCT_SITES) + paragraphs = [text_parts(b["eg_next_lead"], {}, {"executive_id": executive_id})] + if uploaded: + paragraphs.append([b["eg_next_uploaded"]]) + if platform_url: + paragraphs.append(text_parts(b["fa_uploaded_link"], {"link": platform_url})) + elif assessment_id: + paragraphs.append( + text_parts(b["fa_uploaded_id"], {}, {"assessment_id": assessment_id}) + ) + else: + paragraphs.append( + text_parts( + b["eg_next_platforms"], + {"exitcloud_url": sites["exitcloud"], "book_demo_url": BOOK_DEMO_URL}, + ) ) - ) - # Build the header table and draw it on the canvas - table.wrapOn(canvas, doc.leftMargin, height - doc.topMargin) - table.drawOn(canvas, doc.leftMargin, height - doc.topMargin) - - # Add the logo - logo_path = find_asset(report_path, "img/logo/report_logo.png") - logo = Image(str(logo_path), width=150, height=30) - - # Aligning logo vertically with the text - logo_y = height - doc.topMargin + 5 - logo.drawOn(canvas, width - 150 - doc.rightMargin, logo_y) - - # Line below the header - canvas.setStrokeColor(HexColor("#115e59")) - canvas.setLineWidth(1) - line_y = height - doc.topMargin - 10 - canvas.line(doc.leftMargin, line_y, width - doc.rightMargin, line_y) - - # Footer - footer_padding = 15 # Add padding under the page number - canvas.setStrokeColor(HexColor("#115e59")) - canvas.line(40, 60 + footer_padding, A4[0] - 40, 60 + footer_padding) - - canvas.setFont("Helvetica", 8) - canvas.drawString(A4[0] / 2 - 30, 60 + footer_padding - 15, f"Page {doc.page}") - - canvas.setFont("Helvetica-Oblique", 8) - canvas.setFillColor(HexColor("#9cafae")) - canvas.drawCentredString( - A4[0] / 2, - 40, - "EscapeCloud Community Edition - This report is provided 'As Is,' without any warranty of any kind.", - ) - canvas.drawCentredString( - A4[0] / 2, - 30, - "EscapeCloud makes no warranty that the information contained in this report is complete or error-free. Copyright 2024-2026", - ) + return {"heading": b["eg_heading_next_steps"], "paragraphs": paragraphs} - # Restore the state of the canvas - canvas.restoreState() - -def generate_egress_pdf_report( - report_path: str, json_path: str, provider_details: dict[str, Any] +def generate_egress_html_report( + report_path: str, + json_path: str, + provider_details: dict[str, Any] | None = None, + run_id: str | None = None, ) -> dict[str, Any]: try: - meta, rows, totals, pricing, total_fee, type_groups = _load_estimate( - report_path, json_path - ) + estimate = load_egress_estimate(report_path, json_path) + meta, rows = estimate["meta"], estimate["rows"] + csp = meta["cloud_service_provider"] + summary = estimate["summary"] + b = BLOCKS - summary = _summarize_totals( - totals, total_fee, pricing, meta["cloud_service_provider"] + allocation_labels, allocation_values, allocation_colors = _build_allocation( + rows ) - 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) + type_groups = estimate["type_groups"] + 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] - 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, + template_content = asset_path("template", "egress.html").read_text( + encoding="utf-8" ) - content += _build_pricing_basis_section( - pricing, meta["cloud_service_provider"], styles, content_style + + scope = scope_facts(provider_details or {}, csp, None) + executive_id = format_document_id(DOCUMENT_CODE, run_id) + env = Environment(autoescape=True) + template = env.from_string(template_content) + html_content = template.render( + page_title=f"{b['cover_title_egress']} – {meta['name']}", + document_id=format_document_id(EGRESS_DOCUMENT_CODE, run_id), + name=meta["name"], + cloud_service_provider=csp, + 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_gib(estimate["totals"]["known_size_bytes"]), + type_groups=type_groups, + introduction=introduction_facts(scope, b), + coverage=coverage_facts(csp, b), + landscape=landscape_facts(estimate, b), + costs=cost_facts(estimate, b), + pricing=pricing_facts(estimate, b), + next_steps=next_steps_facts(executive_id, b=b), + notices=notices_html(csp, _generated_at(meta["timestamp"]), b, egress=True), ) - doc.build(content, onFirstPage=header_footer, onLaterPages=header_footer) + html_path = os.path.join(report_path, "egress.html") + with open(html_path, "w", encoding="utf-8") as report_file: + report_file.write(html_content) return { "success": True, - "logs": "Egress PDF report generated successfully.", - "pdf_path": pdf_path, + "logs": "Egress report generated successfully.", + "html_path": html_path, } except Exception as e: - logger.error(f"Error generating egress PDF report: {str(e)}", exc_info=True) + logger.error(f"Error generating egress report: {str(e)}", exc_info=True) return {"success": False, "logs": str(e)} diff --git a/cloudexit/report/egress_report.py b/cloudexit/report/egress_report.py new file mode 100644 index 0000000..6acc846 --- /dev/null +++ b/cloudexit/report/egress_report.py @@ -0,0 +1,76 @@ +import os +from functools import partial + +from reportlab.lib.pagesizes import A4 +from reportlab.lib.units import cm +from reportlab.platypus import Spacer + +from .base import NumberedCanvas, ReportDocTemplate, set_pdf_metadata +from .context import EGRESS_DOCUMENT_CODE +from .cover import RecordCover +from .egress import load_egress_estimate +from .egress_sections import ( + appendix, + coverage, + data_landscape, + estimated_costs, + introduction, + next_steps, +) +from .evidence import build_document_information_page +from .flow import chapter_break, section_break +from .notices import legal_notices +from .styles import load_styles +from .wording import BLOCKS + + +def build_egress_content(context, estimate, styles, b): + content = [Spacer(1, 0), chapter_break()] + content += build_document_information_page( + context, + b, + document_id=context.document_id_for(EGRESS_DOCUMENT_CODE), + document_type=b["ev_egress_report"], + companion_id=context.document_id, + ) + content += introduction(context, styles, b) + content += coverage(context, styles, b) + content.append(chapter_break()) + content += data_landscape(context, estimate, styles, b) + content.append(section_break()) + content += estimated_costs(context, estimate, styles, b) + content.append(chapter_break()) + content += appendix(context, estimate, styles, b) + content.append(Spacer(1, 12)) + content += next_steps(context, styles, b) + content.append(chapter_break()) + content += legal_notices(context, styles, b, egress=True) + return content + + +def generate_egress_report(context, json_path, path=None): + path = path or os.path.join(context.report_path, "egress.pdf") + b = BLOCKS + estimate = load_egress_estimate(context.report_path, json_path) + cover = RecordCover( + b["cover_title_egress"], + context, + b, + document_id=context.document_id_for(EGRESS_DOCUMENT_CODE), + ) + doc = ReportDocTemplate( + path, + context, + cover=cover, + pagesize=A4, + rightMargin=2.5 * cm, + leftMargin=2.5 * cm, + topMargin=2.5 * cm, + bottomMargin=2.5 * cm, + ) + set_pdf_metadata(doc, context, b) + doc.multiBuild( + build_egress_content(context, estimate, load_styles(context.appearance), b), + canvasmaker=partial(NumberedCanvas, page_of=b["footer_page_of"]), + ) + return path diff --git a/cloudexit/report/egress_sections.py b/cloudexit/report/egress_sections.py new file mode 100644 index 0000000..553a46a --- /dev/null +++ b/cloudexit/report/egress_sections.py @@ -0,0 +1,324 @@ +from xml.sax.saxutils import escape, quoteattr + +from reportlab.lib.colors import HexColor +from reportlab.lib.enums import TA_RIGHT +from reportlab.lib.units import cm +from reportlab.platypus import Paragraph, Spacer, Table, TableStyle + +from cloudexit.resources import find_asset + +from .egress import ( + coverage_facts, + cost_facts, + introduction_facts, + landscape_facts, + next_steps_facts, + pricing_facts, +) +from .evidence import link_markup +from .flow import section +from .notices import parts_markup +from .palette import HAIRLINE, HOTSPOT_COUNT, INK, MUTED, PROVENANCE_BACKGROUND, RULE +from .placeholder import LabelWithPlaceholder +from .sections import GLOSSARY_TERM_WIDTH +from .table_style import data_table, header_cell, share_bar, summary_strip, text_style + +BODY = text_style("EgressBody", 8.5, 11) +BODY_RIGHT = text_style("EgressBodyRight", 8.5, 11, alignment=TA_RIGHT) +BODY_BOLD = text_style("EgressBodyBold", 8.5, 11, bold=True) +BODY_BOLD_RIGHT = text_style( + "EgressBodyBoldRight", 8.5, 11, bold=True, alignment=TA_RIGHT +) +CAPTION = text_style("EgressCaption", 7.5, 10, MUTED) + + +def _text(value, style=BODY): + return Paragraph(escape(str(value)), style) + + +def _right_header(text, appearance): + return header_cell(text, appearance, TA_RIGHT, split_long_words=False) + + +def _total_row_style(rows): + return [ + ("LINEABOVE", (0, len(rows)), (-1, len(rows)), 0.8, HexColor(INK)), + ("NOSPLIT", (0, len(rows) - 1), (-1, len(rows))), + ] + + +def _type_cell(row, report_path): + label = f"{escape(row['label'])}" + icon = find_asset(report_path, row["icon"]) if row["icon"] else None + if icon is None: + # A missing icon must not cost the whole document + return LabelWithPlaceholder(row["label"], label, BODY) + return Paragraph( + f'$0
', html)
+ self.assertIn('$0.00
', html)
self.assertIn("5.0 GiB", html)
self.assertIn("100 GiB free tier", html)
self.assertNotIn("No fee estimate available.", html)
@@ -597,369 +587,3 @@ def test_missing_json_returns_failure(self):
)
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_inventory_raw_data.json")
- with open(json_path, "w", encoding="utf-8") as json_file:
- json.dump(payload, json_file)
- _seed_egress_inventory(
- tmp_dir,
- payload["data"]["resources"],
- _archive_tiers(payload["data"]["resources"]),
- )
-
- with patch(
- "cloudexit.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/tests/report/test_egress_report.py b/tests/report/test_egress_report.py
new file mode 100644
index 0000000..efd0352
--- /dev/null
+++ b/tests/report/test_egress_report.py
@@ -0,0 +1,600 @@
+import copy
+import json
+import re
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from pypdf import PdfReader
+
+from cloudexit.engine import generate_report, perform_risk_assessment
+from cloudexit.report.context import (
+ EGRESS_DOCUMENT_CODE,
+ Provenance,
+ format_document_id,
+)
+from cloudexit.report.egress import generate_egress_html_report
+from cloudexit.report.egress_report import generate_egress_report
+from cloudexit.report.sections import AWS_ICONS_URL, AZURE_ICONS_URL
+from cloudexit.report.wording import BLOCKS
+from cloudexit.version import __version__
+from tests.report import test_egress as egress_helpers
+from tests.report.test_basic_report import (
+ COLLECTED,
+ URL,
+ html_text,
+ metadata_texts,
+ outline_entries,
+ page_texts,
+ provenance,
+ uploaded,
+)
+from tests.report.test_no_credentials import pricing_for_both_clouds
+from tests.support.flow import build_flow_run
+from tests.support.report import (
+ build_fixture_context,
+ build_report_fixture,
+ stage_minimal_assets,
+)
+from tests.support.report_terms import american_hits, find_terms, retired_term_hits
+from tests.support.sync import SYNC_ID
+
+GIB = egress_helpers.GIB
+ES_ID = format_document_id("ES", SYNC_ID)
+EDL_ID = format_document_id(EGRESS_DOCUMENT_CODE, SYNC_ID)
+OUTLINE = [
+ (0, "Introduction"),
+ (1, "Purpose"),
+ (1, "Scope"),
+ (1, "Limitations"),
+ (1, "Scope of the Estimation"),
+ (0, "Data Landscape"),
+ (0, "Estimated Costs"),
+ (0, "Appendix – Egress Prices"),
+ (1, "Glossary"),
+ (0, "Next Steps"),
+ (0, "Legal Notices"),
+]
+MULTI_TIB = [
+ egress_helpers._row("proddata", "Storage Account", "object", 2048 * GIB),
+ egress_helpers._row("vm-osdisk", "Managed Disk", "block", 512 * GIB),
+ egress_helpers._row("vault1", "Recovery Services Vault", "backup", None),
+]
+FREE_TIER = [egress_helpers._row("sa1", "Storage Account", "object", 20 * GIB)]
+WITH_UNKNOWN = [
+ egress_helpers._row("sa1", "Storage Account", "object", 2048 * GIB),
+ egress_helpers._row("db1", "SQL Database", "database", None, size_unknown=True),
+]
+
+
+class EgressReportTestCase(unittest.TestCase):
+ def setUp(self):
+ tmp = tempfile.TemporaryDirectory()
+ self.addCleanup(tmp.cleanup)
+ self.tmp = Path(tmp.name)
+ self.count = 0
+
+ def fixture(self, provider):
+ fixture = copy.deepcopy(build_report_fixture())
+ fixture["metadata"]["cloud_service_provider"] = provider
+ if provider == 1:
+ fixture["provider_details"] = {
+ "tenantId": "11111111-1111-4111-8111-111111111111",
+ "subscriptionId": "22222222-2222-4222-8222-222222222222",
+ "resourceGroupName": "Prod-RG",
+ "clientSecret": "very-secret-client-value",
+ }
+ return fixture
+
+ def prepare(self, rows, provider):
+ self.count += 1
+ folder = self.tmp / f"report-{self.count}"
+ folder.mkdir()
+ stage_minimal_assets(str(folder))
+ payload = egress_helpers._payload(rows)
+ payload["meta"]["cloud_service_provider"] = provider
+ json_path = folder / "egress_inventory_raw_data.json"
+ json_path.write_text(json.dumps(payload), encoding="utf-8")
+ egress_helpers._seed_egress_inventory(
+ str(folder), rows, egress_helpers._archive_tiers(rows)
+ )
+ return folder, str(json_path)
+
+ def render(self, rows=MULTI_TIB, provider=1, prov=None):
+ folder, json_path = self.prepare(rows, provider)
+ context = build_fixture_context(
+ self.fixture(provider), folder, provenance=prov or provenance()
+ )
+ with patch(
+ "cloudexit.report.egress.load_data", side_effect=pricing_for_both_clouds
+ ):
+ path = generate_egress_report(context, json_path)
+ return path, context, PdfReader(path)
+
+ def render_html(self, rows=MULTI_TIB, provider=1, details=None):
+ folder, json_path = self.prepare(rows, provider)
+ with patch(
+ "cloudexit.report.egress.load_data", side_effect=pricing_for_both_clouds
+ ):
+ result = generate_egress_html_report(
+ str(folder), json_path, details, SYNC_ID
+ )
+ self.assertTrue(result["success"], result["logs"])
+ html = Path(result["html_path"]).read_text(encoding="utf-8")
+ return html, html_text(html)
+
+ def scenarios(self):
+ for provider, name in ((1, "azure"), (2, "aws")):
+ for state, prov in (("offline", provenance()), ("uploaded", uploaded())):
+ yield f"{name} {state}", provider, prov
+
+
+class IdentityTests(unittest.TestCase):
+ def test_a_context_names_a_document_per_code(self):
+ context = build_fixture_context(
+ build_report_fixture(), tempfile.gettempdir(), provenance=provenance()
+ )
+
+ self.assertEqual(context.document_id, ES_ID)
+ self.assertEqual(context.document_id_for("EDL"), EDL_ID)
+ self.assertRegex(EDL_ID, r"^EDL-[0-9a-f]{8}-EN$")
+ # Both end with the same run, so the pair can be matched by eye
+ self.assertEqual(ES_ID[3:], EDL_ID[4:])
+ self.assertIsNone(context.egress_document_id)
+
+ def test_a_run_without_an_id_is_local(self):
+ self.assertEqual(format_document_id("EDL", None), "EDL-LOCAL-EN")
+
+
+class CoverAndPageTwoTests(EgressReportTestCase):
+ def test_the_cover_names_the_document_and_the_edition(self):
+ for name, provider, prov in self.scenarios():
+ with self.subTest(name):
+ _path, _context, reader = self.render(provider=provider, prov=prov)
+ cover = page_texts(reader)[0]
+
+ self.assertIn("Egress & Data Landscape", cover)
+ self.assertIn(EDL_ID, cover)
+ self.assertIn("EDITION Community", cover)
+ self.assertNotIn("Classification", cover)
+ self.assertNotIn("CLASSIFICATION", cover)
+ everything = " ".join(page_texts(reader)).lower()
+ self.assertNotIn("confidential", everything)
+ self.assertNotIn("Page 1 of", cover)
+
+ def test_page_two_is_the_document_information_with_the_companion(self):
+ for name, provider, prov in self.scenarios():
+ with self.subTest(name):
+ _path, _context, reader = self.render(provider=provider, prov=prov)
+ pages = page_texts(reader)
+
+ self.assertIn("Document Information", pages[1])
+ self.assertIn("Assessment Scope", pages[1])
+ self.assertIn("Provenance", pages[1])
+ self.assertIn("Document Type Egress & Data Landscape", pages[1])
+ self.assertIn(f"Companion Document {ES_ID}", pages[1])
+ self.assertIn(EDL_ID, pages[1])
+ self.assertIn(f"Page 2 of {len(pages)}", pages[1])
+
+ def test_an_uploaded_run_shows_the_platform_link_on_page_two(self):
+ # The PDF is written after the upload, so it knows about it
+ _path, _context, reader = self.render(prov=uploaded())
+ self.assertIn(URL, page_texts(reader)[1])
+ self.assertIn("Uploaded to the platform", page_texts(reader)[1])
+
+ _path, _context, offline = self.render(prov=provenance())
+ self.assertNotIn(URL, page_texts(offline)[1])
+ self.assertIn("Self-collected data", page_texts(offline)[1])
+
+ def test_the_pdf_metadata_follows_the_cover(self):
+ _path, _context, reader = self.render()
+ info = reader.metadata
+
+ self.assertEqual(info.title, "Egress & Data Landscape – Smoke Test Assessment")
+ self.assertEqual(info.author, "cloudexit")
+ self.assertEqual(info.creator, f"cloudexit {__version__}")
+ self.assertIn(EDL_ID, info.subject)
+ for keyword in (EDL_ID, SYNC_ID):
+ self.assertIn(keyword, info["/Keywords"])
+ for text in metadata_texts(reader):
+ self.assertNotIn("EscapeCloud_-_", text)
+
+
+class StructureTests(EgressReportTestCase):
+ def test_the_outline_is_the_planned_list(self):
+ for name, provider, prov in self.scenarios():
+ with self.subTest(name):
+ _path, _context, reader = self.render(provider=provider, prov=prov)
+ self.assertEqual(outline_entries(reader.outline), OUTLINE)
+
+ def test_the_outline_does_not_depend_on_the_data(self):
+ for rows in (MULTI_TIB, FREE_TIER, WITH_UNKNOWN, []):
+ with self.subTest(rows=len(rows)):
+ _path, _context, reader = self.render(rows)
+ self.assertEqual(outline_entries(reader.outline), OUTLINE)
+
+ def test_the_last_page_is_the_legal_notices_and_the_document_is_short(self):
+ for name, provider, prov in self.scenarios():
+ with self.subTest(name):
+ _path, _context, reader = self.render(provider=provider, prov=prov)
+ pages = page_texts(reader)
+
+ self.assertIn("Legal Notices", pages[-1][:200])
+ self.assertNotIn("Legal Notices", " ".join(pages[:-1]))
+ self.assertTrue(5 <= len(pages) <= 7, len(pages))
+
+ def test_chapters_start_pages(self):
+ _path, _context, reader = self.render()
+ pages = page_texts(reader)
+
+ self.assertIn("Introduction", pages[2])
+ self.assertIn("Data Landscape", pages[3])
+ starts = {
+ entry.title: reader.get_destination_page_number(entry)
+ for entry in reader.outline
+ if not isinstance(entry, list)
+ }
+ self.assertEqual(starts["Introduction"], 2)
+ self.assertEqual(starts["Data Landscape"], 3)
+ self.assertGreater(starts["Appendix – Egress Prices"], starts["Data Landscape"])
+
+ def test_every_running_header_and_footer_names_the_document(self):
+ _path, _context, reader = self.render()
+ pages = page_texts(reader)
+
+ for index, text in enumerate(pages[1:], start=2):
+ self.assertIn("Smoke Test Assessment", text)
+ self.assertIn(EDL_ID, text)
+ self.assertIn("COMMUNITY EDITION", text)
+ self.assertIn(f"Page {index} of {len(pages)}", text)
+
+ def test_the_fonts_are_plex_and_nothing_else(self):
+ path, _context, _reader = self.render()
+ names = set(re.findall(rb"/BaseFont\s*/(\S+)", Path(path).read_bytes()))
+
+ self.assertTrue(names)
+ self.assertTrue(all(b"IBMPlex" in name for name in names), names)
+
+
+class ContentTests(EgressReportTestCase):
+ def text(self, rows, **kwargs):
+ _path, _context, reader = self.render(rows, **kwargs)
+ return " ".join(page_texts(reader))
+
+ def test_the_scope_of_the_estimation_lists_the_providers_types(self):
+ azure = self.text(MULTI_TIB, provider=1)
+ aws = self.text(MULTI_TIB, provider=2)
+
+ self.assertIn("Recovery Services Vaults", azure)
+ self.assertNotIn("S3 Buckets", azure)
+ self.assertIn("S3 Buckets", aws)
+ self.assertIn("Aurora is not sized", aws)
+ self.assertNotIn("Recovery Services Vaults", aws)
+ self.assertIn("Prod-RG", azure)
+
+ def test_a_multi_tib_footprint_shows_the_cost_table_and_the_total(self):
+ text = self.text(MULTI_TIB)
+
+ # (2560 - 100) GiB at 0.087 per GiB
+ self.assertIn("$214.02", text)
+ self.assertIn("2,560.0 GiB", text)
+ self.assertEqual(text.count("Share of Volume"), 2)
+ self.assertIn("Storage Account", text)
+ self.assertIn("BILLABLE VOLUME", text)
+ self.assertNotIn("TiB", text)
+
+ def test_a_free_tier_footprint_says_so_and_leaves_out_the_cost_table(self):
+ text = self.text(FREE_TIER)
+
+ self.assertIn("20.0 GiB of the 100 GiB monthly free tier", text)
+ self.assertIn("$0.00", text)
+ # Only the landscape table has the share column
+ self.assertEqual(text.count("Share of Volume"), 1)
+ self.assertNotIn("BILLABLE VOLUME", text)
+ self.assertIn("FREE ALLOWANCE / MONTH", text)
+
+ def test_money_has_one_format_everywhere(self):
+ for rows in (MULTI_TIB, FREE_TIER, WITH_UNKNOWN):
+ with self.subTest(rows=len(rows)):
+ text = self.text(rows)
+ amounts = re.findall(r"\$[\d,]+(?:\.\d+)?", text)
+
+ self.assertTrue(amounts)
+ for amount in amounts:
+ self.assertRegex(amount, r"^\$[\d,]+\.\d\d$")
+
+ def test_unsized_types_are_listed_and_the_totals_carry_a_plus(self):
+ text = self.text(MULTI_TIB)
+ self.assertIn("Not sized: Recovery Services Vault (1 resource).", text)
+ self.assertIn("1 resource could not be sized", text)
+
+ text = self.text(WITH_UNKNOWN)
+ self.assertIn("Not sized: SQL Database (1 resource).", text)
+ # (2048 - 100) GiB * 0.087, and the + marks the lower bound
+ self.assertIn("2,048.0 GiB+", text)
+ self.assertIn("$169.48+", text)
+
+ def test_an_empty_inventory_renders_the_notice_and_no_tables(self):
+ text = self.text([])
+
+ self.assertIn("No data-bearing resources were identified", text)
+ self.assertIn("Estimated Costs", text)
+ self.assertEqual(text.count("Share of Volume"), 0)
+
+ def test_the_appendix_names_the_prices_the_sources_and_the_units(self):
+ text = self.text(MULTI_TIB)
+
+ for expected in (
+ "Appendix – Egress Prices",
+ "first 100 GiB / month",
+ "0.087 / GiB",
+ "Prices per GiB as published by Microsoft Azure",
+ "https://azure.microsoft.com/pricing/details/bandwidth/",
+ "https://azure.microsoft.com/pricing/details/storage/blobs/",
+ "Glossary",
+ "Switching charges",
+ "Free tier",
+ ):
+ self.assertIn(expected, text)
+
+ def test_next_steps_name_the_companion_and_the_platforms(self):
+ text = self.text(MULTI_TIB, prov=provenance())
+ self.assertIn(f"Executive Summary of the same run ({ES_ID})", text)
+ self.assertIn("Connected (https://exitcloud.io)", text)
+ self.assertIn("Enterprise (https://escapecloud.io/book-demo/)", text)
+
+ text = self.text(MULTI_TIB, prov=uploaded())
+ self.assertIn(f"Open the reports: {URL}", text)
+ self.assertNotIn("Connected (https://exitcloud.io)", text)
+
+ def test_no_date_is_written_month_first(self):
+ for name, provider, prov in self.scenarios():
+ with self.subTest(name):
+ _path, _context, reader = self.render(provider=provider, prov=prov)
+ for text in page_texts(reader):
+ self.assertIsNone(re.search(r"\b\d{2}-\d{2}-\d{4}\b", text))
+
+ def test_no_credential_is_shown(self):
+ _path, _context, reader = self.render(provider=1)
+ text = " ".join(page_texts(reader) + metadata_texts(reader))
+
+ self.assertNotIn("very-secret-client-value", text)
+ self.assertNotIn("Client", text)
+
+ def test_the_figures_do_not_move_with_the_presentation(self):
+ # The same rows give the same fee as the data functions
+ folder, json_path = self.prepare(MULTI_TIB, 1)
+ from cloudexit.report.egress import load_egress_estimate
+
+ with patch(
+ "cloudexit.report.egress.load_data", side_effect=pricing_for_both_clouds
+ ):
+ estimate = load_egress_estimate(str(folder), json_path)
+
+ self.assertAlmostEqual(estimate["total_fee"], 2460 * 0.087)
+ self.assertEqual(estimate["totals"]["known_size_bytes"], 2560 * GIB)
+
+
+class NoticesTests(EgressReportTestCase):
+ MARKERS = {1: "Microsoft group", 2: "Amazon.com"}
+
+ def test_the_notices_carry_the_egress_paragraph_and_the_providers_icons(self):
+ for name, provider, prov in self.scenarios():
+ with self.subTest(name):
+ _path, _context, reader = self.render(provider=provider, prov=prov)
+ pages = page_texts(reader)
+ other = 1 if provider == 2 else 2
+
+ self.assertIn(BLOCKS["notices_egress"], pages[-1])
+ self.assertIn("12 January 2027", pages[-1])
+ self.assertIn(self.MARKERS[provider], pages[-1])
+ self.assertNotIn(self.MARKERS[other], " ".join(pages))
+ self.assertIn(BLOCKS["notices_disclaimer"], pages[-1])
+
+ def test_the_egress_paragraph_comes_before_the_trademarks(self):
+ _path, _context, reader = self.render()
+ last = page_texts(reader)[-1]
+
+ self.assertLess(last.index(BLOCKS["notices_egress"]), last.index("Trademarks"))
+
+ def test_the_executive_summary_notices_do_not_carry_it(self):
+ from cloudexit.report.notices import notices_html
+
+ plain = notices_html(1, COLLECTED, BLOCKS)
+ egress = notices_html(1, COLLECTED, BLOCKS, egress=True)
+
+ self.assertNotIn([BLOCKS["notices_egress"]], plain["blocks"][0]["paragraphs"])
+ self.assertIn([BLOCKS["notices_egress"]], egress["blocks"][0]["paragraphs"])
+ self.assertEqual(
+ egress["blocks"][1:], plain["blocks"][1:], "only the disclaimer grows"
+ )
+
+ def test_the_icon_links_follow_the_provider(self):
+ for provider, own, other in (
+ (1, AZURE_ICONS_URL, AWS_ICONS_URL),
+ (2, AWS_ICONS_URL, AZURE_ICONS_URL),
+ ):
+ with self.subTest(provider=provider):
+ path, _context, _reader = self.render(provider=provider)
+ from tests.report.test_basic_report import link_targets
+
+ links = link_targets(PdfReader(path))
+ self.assertIn(own, links)
+ self.assertNotIn(other, links)
+
+
+class GuardTests(EgressReportTestCase):
+ def test_the_term_wording_and_retired_term_guards_pass_on_every_page(self):
+ for name, provider, prov in self.scenarios():
+ for rows in (MULTI_TIB, FREE_TIER, []):
+ with self.subTest(name, rows=len(rows)):
+ _path, _context, reader = self.render(rows, provider, prov)
+ for text in metadata_texts(reader):
+ self.assertEqual(find_terms(text), [])
+ for text in page_texts(reader):
+ self.assertEqual(find_terms(text), [])
+ self.assertEqual(retired_term_hits(text), [])
+ self.assertEqual(american_hits(text), [])
+
+ def test_the_html_follows_the_pdf_and_passes_the_guards(self):
+ for provider, details in (
+ (1, {"resourceGroupName": "Prod-RG", "tenantId": "t-1"}),
+ (2, {"region": "eu-central-1", "accountId": "123456789012"}),
+ ):
+ with self.subTest(provider=provider):
+ html, text = self.render_html(MULTI_TIB, provider, details)
+
+ position = 0
+ for heading in (
+ "Introduction",
+ "Purpose",
+ "Scope of the Estimation",
+ "Data Landscape",
+ "Estimated Costs",
+ "Appendix – Egress Prices",
+ "Glossary",
+ "Next Steps",
+ "Legal Notices",
+ ):
+ position = text.index(heading, position)
+ self.assertIn(EDL_ID, text)
+ self.assertIn(f"({ES_ID})", text)
+ self.assertTrue(
+ text.rstrip().endswith("© 2024–2026 EscapeCloud"), text[-80:]
+ )
+ self.assertEqual(find_terms(text), [])
+ self.assertEqual(find_terms(html.split("