from __future__ import annotations

import io
import json
import os
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import BinaryIO, Mapping, Optional

from arr_ingestion.contracts import IngestionError
from arr_ingestion.validation import DeliveryValidator
from arr_storage import (
    CloudClientError,
    CloudObjectBackend,
    FilesystemObjectBackend,
    InMemoryReadGrantBroker,
    ManagedObjectStore,
    ObjectAddress,
    ObjectKeyPolicy,
)
from arr_storage.contracts import BackendError, BackendObject
from tests.test_arr_ingestion_validation import build_delivery, policy
from tests.test_arr_opera_daily_ingest import reservation, xml_document


class FakeCloudClient:
    def __init__(self) -> None:
        self.objects: dict[str, tuple[bytes, str, dict[str, str]]] = {}
        self.calls: list[tuple[str, str]] = []
        self.fail_next: Optional[str] = None

    def _fail(self) -> None:
        if self.fail_next is not None:
            kind = self.fail_next
            self.fail_next = None
            raise CloudClientError(kind)

    def upload_file_if_absent(
        self,
        object_key: str,
        source: str,
        mime_type: str,
        metadata: Mapping[str, str],
    ) -> BackendObject:
        self._fail()
        self.calls.append(("upload", object_key))
        if object_key in self.objects:
            raise CloudClientError("conflict")
        value = Path(source).read_bytes()
        self.objects[object_key] = (value, mime_type, dict(metadata))
        return self.stat_object(object_key)

    def stat_object(self, object_key: str) -> BackendObject:
        self._fail()
        self.calls.append(("head", object_key))
        try:
            value, mime_type, metadata = self.objects[object_key]
        except KeyError:
            raise CloudClientError("not_found") from None
        normalized = dict(metadata)
        normalized["arr-backend-content-type"] = mime_type
        return BackendObject(object_key, len(value), normalized, etag="opaque-etag")

    def stream_object(self, object_key: str) -> BinaryIO:
        self._fail()
        self.calls.append(("get", object_key))
        try:
            value = self.objects[object_key][0]
        except KeyError:
            raise CloudClientError("not_found") from None
        return io.BytesIO(value)

    def copy_object_if_absent(
        self,
        source_key: str,
        destination_key: str,
        metadata: Mapping[str, str],
    ) -> BackendObject:
        self._fail()
        self.calls.append(("copy", destination_key))
        if destination_key in self.objects:
            raise CloudClientError("conflict")
        try:
            value, mime_type, _source_metadata = self.objects[source_key]
        except KeyError:
            raise CloudClientError("not_found") from None
        self.objects[destination_key] = (value, mime_type, dict(metadata))
        return self.stat_object(destination_key)

    def delete_exact_object(self, object_key: str) -> None:
        self._fail()
        self.calls.append(("delete", object_key))
        self.objects.pop(object_key, None)


class CopyFailingBackend:
    def __init__(self, delegate: FilesystemObjectBackend) -> None:
        self.delegate = delegate

    def put_file(self, *args, **kwargs):
        return self.delegate.put_file(*args, **kwargs)

    def head(self, *args, **kwargs):
        return self.delegate.head(*args, **kwargs)

    def open_reader(self, *args, **kwargs):
        return self.delegate.open_reader(*args, **kwargs)

    def copy_object(self, *args, **kwargs):
        raise BackendError("unavailable")

    def delete_object(self, *args, **kwargs):
        return self.delegate.delete_object(*args, **kwargs)


class ObjectKeyPolicyTests(unittest.TestCase):
    def test_exact_key_shape_round_trips(self):
        policy = ObjectKeyPolicy("arr/test")
        address = ObjectAddress("job:opaque-1", 12, "committed", "source_xml")
        key = policy.build(address)
        self.assertEqual(
            key,
            "arr/test/jobs/job:opaque-1/attempts/0012/committed/source_xml/source.xml",
        )
        self.assertEqual(policy.parse(key), address)

    def test_malformed_or_noncanonical_keys_are_rejected(self):
        policy = ObjectKeyPolicy()
        invalid = (
            "../arr/jobs/job-1/attempts/0001/committed/source_xml/source.xml",
            "/arr/jobs/job-1/attempts/0001/committed/source_xml/source.xml",
            "arr/jobs/job-1/attempts/1/committed/source_xml/source.xml",
            "arr/jobs/job-1/attempts/0001/committed/source_xml/private.xml",
            "arr/jobs/job-1/attempts/0001/public/source_xml/source.xml",
            "arr/jobs/job-1/attempts/0001/committed/source_xml/source.xml/extra",
        )
        for key in invalid:
            with self.subTest(key=key), self.assertRaises(IngestionError):
                policy.parse(key)

    def test_invalid_prefix_is_rejected(self):
        for prefix in ("ARR", "arr/private!", "arr//test", "../arr"):
            with self.subTest(prefix=prefix), self.assertRaises(ValueError):
                ObjectKeyPolicy(prefix)

    def test_booking_source_uses_a_canonical_private_xlsx_key(self):
        policy = ObjectKeyPolicy("arr/test")
        address = ObjectAddress("booking-job-1", 1, "committed", "booking_source")

        key = policy.build(address)

        self.assertEqual(
            key,
            "arr/test/jobs/booking-job-1/attempts/0001/committed/booking_source/booking-source.xlsx",
        )
        self.assertEqual(policy.parse(key), address)


class ManagedObjectStoreTests(unittest.TestCase):
    def setUp(self):
        self.temporary = tempfile.TemporaryDirectory()
        self.root = Path(self.temporary.name)
        self.backend = FilesystemObjectBackend(self.root / "objects", create=True)
        self.store = ManagedObjectStore(self.backend)
        self.source = self.root / "upload.xml"
        self.source.write_bytes(b"<RES_DETAIL><LIST_G_RESERVATION/></RES_DETAIL>")

    def tearDown(self):
        self.temporary.cleanup()

    def stage_source(self):
        return self.store.stage_file(
            job_id="job-001",
            attempt_no=1,
            role="source_xml",
            source=self.source,
            original_filename="source.xml",
        )

    def test_stage_commit_materialize_and_artifact_ref(self):
        staged = self.stage_source()
        self.assertEqual(staged.state, "staged")
        self.assertNotIn("upload.xml", staged.object_key)
        with self.assertRaisesRegex(IngestionError, "only committed") as raised:
            self.store.materialize(staged.object_key, self.root / "no.xml", 1024)
        self.assertEqual(raised.exception.code, "ARTIFACT_NOT_COMMITTED")

        committed = self.store.commit(staged)
        self.assertEqual(committed.state, "committed")
        self.assertFalse(self.backend.root.joinpath(*staged.object_key.split("/")).exists())
        destination = self.root / "download" / "source.xml"
        self.store.materialize(committed.object_key, destination, 1024)
        self.assertEqual(destination.read_bytes(), self.source.read_bytes())
        self.assertEqual(destination.stat().st_mode & 0o777, 0o600)
        reference = committed.to_artifact_ref()
        self.assertEqual(reference.role, "source_xml")
        self.assertEqual(reference.object_key, committed.object_key)
        self.assertEqual(reference.sha256, committed.sha256)

    def test_same_bytes_are_idempotent_but_different_bytes_conflict(self):
        staged = self.stage_source()
        again = self.stage_source()
        self.assertEqual(again, staged)
        committed = self.store.commit(staged)
        self.assertEqual(self.store.commit(staged), committed)

        second_source = self.root / "changed.xml"
        second_source.write_bytes(b"<changed/>")
        changed_stage = self.store.stage_file(
            job_id="job-001",
            attempt_no=1,
            role="source_xml",
            source=second_source,
            original_filename="source.xml",
        )
        with self.assertRaisesRegex(IngestionError, "identity conflicts") as raised:
            self.store.commit(changed_stage)
        self.assertEqual(raised.exception.code, "ARTIFACT_CONFLICT")

    def test_expected_hash_and_size_are_enforced_before_upload(self):
        with self.assertRaises(IngestionError) as hash_error:
            self.store.stage_file(
                job_id="job-hash",
                attempt_no=1,
                role="source_xml",
                source=self.source,
                original_filename="source.xml",
                expected_sha256="0" * 64,
            )
        self.assertEqual(hash_error.exception.code, "ARTIFACT_HASH_MISMATCH")
        self.assertFalse(any((self.root / "objects").rglob("source.xml")))

        with self.assertRaises(IngestionError) as size_error:
            self.store.stage_file(
                job_id="job-size",
                attempt_no=1,
                role="source_xml",
                source=self.source,
                original_filename="source.xml",
                expected_byte_size=1,
            )
        self.assertEqual(size_error.exception.code, "ARTIFACT_HASH_MISMATCH")

    def test_tampered_committed_bytes_are_detected_and_destination_removed(self):
        committed = self.store.commit(self.stage_source())
        object_path = self.backend.root.joinpath(*committed.object_key.split("/"))
        original = object_path.read_bytes()
        object_path.write_bytes(b"X" + original[1:])
        destination = self.root / "tampered.xml"
        with self.assertRaises(IngestionError) as raised:
            self.store.materialize(committed.object_key, destination, 1024)
        self.assertEqual(raised.exception.code, "ARTIFACT_HASH_MISMATCH")
        self.assertFalse(destination.exists())

    def test_existing_materialization_destination_is_never_overwritten_or_deleted(self):
        committed = self.store.commit(self.stage_source())
        destination = self.root / "existing.xml"
        destination.write_bytes(b"keep-me")
        with self.assertRaises(IngestionError) as raised:
            self.store.materialize(committed.object_key, destination, 1024)
        self.assertEqual(raised.exception.code, "ARTIFACT_CONFLICT")
        self.assertEqual(destination.read_bytes(), b"keep-me")

    def test_symlinked_backend_object_is_not_followed(self):
        committed = self.store.commit(self.stage_source())
        object_path = self.backend.root.joinpath(*committed.object_key.split("/"))
        outside = self.root / "outside.xml"
        outside.write_bytes(object_path.read_bytes())
        object_path.unlink()
        try:
            object_path.symlink_to(outside)
        except OSError:
            self.skipTest("symlinks unavailable")
        with self.assertRaises(IngestionError) as raised:
            self.store.materialize(committed.object_key, self.root / "followed.xml", 1024)
        self.assertEqual(raised.exception.code, "OBJECT_STORE_UNAVAILABLE")
        self.assertFalse((self.root / "followed.xml").exists())

    def test_tampered_metadata_and_too_small_download_limit_are_rejected(self):
        committed = self.store.commit(self.stage_source())
        metadata_path = self.backend.root.joinpath(*committed.object_key.split("/"))
        metadata_path = metadata_path.with_name(metadata_path.name + ".arr-metadata.json")
        payload = json.loads(metadata_path.read_text(encoding="utf-8"))
        payload["metadata"]["arr-role"] = "result_json"
        metadata_path.write_text(json.dumps(payload), encoding="utf-8")
        with self.assertRaises(IngestionError) as metadata_error:
            self.store.materialize(committed.object_key, self.root / "bad.xml", 1024)
        self.assertEqual(metadata_error.exception.code, "ARTIFACT_METADATA_INVALID")

        payload["metadata"]["arr-role"] = "source_xml"
        metadata_path.write_text(json.dumps(payload), encoding="utf-8")
        with self.assertRaises(IngestionError) as size_error:
            self.store.materialize(committed.object_key, self.root / "small.xml", 1)
        self.assertEqual(size_error.exception.code, "ARTIFACT_TOO_LARGE")

    def test_symlink_source_and_invalid_delivery_filename_are_rejected(self):
        symlink = self.root / "source-link.xml"
        try:
            symlink.symlink_to(self.source)
        except OSError:
            self.skipTest("symlinks unavailable")
        with self.assertRaises(IngestionError) as symlink_error:
            self.store.stage_file(
                job_id="job-link",
                attempt_no=1,
                role="source_xml",
                source=symlink,
                original_filename="source.xml",
            )
        self.assertEqual(symlink_error.exception.code, "ARTIFACT_REFERENCE_INVALID")
        with self.assertRaises(IngestionError):
            self.store.stage_file(
                job_id="job-name",
                attempt_no=1,
                role="source_xml",
                source=self.source,
                original_filename="../guest.xml",
            )

    def test_copy_failure_preserves_staged_object_and_creates_no_commit(self):
        failing_store = ManagedObjectStore(CopyFailingBackend(self.backend))
        staged = failing_store.stage_file(
            job_id="job-copy-fail",
            attempt_no=1,
            role="source_xml",
            source=self.source,
            original_filename="source.xml",
        )
        with self.assertRaises(IngestionError) as raised:
            failing_store.commit(staged)
        self.assertEqual(raised.exception.code, "OBJECT_STORE_UNAVAILABLE")
        self.assertTrue(self.backend.root.joinpath(*staged.object_key.split("/")).is_file())
        committed_key = staged.object_key.replace("/staged/", "/committed/")
        self.assertFalse(self.backend.root.joinpath(*committed_key.split("/")).exists())


class ReadGrantTests(unittest.TestCase):
    def test_grant_is_opaque_job_bound_short_lived_and_single_read(self):
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            source = root / "source.xml"
            source.write_bytes(b"<synthetic/>")
            store = ManagedObjectStore(
                FilesystemObjectBackend(root / "objects", create=True)
            )
            committed = store.upload_committed(
                job_id="job-grant",
                attempt_no=1,
                role="source_xml",
                source=source,
                original_filename="source.xml",
            )
            now = datetime(2026, 7, 28, tzinfo=timezone.utc)
            broker = InMemoryReadGrantBroker(store, clock=lambda: now)
            grant = broker.issue(
                job_id="job-grant",
                object_key=committed.object_key,
                original_filename="source.xml",
                ttl_seconds=60,
            )
            self.assertFalse(hasattr(grant, "object_key"))
            self.assertNotIn(committed.object_key, repr(grant))
            self.assertEqual(grant.max_reads, 1)

            with self.assertRaises(IngestionError):
                broker.materialize(
                    handle=grant.handle,
                    job_id="wrong-job",
                    destination=root / "wrong.xml",
                    max_bytes=1024,
                )
            destination = root / "fetched.xml"
            broker.materialize(
                handle=grant.handle,
                job_id="job-grant",
                destination=destination,
                max_bytes=1024,
            )
            self.assertEqual(destination.read_bytes(), source.read_bytes())
            with self.assertRaises(IngestionError) as replay:
                broker.materialize(
                    handle=grant.handle,
                    job_id="job-grant",
                    destination=root / "replay.xml",
                    max_bytes=1024,
                )
            self.assertEqual(replay.exception.code, "READ_GRANT_INVALID")

    def test_expired_grant_and_excessive_ttl_are_rejected(self):
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            source = root / "source.xml"
            source.write_bytes(b"<synthetic/>")
            store = ManagedObjectStore(
                FilesystemObjectBackend(root / "objects", create=True)
            )
            committed = store.upload_committed(
                job_id="job-expiry",
                attempt_no=1,
                role="source_xml",
                source=source,
                original_filename="source.xml",
            )
            current = [datetime(2026, 7, 28, tzinfo=timezone.utc)]
            broker = InMemoryReadGrantBroker(store, clock=lambda: current[0])
            with self.assertRaises(IngestionError):
                broker.issue(
                    job_id="job-expiry",
                    object_key=committed.object_key,
                    original_filename="source.xml",
                    ttl_seconds=301,
                )
            grant = broker.issue(
                job_id="job-expiry",
                object_key=committed.object_key,
                original_filename="source.xml",
                ttl_seconds=1,
            )
            current[0] += timedelta(seconds=1)
            with self.assertRaises(IngestionError) as expired:
                broker.materialize(
                    handle=grant.handle,
                    job_id="job-expiry",
                    destination=root / "expired.xml",
                    max_bytes=1024,
                )
            self.assertEqual(expired.exception.code, "READ_GRANT_INVALID")
            self.assertEqual(broker.purge_expired(), 1)


class CloudObjectBackendTests(unittest.TestCase):
    def test_fake_cloud_port_runs_full_guarded_workflow(self):
        client = FakeCloudClient()
        store = ManagedObjectStore(CloudObjectBackend(client))
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            source = root / "result.json"
            source.write_bytes(b'{"status":"synthetic"}\n')
            committed = store.upload_committed(
                job_id="job-cloud",
                attempt_no=2,
                role="result_json",
                source=source,
                original_filename="result.json",
            )
            destination = root / "download.json"
            store.materialize(committed.object_key, destination, 1024)
            self.assertEqual(destination.read_bytes(), source.read_bytes())
        operations = [operation for operation, _key in client.calls]
        self.assertIn("upload", operations)
        self.assertIn("copy", operations)
        self.assertIn("delete", operations)
        self.assertIn("get", operations)

    def test_cloud_error_is_mapped_to_safe_storage_error(self):
        client = FakeCloudClient()
        client.fail_next = "unavailable"
        store = ManagedObjectStore(CloudObjectBackend(client))
        with tempfile.TemporaryDirectory() as temporary:
            source = Path(temporary) / "source.xml"
            source.write_bytes(b"<synthetic/>")
            with self.assertRaises(IngestionError) as raised:
                store.stage_file(
                    job_id="job-cloud-fail",
                    attempt_no=1,
                    role="source_xml",
                    source=source,
                    original_filename="source.xml",
                )
        self.assertEqual(raised.exception.code, "OBJECT_STORE_UNAVAILABLE")
        self.assertNotIn("provider", raised.exception.safe_message)


class StorageDeliveryIntegrationTests(unittest.TestCase):
    def test_all_committed_outputs_feed_the_existing_arr_validator(self):
        synthetic_xml = xml_document(
            reservation(1, rate_code="NOT-ALLOWED", rate_amount="INVALID"),
            reservation(2, res_comment="SYN-STORAGE", rooms="1"),
        )
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            raw, _memory_store, values = build_delivery(synthetic_xml, root / "processor")
            envelope = json.loads(raw)
            store = ManagedObjectStore(
                FilesystemObjectBackend(root / "objects", create=True)
            )
            for role, value in values.items():
                upload = root / (role + Path(envelope["artifacts"][role]["original_filename"]).suffix)
                upload.write_bytes(value)
                declared = envelope["artifacts"][role]
                committed = store.upload_committed(
                    job_id=envelope["job_id"],
                    attempt_no=envelope["attempt_no"],
                    role=role,
                    source=upload,
                    original_filename=declared["original_filename"],
                    expected_sha256=declared["sha256"],
                    expected_byte_size=declared["byte_size"],
                )
                declared["object_key"] = committed.object_key

            verified = DeliveryValidator(store, policy()).validate(
                json.dumps(envelope, ensure_ascii=False).encode("utf-8")
            )
            self.assertEqual(verified.envelope.status, "success")
            self.assertEqual(verified.structured_payload["output_rows"], 1)


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