from __future__ import annotations from dataclasses import dataclass from datetime import date, datetime, timezone from decimal import Decimal from io import BytesIO from typing import Any, Iterable import pytest from services.export_serializers import ( CSVExportColumn, CSVExporter, ExcelExporter, default_formatter, default_project_columns, default_scenario_columns, export_projects_to_excel, export_scenarios_to_excel, format_date_iso, format_datetime_utc, format_decimal, stream_projects_to_csv, stream_scenarios_to_csv, ) from openpyxl import load_workbook @dataclass(slots=True) class DummyProject: name: str location: str | None = None operation_type: str = "open_pit" description: str | None = None created_at: datetime | None = None updated_at: datetime | None = None @dataclass(slots=True) class DummyScenario: project: DummyProject | None name: str status: str = "draft" start_date: date | None = None end_date: date | None = None discount_rate: Decimal | None = None currency: str | None = None primary_resource: str | None = None description: str | None = None created_at: datetime | None = None updated_at: datetime | None = None def collect_csv_bytes(chunks: Iterable[bytes]) -> list[str]: return [chunk.decode("utf-8") for chunk in chunks] def load_workbook_bytes(data: bytes): buffer = BytesIO(data) return load_workbook(buffer, read_only=True, data_only=True) def test_csv_exporter_writes_header_and_rows() -> None: exporter = CSVExporter( [ CSVExportColumn("Name", "name"), CSVExportColumn("Location", "location"), ] ) project = DummyProject(name="Alpha", location="Nevada") chunks = collect_csv_bytes(exporter.iter_bytes([project])) assert chunks[0] == "Name,Location\n" assert chunks[1] == "Alpha,Nevada\n" def test_excel_exporter_basic_workbook() -> None: exporter = ExcelExporter(default_project_columns(), sheet_name="Projects") project = DummyProject(name="Alpha", location="Nevada") data = exporter.export([project]) workbook = load_workbook_bytes(data) sheet = workbook["Projects"] rows = list(sheet.rows) assert [cell.value for cell in rows[0]] == [ "name", "location", "operation_type", "description", "created_at", "updated_at", ] assert rows[1][0].value == "Alpha" def test_excel_export_projects_helper_with_metadata() -> None: project = DummyProject(name="Alpha", location="Nevada") data = export_projects_to_excel( [project], metadata={"rows": 1}, workbook_title="Project Export") workbook = load_workbook_bytes(data) assert workbook.properties.title == "Project Export" assert "Projects" in workbook.sheetnames assert any(sheet.title.startswith("Metadata") for sheet in workbook.worksheets) def test_excel_export_scenarios_helper_projects_resolved() -> None: project = DummyProject(name="Alpha") scenario = DummyScenario(project=project, name="Scenario 1") data = export_scenarios_to_excel([scenario]) workbook = load_workbook_bytes(data) sheet = workbook["Scenarios"] rows = list(sheet.rows) assert rows[1][0].value == "Alpha" assert rows[1][1].value == "Scenario 1" exporter = CSVExporter( [ CSVExportColumn("Name", "name"), CSVExportColumn("Location", "location"), ] ) project = DummyProject(name="Alpha", location="Nevada") chunks = collect_csv_bytes(exporter.iter_bytes([project])) assert chunks[0] == "Name,Location\n" assert chunks[1] == "Alpha,Nevada\n" def test_csv_exporter_handles_optional_values_and_default_formatter() -> None: exporter = CSVExporter( [ CSVExportColumn("Name", "name"), CSVExportColumn("Description", "description"), ] ) project = DummyProject(name="Bravo") chunks = collect_csv_bytes(exporter.iter_bytes([project])) assert chunks[-1] == "Bravo,\n" def test_stream_projects_uses_default_columns() -> None: projects = [ DummyProject( name="Alpha", location="Nevada", operation_type="open_pit", description="Primary", created_at=datetime(2025, 1, 1, tzinfo=timezone.utc), updated_at=datetime(2025, 1, 2, tzinfo=timezone.utc), ) ] chunks = collect_csv_bytes(stream_projects_to_csv(projects)) assert chunks[0].startswith("name,location,operation_type") assert any("Alpha" in chunk for chunk in chunks) def test_stream_scenarios_resolves_project_name_accessor() -> None: project = DummyProject(name="Project X") scenario = DummyScenario(project=project, name="Scenario A") chunks = collect_csv_bytes(stream_scenarios_to_csv([scenario])) assert "Project X" in chunks[-1] assert "Scenario A" in chunks[-1] def test_custom_formatter_applies() -> None: def uppercase(value: Any) -> str: return str(value).upper() if value is not None else "" exporter = CSVExporter([ CSVExportColumn("Name", "name", formatter=uppercase), ]) chunks = collect_csv_bytes( exporter.iter_bytes([DummyProject(name="alpha")])) assert chunks[-1] == "ALPHA\n" def test_default_formatter_handles_multiple_types() -> None: assert default_formatter(None) == "" assert default_formatter(True) == "true" assert default_formatter(False) == "false" assert default_formatter(Decimal("1.234")) == "1.23" assert default_formatter( datetime(2025, 1, 1, tzinfo=timezone.utc)).endswith("Z") assert default_formatter(date(2025, 1, 1)) == "2025-01-01" def test_format_helpers() -> None: assert format_date_iso(date(2025, 5, 1)) == "2025-05-01" assert format_date_iso("not-a-date") == "" ts = datetime(2025, 5, 1, 12, 0, tzinfo=timezone.utc) assert format_datetime_utc(ts) == "2025-05-01T12:00:00Z" assert format_datetime_utc("nope") == "" assert format_decimal(None) == "" assert format_decimal(Decimal("12.345")) == "12.35" assert format_decimal(10) == "10.00" def test_default_project_columns_includes_required_fields() -> None: columns = default_project_columns() headers = [column.header for column in columns] assert headers[:3] == ["name", "location", "operation_type"] def test_default_scenario_columns_handles_missing_project() -> None: scenario = DummyScenario(project=None, name="Orphan Scenario") exporter = CSVExporter(default_scenario_columns()) chunks = collect_csv_bytes(exporter.iter_bytes([scenario])) assert chunks[-1].startswith(",Orphan Scenario") def test_csv_exporter_requires_columns() -> None: with pytest.raises(ValueError): CSVExporter([])