from __future__ import annotations

import base64
from io import BytesIO
import unittest

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from openpyxl import Workbook

from govoplan_addresses.backend.db.models import AddressBook, Contact
from govoplan_addresses.backend.import_schemas import (
    AddressImportConfiguration,
    AddressImportPreviewRequest,
    AddressImportProfileCreateRequest,
    AddressImportProfileUpdateRequest,
    AddressImportRollbackRequest,
)
from govoplan_addresses.backend.imports import (
    apply_address_import,
    create_import_profile,
    get_import_run,
    import_run_payload,
    preview_address_import,
    rollback_address_import,
    update_import_profile,
)
from govoplan_core.db.base import Base


class Principal:
    account_id = "account-1"
    group_ids = frozenset({"group-1"})

    @property
    def tenant_id(self) -> str:
        return "tenant-1"

    def has(self, scope: str) -> bool:
        return scope in {
            "addresses:address_book:read",
            "addresses:address_book:write",
            "addresses:contact:read",
            "addresses:contact:write",
        }


class OtherTenantPrincipal(Principal):
    @property
    def tenant_id(self) -> str:
        return "tenant-2"


def encoded(value: str) -> str:
    return base64.b64encode(value.encode()).decode()


class AddressTabularImportTests(unittest.TestCase):
    def setUp(self) -> None:
        engine = create_engine("sqlite+pysqlite:///:memory:")
        Base.metadata.create_all(engine)
        self.session = sessionmaker(bind=engine, expire_on_commit=False)()
        self.principal = Principal()
        self.book = AddressBook(
            tenant_id="tenant-1",
            scope_type="tenant",
            scope_id="tenant-1",
            name="Imported contacts",
            source_kind="local",
            read_only=False,
        )
        self.session.add(self.book)
        self.profile = create_import_profile(
            self.session,
            self.principal,
            AddressImportProfileCreateRequest(
                scope_type="tenant",
                name="Monthly contacts",
                source_format="csv",
                configuration=AddressImportConfiguration(
                    delimiter=";",
                    field_mappings={
                        "source_key": "id",
                        "given_name": "first",
                        "family_name": "last",
                        "email": "email",
                        "organization": "organization",
                    },
                ),
            ),
        )
        self.session.flush()

    def test_preview_apply_repeat_and_guarded_rollback(self) -> None:
        payload = AddressImportPreviewRequest(
            profile_id=self.profile.id,
            filename="contacts.csv",
            content_base64=encoded(
                "id;first;last;email;organization\n"
                "1;Ada;Lovelace;ada@example.test;Analysis Office\n"
                "2;Grace;Hopper;grace@example.test;Computing Office\n"
            ),
        )
        run = preview_address_import(self.session, self.principal, self.book.id, payload)
        self.assertEqual(2, run.statistics["create"])
        self.assertFalse(run.diagnostics)

        applied = apply_address_import(
            self.session,
            self.principal,
            run.id,
            expected_plan_hash=run.plan_hash,
        )
        self.assertEqual("applied", applied.status)
        self.assertEqual(2, self.session.query(Contact).filter(Contact.deleted_at.is_(None)).count())
        self.assertIs(applied, apply_address_import(self.session, self.principal, run.id, expected_plan_hash=run.plan_hash))
        response_evidence = import_run_payload(applied)["result_evidence"]
        self.assertEqual(2, response_evidence["created_contact_count"])
        self.assertNotIn("created_contact_ids", response_evidence)
        self.assertNotIn("updated_contacts", response_evidence)

        repeated = preview_address_import(self.session, self.principal, self.book.id, payload)
        self.assertEqual(2, repeated.statistics["unchanged"])
        apply_address_import(self.session, self.principal, repeated.id, expected_plan_hash=repeated.plan_hash)
        self.assertEqual(2, self.session.query(Contact).filter(Contact.deleted_at.is_(None)).count())

        rolled_back = rollback_address_import(
            self.session,
            self.principal,
            run.id,
            AddressImportRollbackRequest(
                expected_plan_hash=run.plan_hash,
                reason="The operator selected the wrong monthly file.",
            ),
        )
        self.assertEqual("rolled_back", rolled_back.status)
        self.assertEqual(0, self.session.query(Contact).filter(Contact.deleted_at.is_(None)).count())

    def test_rollback_rejects_a_stale_review_hash(self) -> None:
        run = preview_address_import(
            self.session,
            self.principal,
            self.book.id,
            AddressImportPreviewRequest(
                profile_id=self.profile.id,
                filename="contacts.csv",
                content_base64=encoded(
                    "id;first;last;email;organization\n"
                    "1;Ada;Lovelace;ada@example.test;Analysis Office\n"
                ),
            ),
        )
        apply_address_import(
            self.session,
            self.principal,
            run.id,
            expected_plan_hash=run.plan_hash,
        )

        with self.assertRaisesRegex(ValueError, "reviewed import plan changed"):
            rollback_address_import(
                self.session,
                self.principal,
                run.id,
                AddressImportRollbackRequest(
                    expected_plan_hash="0" * 64,
                    reason="The operator selected the wrong monthly file.",
                ),
            )

    def test_persisted_run_read_is_tenant_bounded_and_source_safe(self) -> None:
        source = "id;first;last;email;organization\n1;Ada;Lovelace;ada@example.test;Analysis Office\n"
        run = preview_address_import(
            self.session,
            self.principal,
            self.book.id,
            AddressImportPreviewRequest(
                profile_id=self.profile.id,
                filename="contacts.csv",
                content_base64=encoded(source),
            ),
        )
        self.session.flush()

        payload = import_run_payload(get_import_run(self.session, self.principal, run.id))
        self.assertEqual("previewed", payload["status"])
        self.assertEqual(run.plan_hash, payload["plan_hash"])
        self.assertNotIn("plan_data", payload)
        self.assertNotIn(source, repr(payload))

        with self.assertRaisesRegex(ValueError, "not found"):
            get_import_run(self.session, OtherTenantPrincipal(), run.id)

    def test_duplicate_keys_and_changed_targets_block_apply(self) -> None:
        duplicate = preview_address_import(
            self.session,
            self.principal,
            self.book.id,
            AddressImportPreviewRequest(
                profile_id=self.profile.id,
                filename="duplicates.csv",
                content_base64=encoded(
                    "id;first;last;email;organization\n"
                    "1;Ada;Lovelace;ada@example.test;One\n"
                    "1;Ada;Lovelace;ada@example.test;Two\n"
                ),
            ),
        )
        self.assertEqual(2, duplicate.statistics["conflict"])
        with self.assertRaisesRegex(ValueError, "error diagnostics"):
            apply_address_import(self.session, self.principal, duplicate.id, expected_plan_hash=duplicate.plan_hash)

        initial = preview_address_import(
            self.session,
            self.principal,
            self.book.id,
            AddressImportPreviewRequest(
                profile_id=self.profile.id,
                filename="one.csv",
                content_base64=encoded("id;first;last;email;organization\n1;Ada;Lovelace;ada@example.test;One\n"),
            ),
        )
        apply_address_import(self.session, self.principal, initial.id, expected_plan_hash=initial.plan_hash)
        changed = preview_address_import(
            self.session,
            self.principal,
            self.book.id,
            AddressImportPreviewRequest(
                profile_id=self.profile.id,
                filename="one.csv",
                content_base64=encoded("id;first;last;email;organization\n1;Ada;Lovelace;ada@example.test;Two\n"),
            ),
        )
        contact = self.session.query(Contact).filter(Contact.deleted_at.is_(None)).one()
        contact.organization = "Concurrent edit"
        self.session.flush()
        with self.assertRaisesRegex(ValueError, "changed after preview"):
            apply_address_import(self.session, self.principal, changed.id, expected_plan_hash=changed.plan_hash)

    def test_profile_updates_create_immutable_versions(self) -> None:
        next_profile = update_import_profile(
            self.session,
            self.principal,
            self.profile.id,
            payload=AddressImportProfileUpdateRequest(name="Monthly contacts v2"),
        )
        self.assertFalse(self.profile.is_current)
        self.assertTrue(next_profile.is_current)
        self.assertEqual(self.profile.profile_key, next_profile.profile_key)
        self.assertEqual(2, next_profile.version)

    def test_xlsx_sheet_selection_and_formula_rejection(self) -> None:
        workbook = Workbook()
        workbook.active.title = "Ignore"
        sheet = workbook.create_sheet("Contacts")
        sheet.append(["id", "first", "last", "email", "organization"])
        sheet.append(["1", "Ada", "Lovelace", "ada@example.test", "Analysis Office"])
        content = BytesIO()
        workbook.save(content)
        xlsx_profile = create_import_profile(
            self.session,
            self.principal,
            AddressImportProfileCreateRequest(
                scope_type="tenant",
                name="Workbook contacts",
                source_format="xlsx",
                configuration=AddressImportConfiguration(
                    sheet_name="Contacts",
                    field_mappings={
                        "source_key": "id",
                        "given_name": "first",
                        "family_name": "last",
                        "email": "email",
                        "organization": "organization",
                    },
                ),
            ),
        )
        self.session.flush()
        run = preview_address_import(
            self.session,
            self.principal,
            self.book.id,
            AddressImportPreviewRequest(
                profile_id=xlsx_profile.id,
                filename="contacts.xlsx",
                content_base64=base64.b64encode(content.getvalue()).decode(),
            ),
        )
        self.assertEqual(1, run.statistics["create"])

        sheet["E2"] = "=CONCAT(\"Analysis\", \" Office\")"
        content = BytesIO()
        workbook.save(content)
        with self.assertRaisesRegex(ValueError, "formulas are never evaluated"):
            preview_address_import(
                self.session,
                self.principal,
                self.book.id,
                AddressImportPreviewRequest(
                    profile_id=xlsx_profile.id,
                    filename="contacts.xlsx",
                    content_base64=base64.b64encode(content.getvalue()).decode(),
                ),
            )


if __name__ == "__main__":
    unittest.main()